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                             }
 144                         })
 145                         .onErrorResume(
 146                             e -> e instanceof Exception,
 147                             e -> Mono.empty())
 148                         .map(event -> {
 149                             // For each message, we have to keep separate state. This state will be enhanced
 150                             // in each step and handed off to the next processing step
 151                             PipelineState state = new PipelineState();
 152                             state.setCpeAuthenticationEvent(event);
 153                             return state;
 154                         });
 155             } catch (SSLException e) {
 156                 return Flux.error(e);
 157             }
 158         });
 159     }
 160
 161     private Mono<PipelineState> fetchPnfFromAai(PipelineState state) {
 162
 163         CpeAuthenticationConsumerDmaapModel vesEvent = state.getCpeAuthenticationEvent();
 164         String pnfName = vesEvent.getCorrelationId();
 165         String url = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName);
 166         LOGGER.debug("Processing Step: Retrieve PNF. Url: ({})", url);
 167
 168         return aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, url)
 169                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 170                 .doOnError(TimeoutException.class,
 171                         e -> LOGGER.warn("Timed out waiting for A&AI response")
 172                 )
 173                 .doOnError(e -> LOGGER.error("Error while retrieving PNF: {}",
 174                         e.getMessage())
 175                 )
 176                 .onErrorResume(
 177                     e -> e instanceof Exception,
 178                     e -> Mono.empty())
 179                 .map(p -> {
 180                     state.setPnfAaiObject(p);
 181                     return state;
 182                 });
 183     }
 184
 185     private Mono<PipelineState> fetchHsiCfsServiceInstanceFromAai(PipelineState state) {
 186
 187         if (state == null || state.getPnfAaiObject() == null) {
 188             return Mono.empty();
 189         }
 190
 191         PnfAaiObject pnf = state.getPnfAaiObject();
 192         // Assuming that the PNF will only have a single service-instance relationship pointing
 193         // towards the HSI CFS service
 194         String serviceInstanceId = pnf.getRelationshipListAaiObject().getRelationshipEntries()
 195                 .stream()
 196                 .filter(e -> "service-instance".equals(e.getRelatedTo()))
 197                 .flatMap(e -> e.getRelationshipData().stream())
 198                 .filter(d -> "service-instance.service-instance-id".equals(d.getRelationshipKey()))
 199                 .map(RelationshipListAaiObject.RelationshipDataEntryAaiObject::getRelationshipValue)
 200                 .findFirst().orElse("");
 201
 202         if (StringUtils.isEmpty(serviceInstanceId)) {
 203             LOGGER.error("Unable to retrieve HSI CFS service instance from PNF {}",
 204                     state.getPnfAaiObject().getPnfName());
 205             return Mono.empty();
 206         }
 207
 208         String url = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 209                 serviceInstanceId);
 210         LOGGER.debug("Processing Step: Retrieve HSI CFS Service. Url: ({})", url);
 211         return aaiClientTask.executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, url)
 212                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 213                 .doOnError(TimeoutException.class,
 214                         e -> LOGGER.warn("Timed out waiting for A&AI response")
 215                 )
 216                 .doOnError(e -> LOGGER.error("Error while retrieving HSI CFS Service instance: {}",
 217                         e.getMessage())
 218                 )
 219                 .onErrorResume(
 220                     e -> e instanceof Exception,
 221                     e -> Mono.empty())
 222                 .map(s -> {
 223                     state.setHsiCfsServiceInstance(s);
 224                     return state;
 225                 });
 226     }
 227
 228     private Mono<ResponseEntity<String>> triggerPolicy(PipelineState state) {
 229
 230         if (state == null || state.getHsiCfsServiceInstance() == null) {
 231             return Mono.empty();
 232         }
 233
 234         // At this point, we must check if the PNF RGW MAC address matches the value extracted from VES event
 235         if (!isCorrectMacAddress(state)) {
 236             LOGGER.warn("Processing stopped. RGW MAC address taken from event ({}) "
 237                             + "does not match with A&AI metadata corresponding value",
 238                     state.getCpeAuthenticationEvent().getRgwMacAddress());
 239             return Mono.empty();
 240         }
 241
 242         ControlLoopPublisherDmaapModel event = buildTriggeringPolicyEvent(state);
 243         return publisherTask.execute(event)
 244                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 245                 .doOnError(TimeoutException.class,
 246                         e -> LOGGER.warn("Timed out waiting for DMaaP publish confirmation")
 247                 )
 248                 .doOnError(e -> LOGGER.error("Error while triggering Policy: {}", e.getMessage()))
 249                 .onErrorResume(
 250                     e -> e instanceof Exception,
 251                     e -> Mono.empty());
 252     }
 253
 254
 255     private boolean isCorrectMacAddress(PipelineState state) {
 256         // We need to check if the RGW MAC address received in VES event matches the one found in
 257         // HSIA CFS service (in its metadata section)
 258         Optional<MetadataListAaiObject> optionalMetadata = state.getHsiCfsServiceInstance()
 259                 .getMetadataListAaiObject();
 260         String eventRgwMacAddress = state.getCpeAuthenticationEvent().getRgwMacAddress().orElse("");
 261         return optionalMetadata
 262                 .map(list -> list.getMetadataEntries()
 263                 .stream()
 264                 .anyMatch(m -> "rgw-mac-address".equals(m.getMetaname())
 265                         && m.getMetavalue().equals(eventRgwMacAddress)))
 266                 .orElse(false);
 267     }
 268
 269     private ControlLoopPublisherDmaapModel buildTriggeringPolicyEvent(PipelineState state) {
 270
 271         String cfsServiceInstanceId = state.getHsiCfsServiceInstance().getServiceInstanceId();
 272
 273         Map<String, String> enrichmentData = new HashMap<>();
 274         enrichmentData.put("service-information.hsia-cfs-service-instance-id", cfsServiceInstanceId);
 275         enrichmentData.put("cpe.old-authentication-state", state.cpeAuthenticationEvent.getOldAuthenticationState());
 276         enrichmentData.put("cpe.new-authentication-state", state.cpeAuthenticationEvent.getNewAuthenticationState());
 277         String swVersion = state.getCpeAuthenticationEvent().getSwVersion().orElse("");
 278         if (!StringUtils.isEmpty(swVersion)) {
 279             enrichmentData.put("cpe.swVersion", swVersion);
 280         }
 281
 282         ControlLoopPublisherDmaapModel triggerEvent = ImmutableControlLoopPublisherDmaapModel.builder()
 283                 .closedLoopEventClient(DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE)
 284                 .policyVersion(configuration.getPolicyVersion())
 285                 .policyName(POLICY_NAME)
 286                 .policyScope(configuration.getCpeAuthenticationCloseLoopPolicyScope())
 287                 .targetType(configuration.getCloseLoopTargetType())
 288                 .aaiEnrichmentData(enrichmentData)
 289                 .closedLoopAlarmStart(Instant.now().getEpochSecond())
 290                 .closedLoopEventStatus(configuration.getCloseLoopEventStatus())
 291                 .closedLoopControlName(configuration.getCpeAuthenticationCloseLoopControlName())
 292                 .version(configuration.getCloseLoopVersion())
 293                 .target(configuration.getCloseLoopTarget())
 294                 .requestId(UUID.randomUUID().toString())
 295                 .originator(configuration.getCloseLoopOriginator())
 296                 .build();
 297         LOGGER.debug("Processing Step: Publish for Policy");
 298         LOGGER.trace("Trigger Policy event: ({})",triggerEvent);
 299         return triggerEvent;
 300     }
 301
 302     private static class PipelineState {
 303
 304         private CpeAuthenticationConsumerDmaapModel cpeAuthenticationEvent;
 305         private PnfAaiObject pnfAaiObject;
 306         private ServiceInstanceAaiObject hsiCfsServiceInstance;
 307
 308         CpeAuthenticationConsumerDmaapModel getCpeAuthenticationEvent() {
 309             return cpeAuthenticationEvent;
 310         }
 311
 312         void setCpeAuthenticationEvent(CpeAuthenticationConsumerDmaapModel cpeAuthenticationEvent) {
 313             this.cpeAuthenticationEvent = cpeAuthenticationEvent;
 314         }
 315
 316         PnfAaiObject getPnfAaiObject() {
 317             return pnfAaiObject;
 318         }
 319
 320         void setPnfAaiObject(PnfAaiObject pnfAaiObject) {
 321             this.pnfAaiObject = pnfAaiObject;
 322         }
 323
 324         ServiceInstanceAaiObject getHsiCfsServiceInstance() {
 325             return hsiCfsServiceInstance;
 326         }
 327
 328         void setHsiCfsServiceInstance(ServiceInstanceAaiObject hsiCfsServiceInstance) {
 329             this.hsiCfsServiceInstance = hsiCfsServiceInstance;
 330         }
 331     }
 332 }