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