policy-agent/src/main/java/org/oransc/policyagent/tasks/RepositorySupervision.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2019 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.policyagent.tasks;
  22
  23 import org.oransc.policyagent.clients.A1Client;
  24 import org.oransc.policyagent.repository.Policies;
  25 import org.oransc.policyagent.repository.Ric;
  26 import org.oransc.policyagent.repository.Rics;
  27 import org.oransc.policyagent.repository.Ric.RicState;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30 import org.springframework.beans.factory.annotation.Autowired;
  31 import org.springframework.scheduling.annotation.EnableScheduling;
  32 import org.springframework.scheduling.annotation.Scheduled;
  33 import org.springframework.stereotype.Component;
  34 import reactor.core.publisher.Flux;
  35 import reactor.core.publisher.Mono;
  36
  37 /**
  38  * Regularly checks the exisiting rics towards the local repository to keep it consistent.
  39  */
  40 @Component
  41 @EnableScheduling
  42 public class RepositorySupervision {
  43     private static final Logger logger = LoggerFactory.getLogger(RepositorySupervision.class);
  44
  45     private final Rics rics;
  46     private final Policies policies;
  47     private final A1Client a1Client;
  48
  49     @Autowired
  50     public RepositorySupervision(Rics rics, Policies policies, A1Client a1Client) {
  51         this.rics = rics;
  52         this.policies = policies;
  53         this.a1Client = a1Client;
  54     }
  55
  56     /**
  57      * Regularly contacts all Rics to check if they are alive.
  58      */
  59     @Scheduled(fixedRate = 1000 * 60)
  60     public void checkAllRics() {
  61         logger.debug("Checking Rics starting");
  62         createTask().subscribe(this::onRicChecked, this::onError, this::onComplete);
  63
  64     }
  65
  66     private Flux<Ric> createTask() {
  67         return Flux.fromIterable(rics.getRics()) //
  68             .groupBy(ric -> ric.state()) //
  69             .flatMap(fluxGroup -> handleGroup(fluxGroup.key(), fluxGroup));
  70     }
  71
  72     private Flux<Ric> handleGroup(Ric.RicState key, Flux<Ric> fluxGroup) {
  73         logger.debug("Handling group {}", key);
  74         switch (key) {
  75             case ACTIVE:
  76                 return fluxGroup.flatMap(this::checkActive);
  77
  78             case NOT_REACHABLE:
  79                 return fluxGroup.flatMap(this::checkNotReachable);
  80
  81             default:
  82                 // If not initiated, leave it to the StartupService.
  83                 return Flux.empty();
  84         }
  85     }
  86
  87     private Mono<Ric> checkActive(Ric ric) {
  88         logger.debug("Handling active ric {}", ric);
  89         a1Client.getPolicyIdentities(ric.getConfig().baseUrl()) //
  90         .filter(policyId -> !policies.containsPolicy(policyId)) //
  91         .doOnNext(policyId -> logger.debug("Deleting policy {}", policyId))
  92         .flatMap(policyId -> a1Client.deletePolicy(ric.getConfig().baseUrl(), policyId)) //
  93         .subscribe();
  94         return Mono.just(ric);
  95     }
  96
  97     private Mono<Ric> checkNotReachable(Ric ric) {
  98         logger.debug("Handling not reachable ric {}", ric);
  99         a1Client.getPolicyIdentities(ric.getConfig().baseUrl()) //
 100         .filter(policyId -> !policies.containsPolicy(policyId)) //
 101         .doOnNext(policyId -> logger.debug("Deleting policy {}", policyId))
 102         .flatMap(policyId -> a1Client.deletePolicy(ric.getConfig().baseUrl(), policyId)) //
 103         .subscribe(null, null, () -> ric.setState(RicState.ACTIVE));
 104         return Mono.just(ric);
 105     }
 106
 107     private void onRicChecked(Ric ric) {
 108         logger.info("Ric: " + ric.name() + " checked");
 109     }
 110
 111     private void onError(Throwable t) {
 112         logger.error("Rics supervision failed", t);
 113     }
 114
 115     private void onComplete() {
 116         logger.debug("Checking Rics completed");
 117     }
 118
 119 }