gerrit.onap Code Review - policy/apex-pdp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2020 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.apex.plugins.event.carrier.grpc;
  22
  23 import com.google.protobuf.InvalidProtocolBufferException;
  24 import com.google.protobuf.util.JsonFormat;
  25 import java.util.Properties;
  26 import java.util.concurrent.CountDownLatch;
  27 import java.util.concurrent.TimeUnit;
  28 import java.util.concurrent.atomic.AtomicReference;
  29 import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType;
  30 import org.onap.ccsdk.cds.controllerblueprints.common.api.Status;
  31 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
  32 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput.Builder;
  33 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
  34 import org.onap.policy.apex.service.engine.event.ApexEventConsumer;
  35 import org.onap.policy.apex.service.engine.event.ApexEventException;
  36 import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
  37 import org.onap.policy.apex.service.engine.event.ApexPluginsEventProducer;
  38 import org.onap.policy.apex.service.engine.event.PeeredReference;
  39 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
  40 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerPeeredMode;
  41 import org.onap.policy.cds.api.CdsProcessorListener;
  42 import org.onap.policy.cds.client.CdsProcessorGrpcClient;
  43 import org.onap.policy.cds.properties.CdsServerProperties;
  44 import org.onap.policy.controlloop.actor.cds.constants.CdsActorConstants;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 /**
  49  * Concrete implementation of an Apex gRPC plugin that manages to send a GRPC request.
  50  *
  51  * @author Ajith Sreekumar(ajith.sreekumar@est.tech)
  52  *
  53  */
  54 public class ApexGrpcProducer extends ApexPluginsEventProducer implements CdsProcessorListener {
  55     private static final Logger LOGGER = LoggerFactory.getLogger(ApexGrpcProducer.class);
  56
  57     private CdsServerProperties props;
  58     // The gRPC client
  59     private CdsProcessorGrpcClient client;
  60
  61     private AtomicReference<ExecutionServiceOutput> cdsResponse = new AtomicReference<>();
  62
  63     /**
  64      * {@inheritDoc}.
  65      */
  66     @Override
  67     public void init(final String producerName, final EventHandlerParameters producerParameters)
  68         throws ApexEventException {
  69         this.name = producerName;
  70
  71         // Check and get the gRPC Properties
  72         if (!(producerParameters.getCarrierTechnologyParameters() instanceof GrpcCarrierTechnologyParameters)) {
  73             final String errorMessage =
  74                 "Specified producer properties are not applicable to gRPC producer (" + this.name + ")";
  75             throw new ApexEventException(errorMessage);
  76         }
  77         GrpcCarrierTechnologyParameters grpcProducerProperties =
  78             (GrpcCarrierTechnologyParameters) producerParameters.getCarrierTechnologyParameters();
  79         grpcProducerProperties.validateGrpcParameters(true);
  80         client = makeGrpcClient(grpcProducerProperties);
  81     }
  82
  83     private CdsProcessorGrpcClient makeGrpcClient(GrpcCarrierTechnologyParameters grpcProducerProperties) {
  84         props = new CdsServerProperties();
  85         props.setHost(grpcProducerProperties.getHost());
  86         props.setPort(grpcProducerProperties.getPort());
  87         props.setUsername(grpcProducerProperties.getUsername());
  88         props.setPassword(grpcProducerProperties.getPassword());
  89         props.setTimeout(grpcProducerProperties.getTimeout());
  90
  91         return new CdsProcessorGrpcClient(this, props);
  92     }
  93
  94     /**
  95      * {@inheritDoc}.
  96      */
  97     @Override
  98     public void sendEvent(final long executionId, final Properties executionProperties, final String eventName,
  99         final Object event) {
 100
 101         ExecutionServiceInput executionServiceInput;
 102         Builder builder = ExecutionServiceInput.newBuilder();
 103         try {
 104             JsonFormat.parser().ignoringUnknownFields().merge((String) event, builder);
 105             executionServiceInput = builder.build();
 106         } catch (InvalidProtocolBufferException e) {
 107             throw new ApexEventRuntimeException(
 108                 "Incoming Event cannot be converted to ExecutionServiceInput type for gRPC request." + e.getMessage());
 109         }
 110         try {
 111             CountDownLatch countDownLatch = client.sendRequest(executionServiceInput);
 112             if (!countDownLatch.await(props.getTimeout(), TimeUnit.SECONDS)) {
 113                 cdsResponse.set(ExecutionServiceOutput.newBuilder().setStatus(Status.newBuilder()
 114                     .setErrorMessage(CdsActorConstants.TIMED_OUT).setEventType(EventType.EVENT_COMPONENT_FAILURE))
 115                     .build());
 116                 LOGGER.error("gRPC Request timed out.");
 117             }
 118         } catch (InterruptedException e) {
 119             LOGGER.error("gRPC request failed. {}", e.getMessage());
 120             cdsResponse.set(ExecutionServiceOutput.newBuilder().setStatus(Status.newBuilder()
 121                 .setErrorMessage(CdsActorConstants.INTERRUPTED).setEventType(EventType.EVENT_COMPONENT_FAILURE))
 122                 .build());
 123             Thread.currentThread().interrupt();
 124         }
 125
 126         if (!EventType.EVENT_COMPONENT_EXECUTED.equals(cdsResponse.get().getStatus().getEventType())) {
 127             String errorMessage = "Sending event \"" + eventName + "\" by " + this.name + " to CDS failed, "
 128                 + "response from CDS:\n" + cdsResponse.get();
 129             throw new ApexEventRuntimeException(errorMessage);
 130         }
 131
 132         consumeEvent(executionId, cdsResponse.get());
 133     }
 134
 135     private void consumeEvent(long executionId, ExecutionServiceOutput event) {
 136         // Find the peered consumer for this producer
 137         final PeeredReference peeredRequestorReference = peerReferenceMap.get(EventHandlerPeeredMode.REQUESTOR);
 138         if (peeredRequestorReference == null) {
 139             return;
 140         }
 141         // Find the gRPC Response Consumer that will take in the response to APEX Engine
 142         final ApexEventConsumer consumer = peeredRequestorReference.getPeeredConsumer();
 143         if (!(consumer instanceof ApexGrpcConsumer)) {
 144             final String errorMessage = "Recieve of gRPC response by APEX failed,"
 145                 + "The consumer is not an instance of ApexGrpcConsumer\n. The received gRPC response:" + event;
 146             throw new ApexEventRuntimeException(errorMessage);
 147         }
 148
 149         // Use the consumer to consume this response event in APEX
 150         final ApexGrpcConsumer grpcConsumer = (ApexGrpcConsumer) consumer;
 151         try {
 152             grpcConsumer.getEventReceiver().receiveEvent(executionId, new Properties(),
 153                 JsonFormat.printer().print(event));
 154         } catch (ApexEventException | InvalidProtocolBufferException e) {
 155             throw new ApexEventRuntimeException("Consuming gRPC response failed.", e);
 156         }
 157     }
 158
 159     /**
 160      * {@inheritDoc}.
 161      */
 162     @Override
 163     public void stop() {
 164         client.close();
 165     }
 166
 167     @Override
 168     public void onMessage(ExecutionServiceOutput message) {
 169         LOGGER.info("Received notification from CDS: {}", message);
 170         cdsResponse.set(message);
 171     }
 172
 173     @Override
 174     public void onError(Throwable throwable) {
 175         String errorMsg = throwable.getLocalizedMessage();
 176         cdsResponse.set(ExecutionServiceOutput.newBuilder()
 177             .setStatus(Status.newBuilder().setErrorMessage(errorMsg).setEventType(EventType.EVENT_COMPONENT_FAILURE))
 178             .build());
 179         LOGGER.error("Failed processing blueprint {} {}", errorMsg, throwable);
 180     }
 181 }