diff options
Diffstat (limited to 'a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks')
3 files changed, 86 insertions, 140 deletions
diff --git a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RefreshConfigTask.java b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RefreshConfigTask.java index e3d489b5..983e92e6 100644 --- a/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RefreshConfigTask.java +++ b/a1-policy-management/src/main/java/org/onap/ccsdk/oran/a1policymanagementservice/tasks/RefreshConfigTask.java @@ -36,11 +36,9 @@ import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ApplicationCo import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ApplicationConfig.RicConfigUpdate; import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ApplicationConfigParser; import org.onap.ccsdk.oran.a1policymanagementservice.configuration.ConfigurationFile; -import org.onap.ccsdk.oran.a1policymanagementservice.controllers.ServiceCallbacks; import org.onap.ccsdk.oran.a1policymanagementservice.repository.Policies; import org.onap.ccsdk.oran.a1policymanagementservice.repository.PolicyTypes; import org.onap.ccsdk.oran.a1policymanagementservice.repository.Ric; -import org.onap.ccsdk.oran.a1policymanagementservice.repository.Ric.RicState; import org.onap.ccsdk.oran.a1policymanagementservice.repository.Rics; import org.onap.ccsdk.oran.a1policymanagementservice.repository.Services; import org.slf4j.Logger; @@ -130,7 +128,7 @@ public class RefreshConfigTask { .flatMap(this::parseConfiguration) // .flatMap(this::updateConfig, CONCURRENCY) // .flatMap(this::handleUpdatedRicConfig) // - .doOnTerminate(() -> logger.error("Configuration refresh task is terminated")); + .doFinally(signal -> logger.error("Configuration refresh task is terminated: {}", signal)); } private Flux<Long> regularInterval() { @@ -170,40 +168,16 @@ public class RefreshConfigTask { return new RicSynchronizationTask(a1ClientFactory, policyTypes, policies, services, restClientFactory, rics); } - /** - * for an added RIC after a restart it is nesessary to get the suypported policy - * types from the RIC unless a full synchronization is wanted. - * - * @param ric the ric to get supprted types from - * @return the same ric - */ - private Mono<Ric> trySyncronizeSupportedTypes(Ric ric) { - logger.debug("Synchronizing policy types for new RIC: {}", ric.id()); - // Synchronize the policy types - ric.setState(RicState.SYNCHRONIZING); - return this.a1ClientFactory.createA1Client(ric) // - .flatMapMany(client -> synchronizationTask().synchronizePolicyTypes(ric, client)) // - .collectList() // - .map(list -> ric) // - .doOnNext(notUsed -> ric.setState(RicState.AVAILABLE)) // - .doOnError(t -> { - logger.warn("Failed to synchronize types in new RIC: {}, reason: {}", ric.id(), t.getMessage()); - ric.setState(RicState.UNAVAILABLE); // - }) // - .onErrorResume(t -> Mono.just(ric)); - } - public Mono<RicConfigUpdate.Type> handleUpdatedRicConfig(RicConfigUpdate updatedInfo) { synchronized (this.rics) { - String ricId = updatedInfo.getRicConfig().ricId(); + String ricId = updatedInfo.getRicConfig().getRicId(); RicConfigUpdate.Type event = updatedInfo.getType(); if (event == RicConfigUpdate.Type.ADDED) { logger.debug("RIC added {}", ricId); - - return trySyncronizeSupportedTypes(new Ric(updatedInfo.getRicConfig())) // - .doOnNext(this::addRic) // - .flatMap(this::notifyServicesRicAvailable) // - .flatMap(notUsed -> Mono.just(event)); + Ric ric = new Ric(updatedInfo.getRicConfig()); + this.addRic(ric); + return this.synchronizationTask().synchronizeRic(ric) // + .map(notUsed -> event); } else if (event == RicConfigUpdate.Type.REMOVED) { logger.debug("RIC removed {}", ricId); Ric ric = rics.remove(ricId); @@ -231,17 +205,6 @@ public class RefreshConfigTask { logger.debug("Added RIC: {}", ric.id()); } - private Mono<Ric> notifyServicesRicAvailable(Ric ric) { - if (ric.getState() == RicState.AVAILABLE) { - ServiceCallbacks callbacks = new ServiceCallbacks(this.restClientFactory); - return callbacks.notifyServicesRicAvailable(ric, services) // - .collectList() // - .map(list -> ric); - } else { - return Mono.just(ric); - } - } - /** * Reads the configuration from file. */ 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 8926ec16..e3edaf44 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 @@ -64,11 +64,11 @@ public class RicSupervision { private final Services services; private final AsyncRestClientFactory restClientFactory; - private static class SynchStartedException extends ServiceException { + private static class SynchNeededException extends ServiceException { private static final long serialVersionUID = 1L; - public SynchStartedException(String message) { - super(message); + public SynchNeededException(RicData ric) { + super("SynchNeededException for " + ric.ric.id()); } } @@ -106,68 +106,61 @@ public class RicSupervision { createTask().subscribe(null, null, () -> logger.debug("Checking all RICs completed")); } - private Flux<RicData> createTask() { + private Flux<Ric> createTask() { return Flux.fromIterable(rics.getRics()) // .flatMap(this::createRicData) // - .flatMap(this::checkOneRic, CONCURRENCY); + .onErrorResume(t -> Flux.empty()) // + .flatMap(this::checkOneRic, CONCURRENCY) // + .map(ricData -> ricData.ric); } private Mono<RicData> checkOneRic(RicData ricData) { - return checkRicState(ricData) // - .flatMap(x -> ricData.ric.getLock().lock(LockType.EXCLUSIVE, "checkOneRic")) // - .flatMap(notUsed -> setRicState(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 -> ricData.ric.setState(RicState.CONSISTENCY_CHECK)) // .flatMap(x -> checkRicPolicies(ricData)) // .flatMap(x -> checkRicPolicyTypes(ricData)) // .doOnNext(x -> onRicCheckedOk(ricData)) // - .doOnError(t -> onRicCheckedError(t, ricData)) // + .onErrorResume(t -> onRicCheckedError(t, ricData)) // + .doFinally(sig -> ricData.ric.getLock().unlockBlocking()) // .onErrorResume(throwable -> Mono.empty()); } - private void onRicCheckedError(Throwable t, RicData ricData) { + 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()); - if (!(t instanceof SynchStartedException)) { - // If synch is started, the synch will set the final state - ricData.ric.setState(RicState.UNAVAILABLE); + ricData.ric.setState(RicState.UNAVAILABLE); + if ((t instanceof SynchNeededException)) { + return startSynchronization(ricData); + } else { + logger.warn("RicSupervision, ric: {}, exception: {}", ricData.ric.id(), t.getMessage()); + return Mono.empty(); } - ricData.ric.getLock().unlockBlocking(); } private void onRicCheckedOk(RicData ricData) { logger.debug("Ric: {} checked OK", ricData.ric.id()); ricData.ric.setState(RicState.AVAILABLE); - ricData.ric.getLock().unlockBlocking(); - } - - @SuppressWarnings("squid:S2445") // Blocks should be synchronized on "private final" fields - private Mono<RicData> setRicState(RicData ric) { - synchronized (ric) { - if (ric.ric.getState() == RicState.CONSISTENCY_CHECK) { - logger.debug("Ric: {} is already being checked", ric.ric.getConfig().ricId()); - return Mono.empty(); - } - ric.ric.setState(RicState.CONSISTENCY_CHECK); - return Mono.just(ric); - } } private Mono<RicData> createRicData(Ric ric) { - return Mono.just(ric) // - .flatMap(aRic -> this.a1ClientFactory.createA1Client(ric)) // + return this.a1ClientFactory.createA1Client(ric) // + .doOnError(t -> logger.debug("Could not create A1 client for ric: {}, reason: {}", ric.id(), + t.getMessage())) // .map(a1Client -> new RicData(ric, a1Client)); } - private Mono<RicData> checkRicState(RicData ric) { - if (ric.ric.getState() == RicState.UNAVAILABLE) { - logger.debug("RicSupervision, starting ric: {} synchronization (state == UNAVAILABLE)", ric.ric.id()); - return startSynchronization(ric) // - .onErrorResume(t -> Mono.empty()); - } else if (ric.ric.getState() == RicState.SYNCHRONIZING || ric.ric.getState() == RicState.CONSISTENCY_CHECK) { - return Mono.empty(); - } else { - return Mono.just(ric); - } - } - private Mono<RicData> checkRicPolicies(RicData ric) { return ric.getClient().getPolicyIdentities() // .flatMap(ricP -> validateInstances(ricP, ric)); @@ -178,14 +171,14 @@ public class RicSupervision { if (ricPolicies.size() != policies.getForRic(ric.ric.id()).size()) { logger.debug("RicSupervision, starting ric: {} synchronization (noOfPolicices == {}, expected == {})", ric.ric.id(), ricPolicies.size(), policies.getForRic(ric.ric.id()).size()); - return startSynchronization(ric); + return Mono.error(new SynchNeededException(ric)); } for (String policyId : ricPolicies) { if (!policies.containsPolicy(policyId)) { logger.debug("RicSupervision, starting ric: {} synchronization (unexpected policy in RIC: {})", ric.ric.id(), policyId); - return startSynchronization(ric); + return Mono.error(new SynchNeededException(ric)); } } return Mono.just(ric); @@ -202,22 +195,23 @@ public class RicSupervision { logger.debug( "RicSupervision, starting ric: {} synchronization (unexpected numer of policy types in RIC: {}, expected: {})", ric.ric.id(), ricTypes.size(), ric.ric.getSupportedPolicyTypes().size()); - return startSynchronization(ric); + return Mono.error(new SynchNeededException(ric)); } for (String typeName : ricTypes) { if (!ric.ric.isSupportingType(typeName)) { logger.debug("RicSupervision, starting ric: {} synchronization (unexpected policy type: {})", ric.ric.id(), typeName); - return startSynchronization(ric); + return Mono.error(new SynchNeededException(ric)); } } return Mono.just(ric); } private Mono<RicData> startSynchronization(RicData ric) { + logger.debug("RicSupervision, starting ric: {} synchronization, state: {}", ric.ric.id(), ric.ric.getState()); RicSynchronizationTask synchronizationTask = createSynchronizationTask(); return synchronizationTask.synchronizeRic(ric.ric) // - .flatMap(notUsed -> Mono.error(new SynchStartedException("Syncronization started"))); + .flatMap(notUsed -> Mono.just(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 6305abf0..b3afa7cd 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,8 +36,8 @@ 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.BaseSubscriber; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.SignalType; @@ -78,61 +78,69 @@ public class RicSynchronizationTask { } public void run(Ric ric) { - logger.debug("Ric synchronization task created: {}", ric.getConfig().ricId()); + logger.debug("Ric synchronization task created: {}", ric.getConfig().getRicId()); if (ric.getState() == RicState.SYNCHRONIZING) { - logger.debug("Ric: {} is already being synchronized", ric.getConfig().ricId()); + logger.debug("Ric: {} is already being synchronized", ric.getConfig().getRicId()); return; } ric.getLock().lock(LockType.EXCLUSIVE, "RicSynchronizationTask") // .flatMap(notUsed -> synchronizeRic(ric)) // - .subscribe(new BaseSubscriber<Object>() { - - @Override - protected void hookFinally(SignalType type) { - ric.getLock().unlockBlocking(); - } - }); + .doFinally(sig -> ric.getLock().unlockBlocking()) // + .subscribe(); } public Mono<Ric> synchronizeRic(Ric ric) { - return setRicState(ric) // - .flatMap(lock -> this.a1ClientFactory.createA1Client(ric)) // + 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)); + .onErrorResume(t -> Mono.just(ric)) // + .doFinally(signal -> onFinally(signal, ric)); + } + + private void onFinally(SignalType signal, Ric ric) { + if (ric.getState().equals(RicState.SYNCHRONIZING)) { + logger.debug("Resetting ric state after failed synch, ric: {}, signal: {}", ric.id(), signal); + ric.setState(RicState.UNAVAILABLE); // + } + } + + /** + * 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()) // .flatMapMany(Flux::fromIterable) // - .doOnNext(typeId -> logger.debug("For ric: {}, handling type: {}", ric.getConfig().ricId(), typeId)) // + .doOnNext(typeId -> logger.debug("For ric: {}, handling type: {}", ric.getConfig().getRicId(), typeId)) // .flatMap(policyTypeId -> getPolicyType(policyTypeId, a1Client), CONCURRENCY_RIC) // .doOnNext(ric::addSupportedPolicyType); // } - @SuppressWarnings("squid:S2445") // Blocks should be synchronized on "private final" fields - private Mono<Ric> setRicState(Ric ric) { - synchronized (ric) { - if (ric.getState() == RicState.SYNCHRONIZING) { - logger.debug("Ric: {} is already being synchronized", ric.getConfig().ricId()); - return Mono.empty(); - } - logger.debug("Ric state set to SYNCHRONIZING: {}", ric.getConfig().ricId()); - ric.setState(RicState.SYNCHRONIZING); - return Mono.just(ric); - } - } - private Flux<Object> runSynchronization(Ric ric, A1Client a1Client) { Flux<PolicyType> synchronizedTypes = synchronizePolicyTypes(ric, a1Client); Flux<?> policiesDeletedInRic = a1Client.deleteAllPolicies(); @@ -154,19 +162,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)); @@ -181,14 +176,8 @@ 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().ricId()); + logger.trace("Recreating policy: {}, for ric: {}", policy.getId(), ric.getConfig().getRicId()); return a1Client.putPolicy(policy) // .flatMapMany(notUsed -> Flux.just(policy)); } |