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