diff options
Diffstat (limited to 'a1-policy-management/src')
5 files changed, 53 insertions, 90 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) // diff --git a/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervisionTest.java b/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervisionTest.java index 875698a8..396a4063 100644 --- a/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervisionTest.java +++ b/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSupervisionTest.java @@ -157,14 +157,10 @@ class RicSupervisionTest { doReturn(Mono.just(a1ClientMock)).when(a1ClientFactory).createA1Client(any(Ric.class)); RIC_1.setState(RicState.UNAVAILABLE); rics.put(RIC_1); - RicSupervision supervisorUnderTest = spy(createRicSupervision()); - doReturn(synchronizationTaskMock).when(supervisorUnderTest).createSynchronizationTask(); doReturn(Mono.just(RIC_1)).when(synchronizationTaskMock).synchronizeRic(any()); - supervisorUnderTest.checkAllRics(); - verify(supervisorUnderTest).checkAllRics(); verify(supervisorUnderTest).createSynchronizationTask(); verify(synchronizationTaskMock).synchronizeRic(RIC_1); diff --git a/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTaskTest.java b/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTaskTest.java index 2e5424db..0ea0a8cb 100644 --- a/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTaskTest.java +++ b/a1-policy-management/src/test/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RicSynchronizationTaskTest.java @@ -56,6 +56,7 @@ import org.onap.ccsdk.oran.a1policymanagementservice.repository.Ric.RicState; import org.onap.ccsdk.oran.a1policymanagementservice.repository.Rics; import org.onap.ccsdk.oran.a1policymanagementservice.repository.Service; import org.onap.ccsdk.oran.a1policymanagementservice.repository.Services; +import org.springframework.web.reactive.function.client.WebClientResponseException; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -148,6 +149,21 @@ class RicSynchronizationTaskTest { } @Test + void ricIdleAndErrorDeletingPoliciesAllTheTime_thenSynchronizationWithFailedRecovery() { + setUpCreationOfA1Client(); + simulateRicWithNoPolicyTypes(); + policies.put(policy1); + WebClientResponseException exception = new WebClientResponseException(404, "", null, null, null); + when(a1ClientMock.deleteAllPolicies()).thenReturn(Flux.error(exception)); + RicSynchronizationTask synchronizerUnderTest = createTask(); + ric1.setState(RicState.AVAILABLE); + synchronizerUnderTest.run(ric1); + await().untilAsserted(() -> RicState.UNAVAILABLE.equals(ric1.getState())); + assertThat(policies.size()).isZero(); + assertThat(ric1.getState()).isEqualTo(RicState.UNAVAILABLE); + } + + @Test void ricIdlePolicyTypeInRepo_thenSynchronizationWithReuseOfTypeFromRepoAndCorrectServiceNotified() { rics.put(ric1); ric1.setState(RicState.AVAILABLE); @@ -233,58 +249,6 @@ class RicSynchronizationTaskTest { assertThat(ric1.getState()).isEqualTo(RicState.AVAILABLE); } - @Test - void ricIdleAndErrorDeletingPoliciesFirstTime_thenSynchronizationWithDeletionOfPolicies() { - ric1.setState(RicState.AVAILABLE); - rics.put(ric1); - - policies.put(policy1); - - setUpCreationOfA1Client(); - simulateRicWithNoPolicyTypes(); - - when(a1ClientMock.deleteAllPolicies()) // - .thenReturn(Flux.error(new Exception("Exception"))) // - .thenReturn(Flux.just("OK")); - - RicSynchronizationTask synchronizerUnderTest = createTask(); - - ric1.setState(RicState.UNAVAILABLE); - synchronizerUnderTest.run(ric1); - await().untilAsserted(() -> RicState.AVAILABLE.equals(ric1.getState())); - - verify(a1ClientMock, times(2)).deleteAllPolicies(); - verifyNoMoreInteractions(a1ClientMock); - - assertThat(policyTypes.size()).isZero(); - assertThat(policies.size()).isZero(); - assertThat(ric1.getState()).isEqualTo(RicState.AVAILABLE); - } - - @Test - void ricIdleAndErrorDeletingPoliciesAllTheTime_thenSynchronizationWithFailedRecovery() { - setUpCreationOfA1Client(); - simulateRicWithNoPolicyTypes(); - - policies.put(policy1); - - String originalErrorMessage = "Exception"; - when(a1ClientMock.deleteAllPolicies()).thenReturn(Flux.error(new Exception(originalErrorMessage))); - - RicSynchronizationTask synchronizerUnderTest = createTask(); - - ric1.setState(RicState.AVAILABLE); - synchronizerUnderTest.run(ric1); - await().untilAsserted(() -> RicState.UNAVAILABLE.equals(ric1.getState())); - - verify(a1ClientMock, times(2)).deleteAllPolicies(); - verifyNoMoreInteractions(a1ClientMock); - - assertThat(policyTypes.size()).isZero(); - assertThat(policies.size()).isZero(); - assertThat(ric1.getState()).isEqualTo(RicState.UNAVAILABLE); - } - private void setUpCreationOfA1Client() { when(a1ClientFactoryMock.createA1Client(any(Ric.class))).thenReturn(Mono.just(a1ClientMock)); doReturn(Flux.empty()).when(a1ClientMock).deleteAllPolicies(); |