NONRTRIC PMS, Sporadic instability

Attempt to stablize the synch.

Issue-ID: CCSDK-3683
Signed-off-by: PatrikBuhr <patrik.buhr@est.tech>
Change-Id: Ie0f266e2eef23e91dcf6f5925a577bb930b6d9e8

3 files changed, 37 insertions, 34 deletions

diff --git a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/clients/CcsdkA1AdapterClient.java b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/clients/CcsdkA1AdapterClient.java
index cb4e2ebb..6216a4df 100644
--- a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/clients/CcsdkA1AdapterClient.java
+++ b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/clients/CcsdkA1AdapterClient.java
@@ -34,6 +34,7 @@ import lombok.Getter;
 import org.json.JSONObject;
 import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ControllerConfig;
 import org.onap.ccsdk.oran.a1policymanagementservice.configuration.RicConfig;
+import org.onap.ccsdk.oran.a1policymanagementservice.exceptions.ServiceException;
 import org.onap.ccsdk.oran.a1policymanagementservice.repository.Policy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -132,9 +133,9 @@ public class CcsdkA1AdapterClient implements A1Client {
             this.controllerConfig = controllerConfig;
             logger.debug("CcsdkA1AdapterClient for ric: {}, a1Controller: {}", ricConfig.getRicId(), controllerConfig);
         } else {
+            logger.error("Not supported protocoltype: {}", protocolType);
             throw new IllegalArgumentException("Not handeled protocolversion: " + protocolType);
         }
-
     }
 
     @Override
@@ -146,7 +147,6 @@ public class CcsdkA1AdapterClient implements A1Client {
                     .flatMapMany(A1AdapterJsonHelper::parseJsonArrayOfString) //
                     .collectList();
         }
-
     }
 
     @Override
@@ -173,7 +173,7 @@ public class CcsdkA1AdapterClient implements A1Client {
         } else if (this.protocolType == A1ProtocolType.CCSDK_A1_ADAPTER_STD_V2_0_0) {
             return StdA1ClientVersion2.extractPolicySchema(controllerResponse, policyTypeId);
         } else {
-            throw new NullPointerException("Not supported");
+            return Mono.error(new ServiceException("Not supported " + this.protocolType));
         }
     }
 
@@ -234,6 +234,7 @@ public class CcsdkA1AdapterClient implements A1Client {
         } else if (protocolType == A1ProtocolType.CCSDK_A1_ADAPTER_OSC_V1) {
             return new OscA1Client.UriBuilder(ricConfig);
         }
+        logger.error("Not supported protocoltype: {}", protocolType);
         throw new NullPointerException();
     }
 
diff --git a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervision.java b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervision.java
index e4711c98..f90d462e 100644
--- a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervision.java
+++ b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervision.java
@@ -114,11 +114,12 @@ public class RicSupervision {
 
     private Mono<RicData> checkOneRic(RicData ricData) {
         if (ricData.ric.getState() == RicState.CONSISTENCY_CHECK || ricData.ric.getState() == RicState.SYNCHRONIZING) {
+            logger.debug("Skipping check ric: {}, state: {}", ricData.ric.id(), ricData.ric.getState());
             return Mono.empty(); // Skip, already in progress
         }
         return ricData.ric.getLock().lock(LockType.EXCLUSIVE, "checkOneRic") //
                 .flatMap(lock -> synchIfUnavailable(ricData)) //
-                .doOnNext(ric -> ric.ric.setState(RicState.CONSISTENCY_CHECK)) //
+                .doOnNext(ric -> ricData.ric.setState(RicState.CONSISTENCY_CHECK)) //
                 .flatMap(x -> checkRicPolicies(ricData)) //
                 .flatMap(x -> checkRicPolicyTypes(ricData)) //
                 .doOnNext(x -> onRicCheckedOk(ricData)) //
@@ -127,6 +128,14 @@ public class RicSupervision {
                 .onErrorResume(throwable -> Mono.empty());
     }
 
+    private Mono<RicData> synchIfUnavailable(RicData ric) {
+        if (ric.ric.getState() == RicState.UNAVAILABLE) {
+            return Mono.error(new SynchNeededException(ric));
+        } else {
+            return Mono.just(ric);
+        }
+    }
+
     private Mono<RicData> onRicCheckedError(Throwable t, RicData ricData) {
         logger.debug("Ric: {} check stopped, exception: {}", ricData.ric.id(), t.getMessage());
         ricData.ric.setState(RicState.UNAVAILABLE);
@@ -149,14 +158,6 @@ public class RicSupervision {
                 .map(a1Client -> new RicData(ric, a1Client));
     }
 
-    private Mono<RicData> synchIfUnavailable(RicData ric) {
-        if (ric.ric.getState() == RicState.UNAVAILABLE) {
-            return Mono.error(new SynchNeededException(ric));
-        } else {
-            return Mono.just(ric);
-        }
-    }
-
     private Mono<RicData> checkRicPolicies(RicData ric) {
         return ric.getClient().getPolicyIdentities() //
                 .flatMap(ricP -> validateInstances(ricP, ric));
diff --git a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTask.java b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTask.java
index 94e873ed..2d282f33 100644
--- a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTask.java
+++ b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTask.java
@@ -36,6 +36,7 @@ import org.onap.ccsdk.oran.a1policymanagementservice.repository.Rics;
 import org.onap.ccsdk.oran.a1policymanagementservice.repository.Services;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.web.reactive.function.client.WebClientResponseException;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -93,17 +94,36 @@ public class RicSynchronizationTask {
         return this.a1ClientFactory.createA1Client(ric) //
                 .doOnNext(client -> ric.setState(RicState.SYNCHRONIZING)) //
                 .flatMapMany(client -> runSynchronization(ric, client)) //
-                .onErrorResume(throwable -> deleteAllPolicyInstances(ric, throwable)) //
-                .collectList() //
-                .map(notUsed -> ric) //
                 .doOnError(t -> { //
                     logger.warn("Synchronization failure for ric: {}, reason: {}", ric.id(), t.getMessage()); //
                     ric.setState(RicState.UNAVAILABLE); //
+                    deletePoliciesIfNotRecreatable(t, ric);
                 }) //
+                .collectList() //
                 .flatMap(notUsed -> onSynchronizationComplete(ric)) //
                 .onErrorResume(t -> Mono.just(ric));
     }
 
+    /**
+     * If a 4xx error is received, allpolicies are deleted. This is just to avoid
+     * cyclical receovery due to that the NearRT RIC cannot accept a previously
+     * policy.
+     */
+    private void deletePoliciesIfNotRecreatable(Throwable throwable, Ric ric) {
+        if (throwable instanceof WebClientResponseException) {
+            WebClientResponseException responseException = (WebClientResponseException) throwable;
+            if (responseException.getStatusCode().is4xxClientError()) {
+                deleteAllPoliciesInRepository(ric);
+            }
+        }
+    }
+
+    private void deleteAllPoliciesInRepository(Ric ric) {
+        for (Policy policy : policies.getForRic(ric.id())) {
+            this.policies.remove(policy);
+        }
+    }
+
     public Flux<PolicyType> synchronizePolicyTypes(Ric ric, A1Client a1Client) {
         return a1Client.getPolicyTypeIdentities() //
                 .doOnNext(x -> ric.clearSupportedPolicyTypes()) //
@@ -134,19 +154,6 @@ public class RicSynchronizationTask {
                 .map(list -> ric);
     }
 
-    private Flux<Object> deleteAllPolicyInstances(Ric ric, Throwable t) {
-        logger.warn("Recreation of policies failed for ric: {}, reason: {}", ric.id(), t.getMessage());
-        deleteAllPoliciesInRepository(ric);
-
-        Flux<PolicyType> synchronizedTypes = this.a1ClientFactory.createA1Client(ric) //
-                .flatMapMany(a1Client -> synchronizePolicyTypes(ric, a1Client));
-        Flux<?> deletePoliciesInRic = this.a1ClientFactory.createA1Client(ric) //
-                .flatMapMany(A1Client::deleteAllPolicies) //
-                .doOnComplete(() -> deleteAllPoliciesInRepository(ric));
-
-        return Flux.concat(synchronizedTypes, deletePoliciesInRic);
-    }
-
     private Mono<PolicyType> getPolicyType(String policyTypeId, A1Client a1Client) {
         if (policyTypes.contains(policyTypeId)) {
             return Mono.just(policyTypes.get(policyTypeId));
@@ -161,12 +168,6 @@ public class RicSynchronizationTask {
         return pt;
     }
 
-    private void deleteAllPoliciesInRepository(Ric ric) {
-        for (Policy policy : policies.getForRic(ric.id())) {
-            this.policies.remove(policy);
-        }
-    }
-
     private Flux<Policy> putPolicy(Policy policy, Ric ric, A1Client a1Client) {
         logger.trace("Recreating policy: {}, for ric: {}", policy.getId(), ric.getConfig().getRicId());
         return a1Client.putPolicy(policy) //