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