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