2 * ========================LICENSE_START=================================
5 * Copyright (C) 2020 Nordix Foundation
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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===================================
21 package org.oransc.enrichment.tasks;
23 import java.time.Duration;
25 import org.oransc.enrichment.configuration.ApplicationConfig;
26 import org.oransc.enrichment.controllers.consumer.ConsumerCallbacks;
27 import org.oransc.enrichment.controllers.producer.ProducerCallbacks;
28 import org.oransc.enrichment.repository.EiJob;
29 import org.oransc.enrichment.repository.EiJobs;
30 import org.oransc.enrichment.repository.EiProducer;
31 import org.oransc.enrichment.repository.EiProducers;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
34 import org.springframework.beans.factory.annotation.Autowired;
35 import org.springframework.scheduling.annotation.EnableScheduling;
36 import org.springframework.scheduling.annotation.Scheduled;
37 import org.springframework.stereotype.Component;
39 import reactor.core.publisher.Flux;
40 import reactor.core.publisher.Mono;
41 import reactor.util.retry.Retry;
44 * Regularly checks the availability of the EI Producers
48 @SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally
49 public class ProducerSupervision {
50 private static final Logger logger = LoggerFactory.getLogger(ProducerSupervision.class);
52 private final EiProducers eiProducers;
53 private final EiJobs eiJobs;
54 private final ProducerCallbacks producerCallbacks;
55 private final ConsumerCallbacks consumerCallbacks;
58 public ProducerSupervision(ApplicationConfig applicationConfig, EiProducers eiProducers, EiJobs eiJobs,
59 ProducerCallbacks producerCallbacks, ConsumerCallbacks consumerCallbacks) {
60 this.eiProducers = eiProducers;
62 this.producerCallbacks = producerCallbacks;
63 this.consumerCallbacks = consumerCallbacks;
66 @Scheduled(fixedRate = 1000 * 60 * 5)
67 public void checkAllProducers() {
68 logger.debug("Checking producers starting");
69 createTask().subscribe(null, null, () -> logger.debug("Checking all Producers completed"));
72 public Flux<EiProducer> createTask() {
73 return Flux.fromIterable(eiProducers.getAllProducers()) //
74 .flatMap(this::checkOneProducer);
77 private Mono<EiProducer> checkOneProducer(EiProducer producer) {
78 return this.producerCallbacks.healthCheck(producer) //
79 .onErrorResume(throwable -> {
80 handleNonRespondingProducer(throwable, producer);
83 .doOnNext(response -> handleRespondingProducer(response, producer))
84 .flatMap(response -> checkProducerJobs(producer)) //
85 .flatMap(responses -> Mono.just(producer));
88 private Mono<?> checkProducerJobs(EiProducer producer) {
89 return getEiJobs(producer) //
90 .filter(eiJob -> !producer.isJobEnabled(eiJob)) //
91 .flatMap(eiJob -> startEiJob(producer, eiJob), 1) //
93 .flatMapMany(eiJob -> consumerCallbacks.notifyJobStatus(producer.getEiTypes())) //
97 private Mono<String> startEiJob(EiProducer producer, EiJob eiJob) {
98 Retry retrySpec = Retry.fixedDelay(1, Duration.ofSeconds(1));
99 return producerCallbacks.startEiJob(producer, eiJob, retrySpec);
102 private Flux<EiJob> getEiJobs(EiProducer producer) {
103 return Flux.fromIterable(producer.getEiTypes()) //
104 .flatMap(eiType -> Flux.fromIterable(eiJobs.getJobsForType(eiType)));
107 private void handleNonRespondingProducer(Throwable throwable, EiProducer producer) {
108 logger.warn("Unresponsive producer: {} exception: {}", producer.getId(), throwable.getMessage());
109 producer.setAliveStatus(false);
110 if (producer.isDead()) {
111 this.eiProducers.deregisterProducer(producer);
115 private void handleRespondingProducer(String response, EiProducer producer) {
116 logger.debug("{}", response);
117 producer.setAliveStatus(true);