gerrit.onap Code Review - policy/drools-applications.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2019 Bell Canada.
   4  * Modifications Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  * ============LICENSE_END=========================================================
  18  */
  19
  20 package org.onap.policy.template.demo;
  21
  22 import static org.junit.Assert.assertEquals;
  23 import static org.junit.Assert.assertNotNull;
  24 import static org.junit.Assert.assertTrue;
  25 import static org.junit.Assert.fail;
  26
  27 import io.grpc.Server;
  28 import io.grpc.ServerBuilder;
  29 import io.grpc.stub.StreamObserver;
  30
  31 import java.io.IOException;
  32 import java.time.Instant;
  33 import java.util.HashMap;
  34 import java.util.Map;
  35 import java.util.UUID;
  36 import java.util.concurrent.atomic.AtomicReference;
  37
  38 import org.junit.After;
  39 import org.junit.Before;
  40 import org.junit.BeforeClass;
  41 import org.junit.Test;
  42 import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader;
  43 import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType;
  44 import org.onap.ccsdk.cds.controllerblueprints.common.api.Status;
  45 import org.onap.ccsdk.cds.controllerblueprints.processing.api.BluePrintProcessingServiceGrpc;
  46 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
  47 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
  48 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  49 import org.onap.policy.common.endpoints.event.comm.TopicListener;
  50 import org.onap.policy.common.endpoints.event.comm.TopicSink;
  51 import org.onap.policy.controlloop.ControlLoopEventStatus;
  52 import org.onap.policy.controlloop.ControlLoopNotificationType;
  53 import org.onap.policy.controlloop.ControlLoopTargetType;
  54 import org.onap.policy.controlloop.VirtualControlLoopEvent;
  55 import org.onap.policy.controlloop.VirtualControlLoopNotification;
  56 import org.onap.policy.controlloop.policy.ControlLoopPolicy;
  57 import org.onap.policy.controlloop.util.Serialization;
  58 import org.slf4j.Logger;
  59 import org.slf4j.LoggerFactory;
  60
  61 /**
  62  * Test class for vfirewall use case using CDS actor.
  63  */
  64 public class VfwControlLoopCdsTest extends ControlLoopBase implements TopicListener {
  65
  66     private static final Logger LOGGER = LoggerFactory.getLogger(VfwControlLoopCdsTest.class);
  67
  68     private final AtomicReference<StreamObserver<ExecutionServiceOutput>> responseObserverRef = new AtomicReference<>();
  69     private Server server;
  70
  71     /**
  72      * Setup the simulator.
  73      */
  74     @BeforeClass
  75     public static void setUpBeforeClass() {
  76         ControlLoopBase.setUpBeforeClass("../archetype-cl-amsterdam/src/main/resources/archetype-resources/src/"
  77                                                  + "main/resources/__closedLoopControlName__.drl",
  78                 "src/test/resources/yaml/policy_ControlLoop_vFW_CDS.yaml",
  79                 "service=ServiceDemo;resource=Res1Demo;type=operational", "CL_vFW",
  80                 "org.onap.closed_loop.ServiceDemo:VNFS:1.0.0");
  81         SupportUtil.setCustomQuery("true");
  82     }
  83
  84     @Before
  85     public void setUp() throws IOException {
  86         this.startGrpcServer();
  87     }
  88
  89     @After
  90     public void tearDown() {
  91         this.stopGrpcServer();
  92     }
  93
  94     private void startGrpcServer() throws IOException {
  95
  96         BluePrintProcessingServiceGrpc.BluePrintProcessingServiceImplBase cdsBlueprintServerImpl =
  97
  98             new BluePrintProcessingServiceGrpc.BluePrintProcessingServiceImplBase() {
  99
 100                 @Override
 101                 public StreamObserver<ExecutionServiceInput> process(
 102                         final StreamObserver<ExecutionServiceOutput> responseObserver) {
 103
 104                     responseObserverRef.set(responseObserver);
 105
 106                     return new StreamObserver<ExecutionServiceInput>() {
 107                         @Override
 108                         public void onNext(final ExecutionServiceInput input) {
 109                             LOGGER.info("gRPC server onNext() for input: {} ...", input);
 110                             ExecutionServiceOutput output =
 111                                     ExecutionServiceOutput.newBuilder()
 112                                             .setCommonHeader(
 113                                                     CommonHeader.newBuilder().setRequestId(
 114                                                             input.getCommonHeader().getRequestId()).build())
 115                                             .setStatus(
 116                                                     Status.newBuilder().setEventType(
 117                                                             EventType.EVENT_COMPONENT_EXECUTED).build())
 118                                             .build();
 119                             responseObserver.onNext(output);
 120                         }
 121
 122                         @Override
 123                         public void onError(final Throwable throwable) {
 124                             LOGGER.error("gRPC server onError() for throwable: {} ...", throwable);
 125                         }
 126
 127                         @Override
 128                         public void onCompleted() {
 129                             LOGGER.info("gRPC server onCompleted() ...");
 130                             responseObserver.onCompleted();
 131                         }
 132                     };
 133                 }
 134             };
 135
 136         server = ServerBuilder.forPort(SupportUtil.GRPC_SERVER_PORT).addService(cdsBlueprintServerImpl).build().start();
 137         LOGGER.info("gRPC server is listening for CDS requests on port {}", SupportUtil.GRPC_SERVER_PORT);
 138
 139     }
 140
 141     private void stopGrpcServer() {
 142         if (server != null) {
 143             this.server.shutdown();
 144             LOGGER.info("gRPC server handling CDS requests has been successfully shut down.");
 145         }
 146     }
 147
 148     @Test
 149     public void testSuccess() {
 150
 151         /*
 152          * Allows the PolicyEngine to callback to this object to notify that there is an event ready
 153          * to be pulled from the queue
 154          */
 155         for (TopicSink sink : noopTopics) {
 156             assertTrue(sink.start());
 157             sink.register(this);
 158         }
 159
 160         /*
 161          * Create a unique requestId
 162          */
 163         requestId = UUID.randomUUID();
 164
 165         /*
 166          * Simulate an onset event the policy engine will receive from DCAE to kick off processing
 167          * through the rules
 168          */
 169         sendEvent(pair.first, requestId, ControlLoopEventStatus.ONSET);
 170
 171         kieSession.fireUntilHalt();
 172         kieSession.fireAllRules();
 173
 174         /*
 175          * The only fact in memory should be Params
 176          */
 177         assertEquals(1, kieSession.getFactCount());
 178
 179         /*
 180          * Print what's left in memory
 181          */
 182         dumpFacts(kieSession);
 183     }
 184
 185
 186     /*
 187      * @see org.onap.policy.drools.PolicyEngineListener#newEventNotification(java.lang.String)
 188      */
 189     @Override
 190     public void onTopicEvent(CommInfrastructure commType, String topic, String event) {
 191         /*
 192          * Pull the object that was sent out to DMAAP and make sure it is a ControlLoopNotification
 193          * of type active
 194          */
 195         assertEquals("POLICY-CL-MGT", topic);
 196         VirtualControlLoopNotification notification =
 197                 Serialization.gsonJunit.fromJson(event, VirtualControlLoopNotification.class);
 198         assertNotNull(notification);
 199         String policyName = notification.getPolicyName();
 200         if (policyName.endsWith("EVENT")) {
 201             logger.debug("Rule Fired: " + notification.getPolicyName());
 202             assertEquals(ControlLoopNotificationType.ACTIVE, notification.getNotification());
 203         } else if (policyName.endsWith("GUARD_NOT_YET_QUERIED")) {
 204             logger.debug("Rule Fired: " + notification.getPolicyName());
 205             assertEquals(ControlLoopNotificationType.OPERATION, notification.getNotification());
 206             assertNotNull(notification.getMessage());
 207             assertTrue(notification.getMessage().startsWith("Sending guard query"));
 208         } else if (policyName.endsWith("GUARD.RESPONSE")) {
 209             logger.debug("Rule Fired: " + notification.getPolicyName());
 210             assertEquals(ControlLoopNotificationType.OPERATION, notification.getNotification());
 211             assertNotNull(notification.getMessage());
 212             assertTrue(notification.getMessage().toLowerCase().endsWith("permit"));
 213         } else if (policyName.endsWith("GUARD_PERMITTED")) {
 214             logger.debug("Rule Fired: " + notification.getPolicyName());
 215             assertEquals(ControlLoopNotificationType.OPERATION, notification.getNotification());
 216             assertNotNull(notification.getMessage());
 217             assertTrue(notification.getMessage().startsWith("actor=CDS"));
 218         } else if (policyName.endsWith("OPERATION.TIMEOUT")) {
 219             logger.debug("Rule Fired: " + notification.getPolicyName());
 220             kieSession.halt();
 221             logger.debug("The operation timed out");
 222             fail("Operation Timed Out");
 223         } else if (policyName.endsWith("CDS.RESPONSE")) {
 224             logger.debug("Rule Fired: " + notification.getPolicyName());
 225             assertEquals(ControlLoopNotificationType.OPERATION_SUCCESS, notification.getNotification());
 226             assertNotNull(notification.getMessage());
 227             assertTrue(notification.getMessage().startsWith("actor=CDS"));
 228             sendEvent(pair.first, requestId, ControlLoopEventStatus.ABATED);
 229         } else if (policyName.endsWith("EVENT.MANAGER")) {
 230             logger.debug("Rule Fired: " + notification.getPolicyName());
 231             if ("error".equals(notification.getAai().get("generic-vnf.vnf-name"))) {
 232                 assertEquals(ControlLoopNotificationType.FINAL_FAILURE, notification.getNotification());
 233                 assertEquals("Target vnf-id could not be found", notification.getMessage());
 234             } else if ("getFail".equals(notification.getAai().get("generic-vnf.vnf-name"))) {
 235                 assertEquals(ControlLoopNotificationType.FINAL_FAILURE, notification.getNotification());
 236             } else {
 237                 assertEquals(ControlLoopNotificationType.FINAL_SUCCESS, notification.getNotification());
 238             }
 239             kieSession.halt();
 240         } else if (policyName.endsWith("EVENT.MANAGER.TIMEOUT")) {
 241             logger.debug("Rule Fired: " + notification.getPolicyName());
 242             kieSession.halt();
 243             logger.debug("The control loop timed out");
 244             fail("Control Loop Timed Out");
 245         }
 246     }
 247
 248     /**
 249      * This method is used to simulate event messages from DCAE that start the control loop (onset
 250      * message) or end the control loop (abatement message).
 251      *
 252      * @param policy the controlLoopName comes from the policy
 253      * @param requestId the requestId for this event
 254      * @param status could be onset or abated
 255      */
 256     private void sendEvent(ControlLoopPolicy policy, UUID requestId, ControlLoopEventStatus status) {
 257         VirtualControlLoopEvent event = new VirtualControlLoopEvent();
 258         event.setClosedLoopControlName(policy.getControlLoop().getControlLoopName());
 259         event.setRequestId(requestId);
 260         event.setTargetType(ControlLoopTargetType.VNF);
 261         event.setTarget("generic-vnf.vnf-name");
 262         event.setClosedLoopAlarmStart(Instant.now());
 263         event.setAai(new HashMap<>());
 264         event.getAai().put("generic-vnf.vnf-name", "testGenericVnfID");
 265         event.getAai().put("vserver.vserver-name", "OzVServer");
 266         event.setClosedLoopEventStatus(status);
 267         Map<String, String> map = new HashMap<>();
 268         map.put("my-key", "my-value");
 269         event.setAdditionalEventParams(map);
 270         kieSession.insert(event);
 271     }
 272
 273 }