enrichment-coordinator-service/src/main/java/org/oransc/enrichment/tasks/ProducerSupervision.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2020 Nordix Foundation
   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.oransc.enrichment.tasks;
  22
  23 import org.oransc.enrichment.configuration.ApplicationConfig;
  24 import org.oransc.enrichment.controllers.a1e.A1eCallbacks;
  25 import org.oransc.enrichment.controllers.r1producer.ProducerCallbacks;
  26 import org.oransc.enrichment.repository.InfoJob;
  27 import org.oransc.enrichment.repository.InfoJobs;
  28 import org.oransc.enrichment.repository.InfoProducer;
  29 import org.oransc.enrichment.repository.InfoProducers;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32 import org.springframework.beans.factory.annotation.Autowired;
  33 import org.springframework.scheduling.annotation.EnableScheduling;
  34 import org.springframework.scheduling.annotation.Scheduled;
  35 import org.springframework.stereotype.Component;
  36
  37 import reactor.core.publisher.Flux;
  38 import reactor.core.publisher.Mono;
  39 import reactor.util.retry.Retry;
  40
  41 /**
  42  * Regularly checks the availability of the Info Producers
  43  */
  44 @Component
  45 @EnableScheduling
  46 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
  47 public class ProducerSupervision {
  48     private static final Logger logger = LoggerFactory.getLogger(ProducerSupervision.class);
  49
  50     private final InfoProducers infoProducers;
  51     private final InfoJobs infoJobs;
  52     private final ProducerCallbacks producerCallbacks;
  53     private final A1eCallbacks consumerCallbacks;
  54
  55     @Autowired
  56     public ProducerSupervision(ApplicationConfig applicationConfig, InfoProducers infoProducers, InfoJobs infoJobs,
  57         ProducerCallbacks producerCallbacks, A1eCallbacks consumerCallbacks) {
  58         this.infoProducers = infoProducers;
  59         this.infoJobs = infoJobs;
  60         this.producerCallbacks = producerCallbacks;
  61         this.consumerCallbacks = consumerCallbacks;
  62     }
  63
  64     @Scheduled(fixedRate = 1000 * 60 * 5)
  65     public void checkAllProducers() {
  66         logger.debug("Checking producers starting");
  67         createTask().subscribe(null, null, () -> logger.debug("Checking all Producers completed"));
  68     }
  69
  70     public Flux<InfoProducer> createTask() {
  71         return Flux.fromIterable(infoProducers.getAllProducers()) //
  72             .flatMap(this::checkOneProducer);
  73     }
  74
  75     private Mono<InfoProducer> checkOneProducer(InfoProducer producer) {
  76         return this.producerCallbacks.healthCheck(producer) //
  77             .onErrorResume(throwable -> {
  78                 handleNonRespondingProducer(throwable, producer);
  79                 return Mono.empty();
  80             })//
  81             .doOnNext(response -> handleRespondingProducer(response, producer))
  82             .flatMap(response -> checkProducerJobs(producer)) //
  83             .map(responses -> producer);
  84     }
  85
  86     private Mono<?> checkProducerJobs(InfoProducer producer) {
  87         final int MAX_CONCURRENCY = 10;
  88         return getEiJobs(producer) //
  89             .filter(infoJob -> !producer.isJobEnabled(infoJob)) //
  90             .flatMap(infoJob -> producerCallbacks.startInfoJob(producer, infoJob, Retry.max(1)), MAX_CONCURRENCY) //
  91             .collectList() //
  92             .flatMapMany(startedJobs -> consumerCallbacks.notifyJobStatus(producer.getInfoTypes())) //
  93             .collectList();
  94     }
  95
  96     private Flux<InfoJob> getEiJobs(InfoProducer producer) {
  97         return Flux.fromIterable(producer.getInfoTypes()) //
  98             .flatMap(infoType -> Flux.fromIterable(infoJobs.getJobsForType(infoType)));
  99     }
 100
 101     private void handleNonRespondingProducer(Throwable throwable, InfoProducer producer) {
 102         logger.warn("Unresponsive producer: {} exception: {}", producer.getId(), throwable.getMessage());
 103         producer.setAliveStatus(false);
 104         if (producer.isDead()) {
 105             this.infoProducers.deregisterProducer(producer);
 106         }
 107     }
 108
 109     private void handleRespondingProducer(String response, InfoProducer producer) {
 110         logger.debug("{}", response);
 111         producer.setAliveStatus(true);
 112     }
 113
 114 }