gerrit.onap Code Review - dcaegen2/services.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
   4  * ================================================================================
   5  * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.bbs.event.processor.pipelines;
  22
  23 import static org.onap.bbs.event.processor.config.ApplicationConstants.CONSUME_CPE_AUTHENTICATION_TASK_NAME;
  24 import static org.onap.bbs.event.processor.config.ApplicationConstants.DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE;
  25 import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME;
  26 import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_PNF_TASK_NAME;
  27 import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INSTANCE_UUID;
  28 import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.RESPONSE_CODE;
  29
  30 import java.time.Duration;
  31 import java.time.Instant;
  32 import java.util.HashMap;
  33 import java.util.Map;
  34 import java.util.Optional;
  35 import java.util.UUID;
  36 import java.util.concurrent.TimeoutException;
  37
  38 import javax.net.ssl.SSLException;
  39
  40 import org.onap.bbs.event.processor.config.ApplicationConfiguration;
  41 import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
  42 import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
  43 import org.onap.bbs.event.processor.model.CpeAuthenticationConsumerDmaapModel;
  44 import org.onap.bbs.event.processor.model.ImmutableControlLoopPublisherDmaapModel;
  45 import org.onap.bbs.event.processor.model.MetadataListAaiObject;
  46 import org.onap.bbs.event.processor.model.PnfAaiObject;
  47 import org.onap.bbs.event.processor.model.RelationshipListAaiObject;
  48 import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
  49 import org.onap.bbs.event.processor.tasks.AaiClientTask;
  50 import org.onap.bbs.event.processor.tasks.DmaapCpeAuthenticationConsumerTask;
  51 import org.onap.bbs.event.processor.tasks.DmaapPublisherTask;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54 import org.slf4j.MDC;
  55 import org.springframework.beans.factory.annotation.Autowired;
  56 import org.springframework.http.ResponseEntity;
  57 import org.springframework.stereotype.Component;
  58 import org.springframework.util.StringUtils;
  59
  60 import reactor.core.publisher.Flux;
  61 import reactor.core.publisher.Mono;
  62
  63 @Component
  64 public class CpeAuthenticationPipeline {
  65
  66     private static final Logger LOGGER = LoggerFactory.getLogger(CpeAuthenticationPipeline.class);
  67
  68     private static final String POLICY_NAME = "CPE_Authentication";
  69
  70     private DmaapCpeAuthenticationConsumerTask consumerTask;
  71     private DmaapPublisherTask publisherTask;
  72     private AaiClientTask aaiClientTask;
  73
  74     private ApplicationConfiguration configuration;
  75
  76     private Map<String, String> mdcContextMap;
  77
  78     @Autowired
  79     CpeAuthenticationPipeline(ApplicationConfiguration configuration,
  80                               DmaapCpeAuthenticationConsumerTask consumerTask,
  81                               DmaapPublisherTask publisherTask,
  82                               AaiClientTask aaiClientTask,
  83                               Map<String, String> mdcContextMap) {
  84         this.configuration = configuration;
  85         this.consumerTask = consumerTask;
  86         this.publisherTask = publisherTask;
  87         this.aaiClientTask = aaiClientTask;
  88         this.mdcContextMap = mdcContextMap;
  89     }
  90
  91     /**
  92      * PNF CPE Authentication processing pipeline for BBS uS.
  93      */
  94     public void processPnfCpeAuthenticationEvents() {
  95         MDC.setContextMap(mdcContextMap);
  96         LOGGER.info("Process next CPE Authentication events");
  97         executePipeline()
  98                 .subscribe(this::onSuccess, this::onError, this::onComplete);
  99         LOGGER.trace("Reactive CPE Authentication pipeline subscribed - Execution started");
 100     }
 101
 102     Flux<ResponseEntity<String>> executePipeline() {
 103         return
 104             // Consume CPE Authentication from DMaaP
 105             consumeCpeAuthenticationFromDmaap()
 106             // Fetch PNF from A&AI
 107             .flatMap(this::fetchPnfFromAai)
 108             // Fetch related HSI CFS instance from A&AI
 109             .flatMap(this::fetchHsiCfsServiceInstanceFromAai)
 110             // Trigger Policy for relocation
 111             .flatMap(this::triggerPolicy);
 112     }
 113
 114     private void onSuccess(ResponseEntity<String> responseCode) {
 115         MDC.put(RESPONSE_CODE, responseCode.getStatusCode().toString());
 116         LOGGER.info("CPE Authentication event successfully handled. "
 117                         + "Publishing to DMaaP for Policy returned a status code of ({} {})",
 118                 responseCode.getStatusCode().value(), responseCode.getStatusCode().getReasonPhrase());
 119         MDC.remove(RESPONSE_CODE);
 120     }
 121
 122     private void onError(Throwable throwable) {
 123         LOGGER.error("Aborted CPE Authentication events processing. Error: {}", throwable.getMessage());
 124     }
 125
 126     private void onComplete() {
 127         LOGGER.info("CPE Authentication processing pipeline has been completed");
 128     }
 129
 130     private Flux<PipelineState> consumeCpeAuthenticationFromDmaap() {
 131         return Flux.defer(() -> {
 132             MDC.put(INSTANCE_UUID, UUID.randomUUID().toString());
 133             try {
 134                 return consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME)
 135                         .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 136                         .doOnError(e -> {
 137                             if (e instanceof TimeoutException) {
 138                                 LOGGER.warn("Timed out waiting for DMaaP response");
 139                             } else if (e instanceof EmptyDmaapResponseException) {
 140                                 LOGGER.info("Nothing to consume from DMaaP");
 141                             } else {
 142                                 LOGGER.error("DMaaP Consumer error: {}", e.getMessage());
 143                                 LOGGER.debug("Error\n", e);
 144                             }
 145                         })
 146                         .onErrorResume(
 147                             e -> e instanceof Exception,
 148                             e -> Mono.empty())
 149                         .map(event -> {
 150                             // For each message, we have to keep separate state. This state will be enhanced
 151                             // in each step and handed off to the next processing step
 152                             PipelineState state = new PipelineState();
 153                             state.setCpeAuthenticationEvent(event);
 154                             return state;
 155                         });
 156             } catch (SSLException e) {
 157                 return Flux.error(e);
 158             }
 159         });
 160     }
 161
 162     private Mono<PipelineState> fetchPnfFromAai(PipelineState state) {
 163
 164         CpeAuthenticationConsumerDmaapModel vesEvent = state.getCpeAuthenticationEvent();
 165         String pnfName = vesEvent.getCorrelationId();
 166         String url = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName);
 167         LOGGER.debug("Processing Step: Retrieve PNF. Url: ({})", url);
 168
 169         return aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, url)
 170                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 171                 .doOnError(TimeoutException.class,
 172                         e -> LOGGER.warn("Timed out waiting for A&AI response")
 173                 )
 174                 .doOnError(e -> LOGGER.error("Error while retrieving PNF: {}",
 175                         e.getMessage())
 176                 )
 177                 .onErrorResume(
 178                     e -> e instanceof Exception,
 179                     e -> Mono.empty())
 180                 .map(p -> {
 181                     state.setPnfAaiObject(p);
 182                     return state;
 183                 });
 184     }
 185
 186     private Mono<PipelineState> fetchHsiCfsServiceInstanceFromAai(PipelineState state) {
 187
 188         if (state == null || state.getPnfAaiObject() == null) {
 189             return Mono.empty();
 190         }
 191
 192         PnfAaiObject pnf = state.getPnfAaiObject();
 193         // Assuming that the PNF will only have a single service-instance relationship pointing
 194         // towards the HSI CFS service
 195         String serviceInstanceId = pnf.getRelationshipListAaiObject().getRelationshipEntries()
 196                 .stream()
 197                 .filter(e -> "service-instance".equals(e.getRelatedTo()))
 198                 .flatMap(e -> e.getRelationshipData().stream())
 199                 .filter(d -> "service-instance.service-instance-id".equals(d.getRelationshipKey()))
 200                 .map(RelationshipListAaiObject.RelationshipDataEntryAaiObject::getRelationshipValue)
 201                 .findFirst().orElse("");
 202
 203         if (StringUtils.isEmpty(serviceInstanceId)) {
 204             LOGGER.error("Unable to retrieve HSI CFS service instance from PNF {}",
 205                     state.getPnfAaiObject().getPnfName());
 206             return Mono.empty();
 207         }
 208
 209         String url = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 210                 serviceInstanceId);
 211         LOGGER.debug("Processing Step: Retrieve HSI CFS Service. Url: ({})", url);
 212         return aaiClientTask.executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, url)
 213                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 214                 .doOnError(TimeoutException.class,
 215                         e -> LOGGER.warn("Timed out waiting for A&AI response")
 216                 )
 217                 .doOnError(e -> LOGGER.error("Error while retrieving HSI CFS Service instance: {}",
 218                         e.getMessage())
 219                 )
 220                 .onErrorResume(
 221                     e -> e instanceof Exception,
 222                     e -> Mono.empty())
 223                 .map(s -> {
 224                     state.setHsiCfsServiceInstance(s);
 225                     return state;
 226                 });
 227     }
 228
 229     private Mono<ResponseEntity<String>> triggerPolicy(PipelineState state) {
 230
 231         if (state == null || state.getHsiCfsServiceInstance() == null) {
 232             return Mono.empty();
 233         }
 234
 235         // At this point, we must check if the PNF RGW MAC address matches the value extracted from VES event
 236         if (!isCorrectMacAddress(state)) {
 237             LOGGER.warn("Processing stopped. RGW MAC address taken from event ({}) "
 238                             + "does not match with A&AI metadata corresponding value",
 239                     state.getCpeAuthenticationEvent().getRgwMacAddress());
 240             return Mono.empty();
 241         }
 242
 243         ControlLoopPublisherDmaapModel event = buildTriggeringPolicyEvent(state);
 244         return publisherTask.execute(event)
 245                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 246                 .doOnError(TimeoutException.class,
 247                         e -> LOGGER.warn("Timed out waiting for DMaaP publish confirmation")
 248                 )
 249                 .doOnError(e -> LOGGER.error("Error while triggering Policy: {}", e.getMessage()))
 250                 .onErrorResume(
 251                     e -> e instanceof Exception,
 252                     e -> Mono.empty());
 253     }
 254
 255
 256     private boolean isCorrectMacAddress(PipelineState state) {
 257         // We need to check if the RGW MAC address received in VES event matches the one found in
 258         // HSIA CFS service (in its metadata section)
 259         Optional<MetadataListAaiObject> optionalMetadata = state.getHsiCfsServiceInstance()
 260                 .getMetadataListAaiObject();
 261         String eventRgwMacAddress = state.getCpeAuthenticationEvent().getRgwMacAddress().orElse("");
 262         return optionalMetadata
 263                 .map(list -> list.getMetadataEntries()
 264                 .stream()
 265                 .anyMatch(m -> "rgw-mac-address".equals(m.getMetaname())
 266                         && m.getMetavalue().equals(eventRgwMacAddress)))
 267                 .orElse(false);
 268     }
 269
 270     private ControlLoopPublisherDmaapModel buildTriggeringPolicyEvent(PipelineState state) {
 271
 272         String cfsServiceInstanceId = state.getHsiCfsServiceInstance().getServiceInstanceId();
 273
 274         Map<String, String> enrichmentData = new HashMap<>();
 275         enrichmentData.put("service-information.hsia-cfs-service-instance-id", cfsServiceInstanceId);
 276         enrichmentData.put("cpe.old-authentication-state", state.cpeAuthenticationEvent.getOldAuthenticationState());
 277         enrichmentData.put("cpe.new-authentication-state", state.cpeAuthenticationEvent.getNewAuthenticationState());
 278         String swVersion = state.getCpeAuthenticationEvent().getSwVersion().orElse("");
 279         if (!StringUtils.isEmpty(swVersion)) {
 280             enrichmentData.put("cpe.swVersion", swVersion);
 281         }
 282
 283         ControlLoopPublisherDmaapModel triggerEvent = ImmutableControlLoopPublisherDmaapModel.builder()
 284                 .closedLoopEventClient(DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE)
 285                 .policyVersion(configuration.getPolicyVersion())
 286                 .policyName(POLICY_NAME)
 287                 .policyScope(configuration.getCpeAuthenticationCloseLoopPolicyScope())
 288                 .targetType(configuration.getCloseLoopTargetType())
 289                 .aaiEnrichmentData(enrichmentData)
 290                 .closedLoopAlarmStart(Instant.now().getEpochSecond())
 291                 .closedLoopEventStatus(configuration.getCloseLoopEventStatus())
 292                 .closedLoopControlName(configuration.getCpeAuthenticationCloseLoopControlName())
 293                 .version(configuration.getCloseLoopVersion())
 294                 .target(configuration.getCloseLoopTarget())
 295                 .requestId(UUID.randomUUID().toString())
 296                 .originator(configuration.getCloseLoopOriginator())
 297                 .build();
 298         LOGGER.debug("Processing Step: Publish for Policy");
 299         LOGGER.trace("Trigger Policy event: ({})",triggerEvent);
 300         return triggerEvent;
 301     }
 302
 303     private static class PipelineState {
 304
 305         private CpeAuthenticationConsumerDmaapModel cpeAuthenticationEvent;
 306         private PnfAaiObject pnfAaiObject;
 307         private ServiceInstanceAaiObject hsiCfsServiceInstance;
 308
 309         CpeAuthenticationConsumerDmaapModel getCpeAuthenticationEvent() {
 310             return cpeAuthenticationEvent;
 311         }
 312
 313         void setCpeAuthenticationEvent(CpeAuthenticationConsumerDmaapModel cpeAuthenticationEvent) {
 314             this.cpeAuthenticationEvent = cpeAuthenticationEvent;
 315         }
 316
 317         PnfAaiObject getPnfAaiObject() {
 318             return pnfAaiObject;
 319         }
 320
 321         void setPnfAaiObject(PnfAaiObject pnfAaiObject) {
 322             this.pnfAaiObject = pnfAaiObject;
 323         }
 324
 325         ServiceInstanceAaiObject getHsiCfsServiceInstance() {
 326             return hsiCfsServiceInstance;
 327         }
 328
 329         void setHsiCfsServiceInstance(ServiceInstanceAaiObject hsiCfsServiceInstance) {
 330             this.hsiCfsServiceInstance = hsiCfsServiceInstance;
 331         }
 332     }
 333 }