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_REREGISTRATION_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.List;
  33 import java.util.Map;
  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.ImmutableControlLoopPublisherDmaapModel;
  45 import org.onap.bbs.event.processor.model.PnfAaiObject;
  46 import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
  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.DmaapPublisherTask;
  51 import org.onap.bbs.event.processor.tasks.DmaapReRegistrationConsumerTask;
  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 ReRegistrationPipeline {
  65
  66     private static final Logger LOGGER = LoggerFactory.getLogger(ReRegistrationPipeline.class);
  67
  68     private static final String DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE = "DCAE.BBS_event_processor_mSInstance";
  69     private static final String POLICY_VERSION = "1.0.0.5";
  70     private static final String POLICY_NAME = "Nomadic_ONT";
  71     private static final String CLOSE_LOOP_TARGET_TYPE = "VM";
  72     private static final String CLOSED_LOOP_EVENT_STATUS = "ONSET";
  73     private static final String CLOSE_LOOP_VERSION = "1.0.2";
  74     private static final String CLOSE_LOOP_TARGET = "vserver.vserver-name";
  75     private static final String FROM = "DCAE";
  76
  77     private DmaapReRegistrationConsumerTask consumerTask;
  78     private DmaapPublisherTask publisherTask;
  79     private AaiClientTask aaiClientTask;
  80
  81     private ApplicationConfiguration configuration;
  82
  83     private Map<String, String> mdcContextMap;
  84
  85     @Autowired
  86     ReRegistrationPipeline(ApplicationConfiguration configuration,
  87                            DmaapReRegistrationConsumerTask consumerTask,
  88                            DmaapPublisherTask publisherTask,
  89                            AaiClientTask aaiClientTask,
  90                            Map<String, String> mdcContextMap) {
  91         this.configuration = configuration;
  92         this.consumerTask = consumerTask;
  93         this.publisherTask = publisherTask;
  94         this.aaiClientTask = aaiClientTask;
  95         this.mdcContextMap = mdcContextMap;
  96     }
  97
  98     /**
  99      * PNF re-registration processing pipeline for BBS uS.
 100      */
 101     public void processPnfReRegistrationEvents() {
 102         MDC.setContextMap(mdcContextMap);
 103         LOGGER.info("Process next Re-Registration events");
 104         executePipeline()
 105                 .subscribe(this::onSuccess, this::onError, this::onComplete);
 106         LOGGER.trace("Reactive PNF Re-registration pipeline subscribed - Execution started");
 107     }
 108
 109     Flux<ResponseEntity<String>> executePipeline() {
 110         return
 111             // Consume Re-Registration from DMaaP
 112             consumeReRegistrationsFromDmaap()
 113             // Fetch PNF from A&AI
 114             .flatMap(this::fetchPnfFromAai)
 115             // Fetch related HSI CFS instance from A&AI
 116             .flatMap(this::fetchHsiCfsServiceInstanceFromAai)
 117             // Trigger Policy for relocation
 118             .flatMap(this::triggerPolicy);
 119     }
 120
 121     private void onSuccess(ResponseEntity<String> responseCode) {
 122         MDC.put(RESPONSE_CODE, responseCode.getStatusCode().toString());
 123         LOGGER.info("PNF Re-Registration event successfully handled. "
 124                         + "Publishing to DMaaP for Policy returned a status code of ({} {})",
 125                 responseCode.getStatusCode().value(), responseCode.getStatusCode().getReasonPhrase());
 126         MDC.remove(RESPONSE_CODE);
 127     }
 128
 129     private void onError(Throwable throwable) {
 130         LOGGER.error("Aborted PNF Re-Registration events processing. Error: {}", throwable.getMessage());
 131     }
 132
 133     private void onComplete() {
 134         LOGGER.info("PNF Re-Registration processing pipeline has been completed");
 135     }
 136
 137     private Flux<PipelineState> consumeReRegistrationsFromDmaap() {
 138         return Flux.defer(() -> {
 139             MDC.put(INSTANCE_UUID, UUID.randomUUID().toString());
 140             try {
 141                 return consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)
 142                         .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 143                         .doOnError(e -> {
 144                             if (e instanceof TimeoutException) {
 145                                 LOGGER.warn("Timed out waiting for DMaaP response");
 146                             } else if (e instanceof EmptyDmaapResponseException) {
 147                                 LOGGER.warn("Nothing to consume from DMaaP");
 148                             }
 149                         })
 150                         .onErrorResume(
 151                             e -> (e instanceof EmptyDmaapResponseException || e instanceof TimeoutException),
 152                             e -> Mono.empty())
 153                         .map(event -> {
 154                             // For each message, we have to keep separate state. This state will be enhanced
 155                             // in each step and handed off to the next processing step
 156                             PipelineState state = new PipelineState();
 157                             state.setReRegistrationEvent(event);
 158                             return state;
 159                         });
 160             } catch (SSLException e) {
 161                 return Flux.error(e);
 162             }
 163         });
 164     }
 165
 166     private Mono<PipelineState> fetchPnfFromAai(PipelineState state) {
 167
 168         ReRegistrationConsumerDmaapModel vesEvent = state.getReRegistrationEvent();
 169         String pnfName = vesEvent.getCorrelationId();
 170         String url = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName);
 171         LOGGER.debug("Processing Step: Retrieve PNF. Url: ({})", url);
 172
 173         return aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, url)
 174                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 175                 .doOnError(TimeoutException.class,
 176                         e -> LOGGER.warn("Timed out waiting for A&AI response")
 177                 )
 178                 .doOnError(e -> LOGGER.error("Error while retrieving PNF: {}",
 179                         e.getMessage())
 180                 )
 181                 .onErrorResume(
 182                     e -> e instanceof AaiTaskException || e instanceof TimeoutException,
 183                     e -> Mono.empty())
 184                 .map(p -> {
 185                     state.setPnfAaiObject(p);
 186                     return state;
 187                 });
 188     }
 189
 190     private Mono<PipelineState> fetchHsiCfsServiceInstanceFromAai(PipelineState state) {
 191
 192         if (state == null || state.getPnfAaiObject() == null) {
 193             return Mono.empty();
 194         }
 195
 196         // At this point, we have both the VES-event of the re-registration and the PNF object retrieved from A&AI
 197         // We can check if this processing needs to continue in case of a true relocation
 198         if (isNotReallyAnOntRelocation(state)) {
 199             return Mono.empty();
 200         }
 201
 202         PnfAaiObject pnf = state.getPnfAaiObject();
 203         // Assuming that the PNF will only have a single service-instance relationship pointing
 204         // towards the HSI CFS service
 205         String serviceInstanceId = pnf.getRelationshipListAaiObject().getRelationshipEntries()
 206                 .stream()
 207                 .filter(e -> e.getRelatedTo().equals("service-instance"))
 208                 .flatMap(e -> e.getRelationshipData().stream())
 209                 .filter(d -> d.getRelationshipKey().equals("service-instance.service-instance-id"))
 210                 .map(RelationshipListAaiObject.RelationshipDataEntryAaiObject::getRelationshipValue)
 211                 .findFirst().orElse("");
 212
 213         if (StringUtils.isEmpty(serviceInstanceId)) {
 214             LOGGER.error("Unable to retrieve HSI CFS service instance from PNF {}",
 215                     state.getPnfAaiObject().getPnfName());
 216             return Mono.empty();
 217         }
 218
 219         String url = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 220                 serviceInstanceId);
 221         LOGGER.debug("Processing Step: Retrieve HSI CFS Service. Url: ({})", url);
 222         return aaiClientTask.executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, url)
 223                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 224                 .doOnError(TimeoutException.class,
 225                         e -> LOGGER.warn("Timed out waiting for A&AI response")
 226                 )
 227                 .doOnError(e -> LOGGER.error("Error while retrieving HSI CFS Service instance: {}",
 228                         e.getMessage())
 229                 )
 230                 .onErrorResume(
 231                     e -> e instanceof AaiTaskException || e instanceof TimeoutException,
 232                     e -> Mono.empty())
 233                 .map(s -> {
 234                     state.setHsiCfsServiceInstance(s);
 235                     return state;
 236                 });
 237     }
 238
 239     private boolean isNotReallyAnOntRelocation(PipelineState state) {
 240         List<RelationshipListAaiObject.RelationshipEntryAaiObject> relationshipEntries =
 241                 state.getPnfAaiObject().getRelationshipListAaiObject().getRelationshipEntries();
 242
 243         // If no logical-link, fail further processing
 244         if (relationshipEntries.stream().noneMatch(e -> e.getRelatedTo().equals("logical-link"))) {
 245             LOGGER.warn("PNF {} does not have any logical-links bridged. Stop further processing",
 246                     state.getPnfAaiObject().getPnfName());
 247             return true;
 248         }
 249
 250         // Assuming PNF will only have one logical-link per BBS use case design
 251         boolean isNotRelocation = relationshipEntries
 252                 .stream()
 253                 .filter(e -> e.getRelatedTo().equals("logical-link"))
 254                 .flatMap(e -> e.getRelationshipData().stream())
 255                 .anyMatch(d -> d.getRelationshipValue()
 256                         .equals(state.getReRegistrationEvent().getAttachmentPoint()));
 257
 258
 259         if (isNotRelocation) {
 260             LOGGER.warn("Not a Relocation for PNF {} with attachment point {}",
 261                     state.getPnfAaiObject().getPnfName(),
 262                     state.getReRegistrationEvent().getAttachmentPoint());
 263         }
 264         return isNotRelocation;
 265     }
 266
 267     private Mono<ResponseEntity<String>> triggerPolicy(PipelineState state) {
 268
 269         if (state == null || state.getHsiCfsServiceInstance() == null) {
 270             return Mono.empty();
 271         }
 272
 273         ControlLoopPublisherDmaapModel event = buildTriggeringPolicyEvent(state);
 274         return publisherTask.execute(event)
 275                 .timeout(Duration.ofSeconds(configuration.getPipelinesTimeoutInSeconds()))
 276                 .doOnError(TimeoutException.class,
 277                         e -> LOGGER.warn("Timed out waiting for DMaaP confirmation")
 278                 )
 279                 .doOnError(e -> LOGGER.error("Error while triggering Policy: {}", e.getMessage()))
 280                 .onErrorResume(
 281                     e -> e instanceof DmaapException || e instanceof TimeoutException,
 282                     e -> Mono.empty());
 283     }
 284
 285     private ControlLoopPublisherDmaapModel buildTriggeringPolicyEvent(PipelineState state) {
 286
 287         String cfsServiceInstanceId = state.getHsiCfsServiceInstance().getServiceInstanceId();
 288
 289         String attachmentPoint = state.getReRegistrationEvent().getAttachmentPoint();
 290         String remoteId = state.getReRegistrationEvent().getRemoteId();
 291         String cvlan = state.getReRegistrationEvent().getCVlan();
 292         String svlan = state.getReRegistrationEvent().getSVlan();
 293
 294         Map<String, String> enrichmentData = new HashMap<>();
 295         enrichmentData.put("service-information.hsia-cfs-service-instance-id", cfsServiceInstanceId);
 296
 297         enrichmentData.put("attachmentPoint", attachmentPoint);
 298         enrichmentData.put("remoteId", remoteId);
 299         enrichmentData.put("cvlan", cvlan);
 300         enrichmentData.put("svlan", svlan);
 301
 302         ControlLoopPublisherDmaapModel triggerEvent = ImmutableControlLoopPublisherDmaapModel.builder()
 303                 .closedLoopEventClient(DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE)
 304                 .policyVersion(POLICY_VERSION)
 305                 .policyName(POLICY_NAME)
 306                 .policyScope(configuration.getReRegistrationCloseLoopPolicyScope())
 307                 .targetType(CLOSE_LOOP_TARGET_TYPE)
 308                 .aaiEnrichmentData(enrichmentData)
 309                 .closedLoopAlarmStart(Instant.now().getEpochSecond())
 310                 .closedLoopEventStatus(CLOSED_LOOP_EVENT_STATUS)
 311                 .closedLoopControlName(configuration.getReRegistrationCloseLoopControlName())
 312                 .version(CLOSE_LOOP_VERSION)
 313                 .target(CLOSE_LOOP_TARGET)
 314                 .requestId(UUID.randomUUID().toString())
 315                 .originator(FROM)
 316                 .build();
 317         LOGGER.debug("Processing Step: Publish for Policy");
 318         LOGGER.trace("Trigger Policy event: ({})",triggerEvent);
 319         return triggerEvent;
 320     }
 321
 322     private static class PipelineState {
 323
 324         private ReRegistrationConsumerDmaapModel reRegistrationEvent;
 325         private PnfAaiObject pnfAaiObject;
 326         private ServiceInstanceAaiObject hsiCfsServiceInstance;
 327
 328         ReRegistrationConsumerDmaapModel getReRegistrationEvent() {
 329             return reRegistrationEvent;
 330         }
 331
 332         void setReRegistrationEvent(ReRegistrationConsumerDmaapModel reRegistrationEvent) {
 333             this.reRegistrationEvent = reRegistrationEvent;
 334         }
 335
 336         PnfAaiObject getPnfAaiObject() {
 337             return pnfAaiObject;
 338         }
 339
 340         void setPnfAaiObject(PnfAaiObject pnfAaiObject) {
 341             this.pnfAaiObject = pnfAaiObject;
 342         }
 343
 344         ServiceInstanceAaiObject getHsiCfsServiceInstance() {
 345             return hsiCfsServiceInstance;
 346         }
 347
 348         void setHsiCfsServiceInstance(ServiceInstanceAaiObject hsiCfsServiceInstance) {
 349             this.hsiCfsServiceInstance = hsiCfsServiceInstance;
 350         }
 351     }
 352 }