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.ApexEventException;
  35 import org.onap.policy.apex.service.engine.event.ApexEventRuntimeException;
  36 import org.onap.policy.apex.service.engine.event.ApexPluginsEventProducer;
  37 import org.onap.policy.apex.service.parameters.eventhandler.EventHandlerParameters;
  38 import org.onap.policy.cds.api.CdsProcessorListener;
  39 import org.onap.policy.cds.client.CdsProcessorGrpcClient;
  40 import org.onap.policy.cds.properties.CdsServerProperties;
  41 import org.onap.policy.controlloop.actor.cds.constants.CdsActorConstants;
  42 import org.slf4j.Logger;
  43 import org.slf4j.LoggerFactory;
  44
  45 /**
  46  * Concrete implementation of an Apex gRPC plugin that manages to send a GRPC request.
  47  *
  48  * @author Ajith Sreekumar(ajith.sreekumar@est.tech)
  49  *
  50  */
  51 public class ApexGrpcProducer extends ApexPluginsEventProducer implements CdsProcessorListener {
  52     private static final Logger LOGGER = LoggerFactory.getLogger(ApexGrpcProducer.class);
  53
  54     private CdsServerProperties props;
  55     // The gRPC client
  56     private CdsProcessorGrpcClient client;
  57
  58     private AtomicReference<ExecutionServiceOutput> cdsResponse = new AtomicReference<>();
  59
  60     /**
  61      * {@inheritDoc}.
  62      */
  63     @Override
  64     public void init(final String producerName, final EventHandlerParameters producerParameters)
  65         throws ApexEventException {
  66         this.name = producerName;
  67
  68         // Check and get the gRPC Properties
  69         if (!(producerParameters.getCarrierTechnologyParameters() instanceof GrpcCarrierTechnologyParameters)) {
  70             final String errorMessage =
  71                 "Specified producer properties are not applicable to gRPC producer (" + this.name + ")";
  72             throw new ApexEventException(errorMessage);
  73         }
  74         GrpcCarrierTechnologyParameters grpcProducerProperties =
  75             (GrpcCarrierTechnologyParameters) producerParameters.getCarrierTechnologyParameters();
  76
  77         client = makeGrpcClient(grpcProducerProperties);
  78     }
  79
  80     private CdsProcessorGrpcClient makeGrpcClient(GrpcCarrierTechnologyParameters grpcProducerProperties) {
  81         props = new CdsServerProperties();
  82         props.setHost(grpcProducerProperties.getHost());
  83         props.setPort(grpcProducerProperties.getPort());
  84         props.setUsername(grpcProducerProperties.getUsername());
  85         props.setPassword(grpcProducerProperties.getPassword());
  86         props.setTimeout(grpcProducerProperties.getTimeout());
  87
  88         return new CdsProcessorGrpcClient(this, props);
  89     }
  90
  91     /**
  92      * {@inheritDoc}.
  93      */
  94     @Override
  95     public void sendEvent(final long executionId, final Properties executionProperties, final String eventName,
  96         final Object event) {
  97
  98         ExecutionServiceInput executionServiceInput;
  99         Builder builder = ExecutionServiceInput.newBuilder();
 100         try {
 101             JsonFormat.parser().ignoringUnknownFields().merge((String) event, builder);
 102             executionServiceInput = builder.build();
 103         } catch (InvalidProtocolBufferException e) {
 104             throw new ApexEventRuntimeException(
 105                 "Incoming Event cannot be converted to ExecutionServiceInput type for gRPC request." + e.getMessage());
 106         }
 107         try {
 108             CountDownLatch countDownLatch = client.sendRequest(executionServiceInput);
 109             if (!countDownLatch.await(props.getTimeout(), TimeUnit.SECONDS)) {
 110                 cdsResponse.set(ExecutionServiceOutput.newBuilder().setStatus(Status.newBuilder()
 111                     .setErrorMessage(CdsActorConstants.TIMED_OUT).setEventType(EventType.EVENT_COMPONENT_FAILURE))
 112                     .build());
 113                 LOGGER.error("gRPC Request timed out.");
 114             }
 115         } catch (InterruptedException e) {
 116             LOGGER.error("gRPC request failed. {}", e.getMessage());
 117             cdsResponse.set(ExecutionServiceOutput.newBuilder().setStatus(Status.newBuilder()
 118                 .setErrorMessage(CdsActorConstants.INTERRUPTED).setEventType(EventType.EVENT_COMPONENT_FAILURE))
 119                 .build());
 120             Thread.currentThread().interrupt();
 121         }
 122
 123         if (!EventType.EVENT_COMPONENT_EXECUTED.equals(cdsResponse.get().getStatus().getEventType())) {
 124             String errorMessage = "Sending event \"" + eventName + "\" by " + this.name + " to CDS failed, "
 125                 + "response from CDS:\n" + cdsResponse.get();
 126             throw new ApexEventRuntimeException(errorMessage);
 127         }
 128     }
 129
 130     /**
 131      * {@inheritDoc}.
 132      */
 133     @Override
 134     public void stop() {
 135         client.close();
 136     }
 137
 138     @Override
 139     public void onMessage(ExecutionServiceOutput message) {
 140         LOGGER.info("Received notification from CDS: {}", message);
 141         cdsResponse.set(message);
 142     }
 143
 144     @Override
 145     public void onError(Throwable throwable) {
 146         String errorMsg = throwable.getLocalizedMessage();
 147         cdsResponse.set(ExecutionServiceOutput.newBuilder()
 148             .setStatus(Status.newBuilder().setErrorMessage(errorMsg).setEventType(EventType.EVENT_COMPONENT_FAILURE))
 149             .build());
 150         LOGGER.error("Failed processing blueprint {} {}", errorMsg, throwable);
 151     }
 152 }