When facing any kind of errors on retrieving PNF and service-instance objects from
A&AI, BBSep now logs the error and keeps the reactive stream active. Synchronized
access in tasks when there is a configuration update. Performance improvements in
polling from DMaaP.
Change-Id: I654fd1a7267f2b723cc66b0a93e4855003af2914
Issue-ID: DCAEGEN2-1354
Signed-off-by: Stavros Kanarakis <stavros.kanarakis@nokia.com>
application_cpeAuth_policy_scope: policyScopeCpeAuth
application_cpeAuth_cl_control_name: clControlNameCpeAuth
application_cbs_polling_interval_sec: 120
-dmaap_consumer_id: c12
+dmaap_consumer_id: c12
dmaap_consumer_group: OpenDcae-c12
imports:
- "http://www.getcloudify.org/spec/cloudify/3.4/types.yaml"
- - {{ ONAPTEMPLATE_RAWREPOURL_org_onap_dcaegen2_platform_plugins_releases }}/k8splugin/1.4.5/k8splugin_types.yaml
+ - https://nexus.onap.org/service/local/repositories/raw/content/org.onap.dcaegen2.platform.plugins/R3/k8splugin/1.4.4/k8splugin_types.yaml
inputs:
aai_enrichment_host:
cpe_authentication_url:
type: string
close_loop_url:
- type: string
+ type: string
+ application_policy_version:
+ description: Policy version value for building CL events
+ type: string
+ default: "1.0.0.5"
+ application_cl_target_type:
+ description: Close Loop target type value for building CL events
+ type: string
+ default: "VM"
+ application_cl_event_status:
+ description: Close Loop event status value for building CL events
+ type: string
+ default: "ONSET"
+ application_cl_version:
+ description: Close Loop version value for building CL events
+ type: string
+ default: "1.0.2"
+ application_cl_target:
+ description: Close Loop target value for building CL events
+ type: string
+ default: "vserver.vserver-name"
+ application_cl_originator:
+ description: Close Loop originator value for building CL events
+ type: string
+ default: "DCAE-BBS-ep"
application_rereg_policy_scope:
description: Policy Scope value for building PNF relocation CL event
type: string
application.pipelinesPollingIntervalSec: 30
application.pipelinesTimeoutSec: 15
application.cbsPollingIntervalSec: { get_input: application_cbs_polling_interval_sec }
+ application.policyVersion: { get_input: application_policy_version }
+ application.clTargetType: { get_input: application_cl_target_type }
+ application.clEventStatus: { get_input: application_cl_event_status }
+ application.clVersion: { get_input: application_cl_version }
+ application.clTarget: { get_input: application_cl_target }
+ application.clOriginator: { get_input: application_cl_originator }
application.reregistration.policyScope: { get_input: application_rereg_policy_scope }
application.reregistration.clControlName: { get_input: application_rereg_cl_control_name }
application.cpe.authentication.policyScope: { get_input: application_cpeAuth_policy_scope }
log_directory: "/opt/app/bbs-event-processor/logs"
tls_info:
cert_directory: '/opt/app/bbs-event-processor/etc/cert/'
- use_tls: false
+ use_tls: false
\ No newline at end of file
"designer_editable": true,
"policy_editable": false,
"sourced_at_deployment": false,
- "description": "DmaaP protocol used for any DMaaP interaction"
+ "description": "DMaaP protocol used for any DMaaP interaction"
},
{
"name": "dmaap.contentType",
"designer_editable": true,
"policy_editable": false,
"sourced_at_deployment": false,
- "description": "DmaaP content type"
+ "description": "DMaaP content type"
},
{
"name": "dmaap.consumer.consumerId",
"designer_editable": true,
"policy_editable": false,
"sourced_at_deployment": true,
- "description": "DmaaP consumer consumer ID"
+ "description": "DMaaP consumer consumer ID"
},
{
"name": "dmaap.consumer.consumerGroup",
"designer_editable": true,
"policy_editable": false,
"sourced_at_deployment": true,
- "description": "DmaaP consumer consumer group"
+ "description": "DMaaP consumer consumer group"
},
{
"name": "dmaap.messageLimit",
"designer_editable": true,
"policy_editable": false,
"sourced_at_deployment": false,
- "description": "DmaaP message limit"
+ "description": "DMaaP message limit"
},
{
"name": "dmaap.timeoutMs",
"designer_editable": true,
"policy_editable": false,
"sourced_at_deployment": false,
- "description": "DmaaP timeout in millis"
+ "description": "DMaaP timeout in millis"
},
{
"name": "aai.host",
"value": 300,
"constraints": [
{
- "greater_or_equal": 20
+ "greater_or_equal": 30
}
],
"designer_editable": true,
"sourced_at_deployment": true,
"description": "Polling interval in seconds for fetching configuration from Consul via CBS service"
},
+ {
+ "name": "application.policyVersion",
+ "value": "1.0.0.5",
+ "designer_editable": false,
+ "policy_editable": true,
+ "sourced_at_deployment": true,
+ "description": "Policy Version parameter in Policy triggering event to be published"
+ },
+ {
+ "name": "application.clTargetType",
+ "value": "VM",
+ "designer_editable": false,
+ "policy_editable": true,
+ "sourced_at_deployment": true,
+ "description": "Close Loop target type parameter in Policy triggering event to be published"
+ },
+ {
+ "name": "application.clEventStatus",
+ "value": "ONSET",
+ "designer_editable": false,
+ "policy_editable": true,
+ "sourced_at_deployment": true,
+ "description": "Close Loop event status parameter in Policy triggering event to be published"
+ },
+ {
+ "name": "application.clVersion",
+ "value": "1.0.2",
+ "designer_editable": false,
+ "policy_editable": true,
+ "sourced_at_deployment": true,
+ "description": "Close Loop version parameter in Policy triggering event to be published"
+ },
+ {
+ "name": "application.clTarget",
+ "value": "vserver.vserver-name",
+ "designer_editable": false,
+ "policy_editable": true,
+ "sourced_at_deployment": true,
+ "description": "Close Loop target parameter in Policy triggering event to be published"
+ },
+ {
+ "name": "application.clOriginator",
+ "value": "DCAE-BBS-ep",
+ "designer_editable": false,
+ "policy_editable": true,
+ "sourced_at_deployment": true,
+ "description": "Close Loop originator parameter in Policy triggering event to be published"
+ },
{
"name": "application.reregistration.policyScope",
"value": "policyScope",
"designer_editable": false,
"policy_editable": true,
"sourced_at_deployment": true,
- "description": "Hard-coded value for Policy Scope parameter in Policy triggering event to be published"
+ "description": "Policy Scope parameter in Policy triggering event to be published"
},
{
"name": "application.reregistration.clControlName",
"designer_editable": false,
"policy_editable": true,
"sourced_at_deployment": true,
- "description": "Hard-coded value for CL control name parameter in Policy triggering event to be published"
+ "description": "Close Loop control name parameter in Policy triggering event to be published"
},
{
"name": "application.cpe.authentication.policyScope",
"designer_editable": false,
"policy_editable": true,
"sourced_at_deployment": true,
- "description": "Hard-coded value for CL control nam parameter in Policy triggering event to be published"
+ "description": "Policy Scope parameter in Policy triggering event to be published"
},
{
"name": "application.cpe.authentication.clControlName",
"designer_editable": false,
"policy_editable": true,
"sourced_at_deployment": true,
- "description": "Hard-coded value for CL control nam parameter in Policy triggering event to be published"
+ "description": "Close Loop control nam parameter in Policy triggering event to be published"
},
{
"name": "application.reregistration.configKey",
@Override
public synchronized void notifyObservers() {
- observers.forEach(o -> o.updateConfiguration(this));
+ observers.forEach(ConfigurationChangeObserver::updateConfiguration);
}
- public DmaapConsumerConfiguration getDmaapReRegistrationConsumerConfiguration() {
+ public synchronized DmaapConsumerConfiguration getDmaapReRegistrationConsumerConfiguration() {
return dmaapReRegistrationConsumerConfiguration;
}
- public DmaapConsumerConfiguration getDmaapCpeAuthenticationConsumerConfiguration() {
+ public synchronized DmaapConsumerConfiguration getDmaapCpeAuthenticationConsumerConfiguration() {
return dmaapCpeAuthenticationConsumerConfiguration;
}
- public DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
+ public synchronized DmaapPublisherConfiguration getDmaapPublisherConfiguration() {
return dmaapPublisherConfiguration;
}
- public AaiClientConfiguration getAaiClientConfiguration() {
+ public synchronized AaiClientConfiguration getAaiClientConfiguration() {
return aaiClientConfiguration;
}
- public int getPipelinesPollingIntervalInSeconds() {
+ public synchronized int getPipelinesPollingIntervalInSeconds() {
return genericProperties.getPipelinesPollingIntervalSec();
}
- public int getPipelinesTimeoutInSeconds() {
+ public synchronized int getPipelinesTimeoutInSeconds() {
return genericProperties.getPipelinesTimeoutSec();
}
- public int getCbsPollingInterval() {
- return cbsPollingInterval;
+ public synchronized String getPolicyVersion() {
+ return genericProperties.getPolicyVersion();
+ }
+
+ public synchronized String getCloseLoopTargetType() {
+ return genericProperties.getClTargetType();
+ }
+
+ public synchronized String getCloseLoopEventStatus() {
+ return genericProperties.getClEventStatus();
+ }
+
+ public synchronized String getCloseLoopVersion() {
+ return genericProperties.getClVersion();
+ }
+
+ public synchronized String getCloseLoopTarget() {
+ return genericProperties.getClTarget();
}
- public String getReRegistrationCloseLoopPolicyScope() {
+ public String getCloseLoopOriginator() {
+ return genericProperties.getClOriginator();
+ }
+ public synchronized int getCbsPollingInterval() {
+ return cbsPollingInterval;
+ }
+
+ public synchronized String getReRegistrationCloseLoopPolicyScope() {
return genericProperties.getReRegistration().getPolicyScope();
}
- public String getReRegistrationCloseLoopControlName() {
+ public synchronized String getReRegistrationCloseLoopControlName() {
return genericProperties.getReRegistration().getClControlName();
}
- public String getCpeAuthenticationCloseLoopPolicyScope() {
+ public synchronized String getCpeAuthenticationCloseLoopPolicyScope() {
return genericProperties.getCpeAuthentication().getPolicyScope();
}
- public String getCpeAuthenticationCloseLoopControlName() {
+ public synchronized String getCpeAuthenticationCloseLoopControlName() {
return genericProperties.getCpeAuthentication().getClControlName();
}
*/
public void updateCurrentConfiguration(GeneratedAppConfigObject newConfiguration) {
- cbsPollingInterval = newConfiguration.cbsPollingIntervalSec();
-
- GeneratedAppConfigObject.StreamsObject reRegObject = getStreamsObject(newConfiguration.streamSubscribesMap(),
- newConfiguration.reRegConfigKey(), "PNF Re-Registration");
- TopicUrlInfo topicUrlInfo = parseTopicUrl(reRegObject.dmaapInfo().topicUrl());
- dmaapReRegistrationConsumerProperties.setDmaapHostName(topicUrlInfo.getHost());
- dmaapReRegistrationConsumerProperties.setDmaapPortNumber(topicUrlInfo.getPort());
- dmaapReRegistrationConsumerProperties.setDmaapProtocol(newConfiguration.dmaapProtocol());
- dmaapReRegistrationConsumerProperties.setDmaapContentType(newConfiguration.dmaapContentType());
- dmaapReRegistrationConsumerProperties.setDmaapTopicName(topicUrlInfo.getTopicName());
- dmaapReRegistrationConsumerProperties.setConsumerId(newConfiguration.dmaapConsumerConsumerId());
- dmaapReRegistrationConsumerProperties.setConsumerGroup(newConfiguration.dmaapConsumerConsumerGroup());
- dmaapReRegistrationConsumerProperties.setMessageLimit(newConfiguration.dmaapMessageLimit());
- dmaapReRegistrationConsumerProperties.setTimeoutMs(newConfiguration.dmaapTimeoutMs());
- constructDmaapReRegistrationConfiguration();
-
- GeneratedAppConfigObject.StreamsObject cpeAuthObject = getStreamsObject(newConfiguration.streamSubscribesMap(),
- newConfiguration.cpeAuthConfigKey(), "CPE Authentication");
- topicUrlInfo = parseTopicUrl(cpeAuthObject.dmaapInfo().topicUrl());
- dmaapCpeAuthenticationConsumerProperties.setDmaapHostName(topicUrlInfo.getHost());
- dmaapCpeAuthenticationConsumerProperties.setDmaapPortNumber(topicUrlInfo.getPort());
- dmaapCpeAuthenticationConsumerProperties.setDmaapProtocol(newConfiguration.dmaapProtocol());
- dmaapCpeAuthenticationConsumerProperties.setDmaapContentType(newConfiguration.dmaapContentType());
- dmaapCpeAuthenticationConsumerProperties.setDmaapTopicName(topicUrlInfo.getTopicName());
- dmaapCpeAuthenticationConsumerProperties.setConsumerId(newConfiguration.dmaapConsumerConsumerId());
- dmaapCpeAuthenticationConsumerProperties.setConsumerGroup(newConfiguration.dmaapConsumerConsumerGroup());
- dmaapCpeAuthenticationConsumerProperties.setMessageLimit(newConfiguration.dmaapMessageLimit());
- dmaapCpeAuthenticationConsumerProperties.setTimeoutMs(newConfiguration.dmaapTimeoutMs());
- constructDmaapCpeAuthenticationConfiguration();
-
- GeneratedAppConfigObject.StreamsObject closeLoopObject = getStreamsObject(newConfiguration.streamPublishesMap(),
- newConfiguration.closeLoopConfigKey(), "Close Loop");
- topicUrlInfo = parseTopicUrl(closeLoopObject.dmaapInfo().topicUrl());
- dmaapProducerProperties.setDmaapHostName(topicUrlInfo.getHost());
- dmaapProducerProperties.setDmaapPortNumber(topicUrlInfo.getPort());
- dmaapProducerProperties.setDmaapProtocol(newConfiguration.dmaapProtocol());
- dmaapProducerProperties.setDmaapContentType(newConfiguration.dmaapContentType());
- dmaapProducerProperties.setDmaapTopicName(topicUrlInfo.getTopicName());
- constructDmaapProducerConfiguration();
-
- aaiClientProperties.setAaiHost(newConfiguration.aaiHost());
- aaiClientProperties.setAaiPort(newConfiguration.aaiPort());
- aaiClientProperties.setAaiProtocol(newConfiguration.aaiProtocol());
- aaiClientProperties.setAaiUserName(newConfiguration.aaiUsername());
- aaiClientProperties.setAaiUserPassword(newConfiguration.aaiPassword());
- aaiClientProperties.setAaiIgnoreSslCertificateErrors(newConfiguration.aaiIgnoreSslCertificateErrors());
- constructAaiConfiguration();
-
-
- genericProperties.setPipelinesPollingIntervalSec(newConfiguration.pipelinesPollingIntervalSec());
- genericProperties.setPipelinesTimeoutSec(newConfiguration.pipelinesTimeoutSec());
- genericProperties.getReRegistration().setPolicyScope(newConfiguration.reRegistrationPolicyScope());
- genericProperties.getReRegistration().setClControlName(newConfiguration.reRegistrationClControlName());
- genericProperties.getCpeAuthentication().setPolicyScope(newConfiguration.cpeAuthPolicyScope());
- genericProperties.getCpeAuthentication().setClControlName(newConfiguration.cpeAuthClControlName());
+ synchronized (this) {
+ cbsPollingInterval = newConfiguration.cbsPollingIntervalSec();
+
+ GeneratedAppConfigObject.StreamsObject reRegObject =
+ getStreamsObject(newConfiguration.streamSubscribesMap(), newConfiguration.reRegConfigKey(),
+ "PNF Re-Registration");
+ TopicUrlInfo topicUrlInfo = parseTopicUrl(reRegObject.dmaapInfo().topicUrl());
+ dmaapReRegistrationConsumerProperties.setDmaapHostName(topicUrlInfo.getHost());
+ dmaapReRegistrationConsumerProperties.setDmaapPortNumber(topicUrlInfo.getPort());
+ dmaapReRegistrationConsumerProperties.setDmaapProtocol(newConfiguration.dmaapProtocol());
+ dmaapReRegistrationConsumerProperties.setDmaapContentType(newConfiguration.dmaapContentType());
+ dmaapReRegistrationConsumerProperties.setDmaapTopicName(topicUrlInfo.getTopicName());
+ dmaapReRegistrationConsumerProperties.setConsumerId(newConfiguration.dmaapConsumerConsumerId());
+ dmaapReRegistrationConsumerProperties.setConsumerGroup(newConfiguration.dmaapConsumerConsumerGroup());
+ dmaapReRegistrationConsumerProperties.setMessageLimit(newConfiguration.dmaapMessageLimit());
+ dmaapReRegistrationConsumerProperties.setTimeoutMs(newConfiguration.dmaapTimeoutMs());
+ constructDmaapReRegistrationConfiguration();
+
+ GeneratedAppConfigObject.StreamsObject cpeAuthObject =
+ getStreamsObject(newConfiguration.streamSubscribesMap(), newConfiguration.cpeAuthConfigKey(),
+ "CPE Authentication");
+ topicUrlInfo = parseTopicUrl(cpeAuthObject.dmaapInfo().topicUrl());
+ dmaapCpeAuthenticationConsumerProperties.setDmaapHostName(topicUrlInfo.getHost());
+ dmaapCpeAuthenticationConsumerProperties.setDmaapPortNumber(topicUrlInfo.getPort());
+ dmaapCpeAuthenticationConsumerProperties.setDmaapProtocol(newConfiguration.dmaapProtocol());
+ dmaapCpeAuthenticationConsumerProperties.setDmaapContentType(newConfiguration.dmaapContentType());
+ dmaapCpeAuthenticationConsumerProperties.setDmaapTopicName(topicUrlInfo.getTopicName());
+ dmaapCpeAuthenticationConsumerProperties.setConsumerId(newConfiguration.dmaapConsumerConsumerId());
+ dmaapCpeAuthenticationConsumerProperties.setConsumerGroup(newConfiguration.dmaapConsumerConsumerGroup());
+ dmaapCpeAuthenticationConsumerProperties.setMessageLimit(newConfiguration.dmaapMessageLimit());
+ dmaapCpeAuthenticationConsumerProperties.setTimeoutMs(newConfiguration.dmaapTimeoutMs());
+ constructDmaapCpeAuthenticationConfiguration();
+
+ GeneratedAppConfigObject.StreamsObject closeLoopObject =
+ getStreamsObject(newConfiguration.streamPublishesMap(), newConfiguration.closeLoopConfigKey(),
+ "Close Loop");
+ topicUrlInfo = parseTopicUrl(closeLoopObject.dmaapInfo().topicUrl());
+ dmaapProducerProperties.setDmaapHostName(topicUrlInfo.getHost());
+ dmaapProducerProperties.setDmaapPortNumber(topicUrlInfo.getPort());
+ dmaapProducerProperties.setDmaapProtocol(newConfiguration.dmaapProtocol());
+ dmaapProducerProperties.setDmaapContentType(newConfiguration.dmaapContentType());
+ dmaapProducerProperties.setDmaapTopicName(topicUrlInfo.getTopicName());
+ constructDmaapProducerConfiguration();
+
+ aaiClientProperties.setAaiHost(newConfiguration.aaiHost());
+ aaiClientProperties.setAaiPort(newConfiguration.aaiPort());
+ aaiClientProperties.setAaiProtocol(newConfiguration.aaiProtocol());
+ aaiClientProperties.setAaiUserName(newConfiguration.aaiUsername());
+ aaiClientProperties.setAaiUserPassword(newConfiguration.aaiPassword());
+ aaiClientProperties.setAaiIgnoreSslCertificateErrors(newConfiguration.aaiIgnoreSslCertificateErrors());
+ constructAaiConfiguration();
+
+ genericProperties.setPipelinesPollingIntervalSec(newConfiguration.pipelinesPollingIntervalSec());
+ genericProperties.setPipelinesTimeoutSec(newConfiguration.pipelinesTimeoutSec());
+ genericProperties.setPolicyVersion(newConfiguration.policyVersion());
+ genericProperties.setClTargetType(newConfiguration.closeLoopTargetType());
+ genericProperties.setClEventStatus(newConfiguration.closeLoopEventStatus());
+ genericProperties.setClVersion(newConfiguration.closeLoopVersion());
+ genericProperties.setClTarget(newConfiguration.closeLoopTarget());
+ genericProperties.setClOriginator(newConfiguration.closeLoopOriginator());
+ genericProperties.getReRegistration().setPolicyScope(newConfiguration.reRegistrationPolicyScope());
+ genericProperties.getReRegistration().setClControlName(newConfiguration.reRegistrationClControlName());
+ genericProperties.getCpeAuthentication().setPolicyScope(newConfiguration.cpeAuthPolicyScope());
+ genericProperties.getCpeAuthentication().setClControlName(newConfiguration.cpeAuthClControlName());
+ }
notifyObservers();
}
private GeneratedAppConfigObject.StreamsObject getStreamsObject(
Map<String, GeneratedAppConfigObject.StreamsObject> map, String configKey, String messageName) {
GeneratedAppConfigObject.StreamsObject streamsObject = map.get(configKey);
- if (!streamsObject.type().equals(STREAMS_TYPE)) {
+ if (!STREAMS_TYPE.equals(streamsObject.type())) {
throw new ApplicationEnvironmentException(String.format("%s requires information about"
+ " message-router topic in ONAP", messageName));
}
public static final String RETRIEVE_PNF_TASK_NAME = "PNF Retrieval";
public static final String RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME = "HSI CFS Service Instance Retrieval";
- public static final String STREAMS_TYPE = "message_router";
+ static final String STREAMS_TYPE = "message_router";
public static final String IN_SERVICE_NAME_IN_ONAP = "inService";
public static final String OUT_OF_SERVICE_NAME_IN_ONAP = "outOfService";
+ // Close Loop Constants
+ public static final String DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE = "DCAE.BBS_event_processor_mSInstance";
+
private ApplicationConstants() {}
}
\ No newline at end of file
public interface ConfigurationChangeObserver {
/**
- * Take actions upon new application configuration.
- * @param configuration new application configuration (complete configuration)
+ * Take actions upon updates in application configuration.
*/
- void updateConfiguration(ApplicationConfiguration configuration);
+ void updateConfiguration();
}
getMissingEnvironmentVariables()));
}
- RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
-
- // Necessary properties from the environment (Consul host:port, service-name (hostname), CBS name)
- EnvProperties env = EnvProperties.fromEnvironment();
-
- // Create the client and use it to get the configuration
- cbsFetchPipeline = CbsClientFactory.createCbsClient(env)
- .doOnError(e -> LOGGER.warn("CBS Configuration fetch failed with error: {}", e))
- .retry(e -> true)
- .flatMapMany(cbsClient -> cbsClient.updates(diagnosticContext, initialDelay, period))
- .subscribe(this::parseConsulRetrievedConfiguration, this::handleErrors);
+ fetchConfig(initialDelay, period);
}
boolean environmentNotReady() {
private void handleErrors(Throwable throwable) {
LOGGER.error("Periodic CBS configuration polling was terminated with error: {}", throwable);
+ LOGGER.info("Will restart CBS configuration fetching job due to abnormal termination."
+ + " Will start fetching after 60 seconds (please correct configuration in the meantime)"
+ + " and it will then poll every {} seconds (reverting to default)",
+ configuration.getCbsPollingInterval());
+ fetchConfig(Duration.ofSeconds(60), Duration.ofSeconds(configuration.getCbsPollingInterval()));
+ }
+
+ private void fetchConfig(Duration initialDelay, Duration period) {
+ RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
+
+ // Necessary properties from the environment (Consul host:port, service-name (hostname), CBS name)
+ EnvProperties env = EnvProperties.fromEnvironment();
+
+ // Create the client and use it to get the configuration
+ cbsFetchPipeline = CbsClientFactory.createCbsClient(env)
+ .doOnError(e -> LOGGER.warn("CBS Configuration fetch failed with error: {}", e))
+ .retry(e -> true)
+ .flatMapMany(cbsClient -> cbsClient.updates(diagnosticContext, initialDelay, period))
+ .subscribe(this::parseConsulRetrievedConfiguration, this::handleErrors);
}
@NotNull
final int pipelinesPollingIntervalSec = configObject.get("application.pipelinesPollingIntervalSec").getAsInt();
final int pipelinesTimeoutSec = configObject.get("application.pipelinesTimeoutSec").getAsInt();
final int cbsPollingIntervalSec = configObject.get("application.cbsPollingIntervalSec").getAsInt();
+
final String reRegPolicyScope = configObject.get("application.reregistration.policyScope").getAsString();
final String reRegClControlName = configObject.get("application.reregistration.clControlName").getAsString();
final String cpeAuthPolicyScope = configObject.get("application.cpe.authentication.policyScope").getAsString();
final String cpeAuthClControlName =
configObject.get("application.cpe.authentication.clControlName").getAsString();
+
+ final String policyVersion = configObject.get("application.policyVersion").getAsString();
+ final String closeLoopTargetType = configObject.get("application.clTargetType").getAsString();
+ final String closeLoopEventStatus = configObject.get("application.clEventStatus").getAsString();
+ final String closeLoopVersion = configObject.get("application.clVersion").getAsString();
+ final String closeLoopTarget = configObject.get("application.clTarget").getAsString();
+ final String closeLoopOriginator = configObject.get("application.clOriginator").getAsString();
+
final String reRegConfigKey = configObject.get("application.reregistration.configKey").getAsString();
final String cpeAuthConfigKey = configObject.get("application.cpeAuth.configKey").getAsString();
final String closeLoopConfigKey = configObject.get("application.closeLoop.configKey").getAsString();
.cbsPollingIntervalSec(cbsPollingIntervalSec)
.reRegistrationPolicyScope(reRegPolicyScope)
.reRegistrationClControlName(reRegClControlName)
+ .policyVersion(policyVersion)
+ .closeLoopTargetType(closeLoopTargetType)
+ .closeLoopEventStatus(closeLoopEventStatus)
+ .closeLoopVersion(closeLoopVersion)
+ .closeLoopTarget(closeLoopTarget)
+ .closeLoopOriginator(closeLoopOriginator)
.cpeAuthPolicyScope(cpeAuthPolicyScope)
.cpeAuthClControlName(cpeAuthClControlName)
.reRegConfigKey(reRegConfigKey)
private int pipelinesTimeoutSec;
+ @NotBlank
+ private String policyVersion;
+
+ @NotBlank
+ private String clTargetType;
+
+ @NotBlank
+ private String clEventStatus;
+
+ @NotBlank
+ private String clVersion;
+
+ @NotBlank
+ private String clTarget;
+
+ @NotBlank
+ private String clOriginator;
+
private ReRegistrationGenericProperties reRegistration;
private CpeAuthenticationGenericProperties cpeAuthentication;
@SerializedName(value = "application.pipelinesTimeoutSec", alternate = "application.pipelinesTimeoutSec")
int pipelinesTimeoutSec();
- @SerializedName(value = "application.cbsPollingIntervalSec",
- alternate = "application.cbsPollingIntervalSec")
+ @SerializedName(value = "application.cbsPollingIntervalSec", alternate = "application.cbsPollingIntervalSec")
int cbsPollingIntervalSec();
@SerializedName(value = "application.reregistration.policyScope",
alternate = "application.reregistration.clControlName")
String cpeAuthClControlName();
+ @SerializedName(value = "application.policyVersion", alternate = "application.policyVersion")
+ String policyVersion();
+
+ @SerializedName(value = "application.clTargetType", alternate = "application.clTargetType")
+ String closeLoopTargetType();
+
+ @SerializedName(value = "application.clEventStatus", alternate = "application.clEventStatus")
+ String closeLoopEventStatus();
+
+ @SerializedName(value = "application.clVersion", alternate = "application.clVersion")
+ String closeLoopVersion();
+
+ @SerializedName(value = "application.clTarget", alternate = "application.clTarget")
+ String closeLoopTarget();
+
+ @SerializedName(value = "application.clOriginator", alternate = "application.clOriginator")
+ String closeLoopOriginator();
+
@SerializedName(value = "application.reregistration.configKey", alternate = "application.reregistration.configKey")
String reRegConfigKey();
package org.onap.bbs.event.processor.pipelines;
import static org.onap.bbs.event.processor.config.ApplicationConstants.CONSUME_CPE_AUTHENTICATION_TASK_NAME;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE;
import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME;
import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_PNF_TASK_NAME;
import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INSTANCE_UUID;
import javax.net.ssl.SSLException;
import org.onap.bbs.event.processor.config.ApplicationConfiguration;
-import org.onap.bbs.event.processor.exceptions.AaiTaskException;
-import org.onap.bbs.event.processor.exceptions.DmaapException;
import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
import org.onap.bbs.event.processor.model.CpeAuthenticationConsumerDmaapModel;
private static final Logger LOGGER = LoggerFactory.getLogger(CpeAuthenticationPipeline.class);
- private static final String DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE = "DCAE.BBS_event_processor_mSInstance";
- private static final String POLICY_VERSION = "1.0.0.5";
private static final String POLICY_NAME = "CPE_Authentication";
- private static final String CLOSE_LOOP_TARGET_TYPE = "VM";
- private static final String CLOSED_LOOP_EVENT_STATUS = "ONSET";
- private static final String CLOSE_LOOP_VERSION = "1.0.2";
- private static final String CLOSE_LOOP_TARGET = "vserver.vserver-name";
- private static final String FROM = "DCAE";
private DmaapCpeAuthenticationConsumerTask consumerTask;
private DmaapPublisherTask publisherTask;
if (e instanceof TimeoutException) {
LOGGER.warn("Timed out waiting for DMaaP response");
} else if (e instanceof EmptyDmaapResponseException) {
- LOGGER.warn("Nothing to consume from DMaaP");
+ LOGGER.info("Nothing to consume from DMaaP");
+ } else {
+ LOGGER.error("DMaaP Consumer error: {}", e.getMessage());
}
})
.onErrorResume(
- e -> (e instanceof EmptyDmaapResponseException || e instanceof TimeoutException),
+ e -> e instanceof Exception,
e -> Mono.empty())
.map(event -> {
// For each message, we have to keep separate state. This state will be enhanced
e.getMessage())
)
.onErrorResume(
- e -> e instanceof AaiTaskException || e instanceof TimeoutException,
+ e -> e instanceof Exception,
e -> Mono.empty())
.map(p -> {
state.setPnfAaiObject(p);
// towards the HSI CFS service
String serviceInstanceId = pnf.getRelationshipListAaiObject().getRelationshipEntries()
.stream()
- .filter(e -> e.getRelatedTo().equals("service-instance"))
+ .filter(e -> "service-instance".equals(e.getRelatedTo()))
.flatMap(e -> e.getRelationshipData().stream())
- .filter(d -> d.getRelationshipKey().equals("service-instance.service-instance-id"))
+ .filter(d -> "service-instance.service-instance-id".equals(d.getRelationshipKey()))
.map(RelationshipListAaiObject.RelationshipDataEntryAaiObject::getRelationshipValue)
.findFirst().orElse("");
e.getMessage())
)
.onErrorResume(
- e -> e instanceof AaiTaskException || e instanceof TimeoutException,
+ e -> e instanceof Exception,
e -> Mono.empty())
.map(s -> {
state.setHsiCfsServiceInstance(s);
)
.doOnError(e -> LOGGER.error("Error while triggering Policy: {}", e.getMessage()))
.onErrorResume(
- e -> e instanceof DmaapException || e instanceof TimeoutException,
+ e -> e instanceof Exception,
e -> Mono.empty());
}
return optionalMetadata
.map(list -> list.getMetadataEntries()
.stream()
- .anyMatch(m -> m.getMetaname().equals("rgw-mac-address")
+ .anyMatch(m -> "rgw-mac-address".equals(m.getMetaname())
&& m.getMetavalue().equals(eventRgwMacAddress)))
.orElse(false);
}
ControlLoopPublisherDmaapModel triggerEvent = ImmutableControlLoopPublisherDmaapModel.builder()
.closedLoopEventClient(DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE)
- .policyVersion(POLICY_VERSION)
+ .policyVersion(configuration.getPolicyVersion())
.policyName(POLICY_NAME)
.policyScope(configuration.getCpeAuthenticationCloseLoopPolicyScope())
- .targetType(CLOSE_LOOP_TARGET_TYPE)
+ .targetType(configuration.getCloseLoopTargetType())
.aaiEnrichmentData(enrichmentData)
.closedLoopAlarmStart(Instant.now().getEpochSecond())
- .closedLoopEventStatus(CLOSED_LOOP_EVENT_STATUS)
+ .closedLoopEventStatus(configuration.getCloseLoopEventStatus())
.closedLoopControlName(configuration.getCpeAuthenticationCloseLoopControlName())
- .version(CLOSE_LOOP_VERSION)
- .target(CLOSE_LOOP_TARGET)
+ .version(configuration.getCloseLoopVersion())
+ .target(configuration.getCloseLoopTarget())
.requestId(UUID.randomUUID().toString())
- .originator(FROM)
+ .originator(configuration.getCloseLoopOriginator())
.build();
LOGGER.debug("Processing Step: Publish for Policy");
LOGGER.trace("Trigger Policy event: ({})",triggerEvent);
package org.onap.bbs.event.processor.pipelines;
import static org.onap.bbs.event.processor.config.ApplicationConstants.CONSUME_REREGISTRATION_TASK_NAME;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE;
import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME;
import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_PNF_TASK_NAME;
import static org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables.INSTANCE_UUID;
import javax.net.ssl.SSLException;
import org.onap.bbs.event.processor.config.ApplicationConfiguration;
-import org.onap.bbs.event.processor.exceptions.AaiTaskException;
-import org.onap.bbs.event.processor.exceptions.DmaapException;
import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
import org.onap.bbs.event.processor.model.ImmutableControlLoopPublisherDmaapModel;
private static final Logger LOGGER = LoggerFactory.getLogger(ReRegistrationPipeline.class);
- private static final String DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE = "DCAE.BBS_event_processor_mSInstance";
- private static final String POLICY_VERSION = "1.0.0.5";
private static final String POLICY_NAME = "Nomadic_ONT";
- private static final String CLOSE_LOOP_TARGET_TYPE = "VM";
- private static final String CLOSED_LOOP_EVENT_STATUS = "ONSET";
- private static final String CLOSE_LOOP_VERSION = "1.0.2";
- private static final String CLOSE_LOOP_TARGET = "vserver.vserver-name";
- private static final String FROM = "DCAE";
private DmaapReRegistrationConsumerTask consumerTask;
private DmaapPublisherTask publisherTask;
if (e instanceof TimeoutException) {
LOGGER.warn("Timed out waiting for DMaaP response");
} else if (e instanceof EmptyDmaapResponseException) {
- LOGGER.warn("Nothing to consume from DMaaP");
+ LOGGER.info("Nothing to consume from DMaaP");
+ } else {
+ LOGGER.error("DMaaP Consumer error: {}", e.getMessage());
}
})
.onErrorResume(
- e -> (e instanceof EmptyDmaapResponseException || e instanceof TimeoutException),
+ e -> e instanceof Exception,
e -> Mono.empty())
.map(event -> {
// For each message, we have to keep separate state. This state will be enhanced
e.getMessage())
)
.onErrorResume(
- e -> e instanceof AaiTaskException || e instanceof TimeoutException,
+ e -> e instanceof Exception,
e -> Mono.empty())
.map(p -> {
state.setPnfAaiObject(p);
// towards the HSI CFS service
String serviceInstanceId = pnf.getRelationshipListAaiObject().getRelationshipEntries()
.stream()
- .filter(e -> e.getRelatedTo().equals("service-instance"))
+ .filter(e -> "service-instance".equals(e.getRelatedTo()))
.flatMap(e -> e.getRelationshipData().stream())
- .filter(d -> d.getRelationshipKey().equals("service-instance.service-instance-id"))
+ .filter(d -> "service-instance.service-instance-id".equals(d.getRelationshipKey()))
.map(RelationshipListAaiObject.RelationshipDataEntryAaiObject::getRelationshipValue)
.findFirst().orElse("");
e.getMessage())
)
.onErrorResume(
- e -> e instanceof AaiTaskException || e instanceof TimeoutException,
+ e -> e instanceof Exception,
e -> Mono.empty())
.map(s -> {
state.setHsiCfsServiceInstance(s);
state.getPnfAaiObject().getRelationshipListAaiObject().getRelationshipEntries();
// If no logical-link, fail further processing
- if (relationshipEntries.stream().noneMatch(e -> e.getRelatedTo().equals("logical-link"))) {
+ if (relationshipEntries.stream().noneMatch(e -> "logical-link".equals(e.getRelatedTo()))) {
LOGGER.warn("PNF {} does not have any logical-links bridged. Stop further processing",
state.getPnfAaiObject().getPnfName());
return true;
// Assuming PNF will only have one logical-link per BBS use case design
boolean isNotRelocation = relationshipEntries
.stream()
- .filter(e -> e.getRelatedTo().equals("logical-link"))
+ .filter(e -> "logical-link".equals(e.getRelatedTo()))
.flatMap(e -> e.getRelationshipData().stream())
.anyMatch(d -> d.getRelationshipValue()
.equals(state.getReRegistrationEvent().getAttachmentPoint()));
)
.doOnError(e -> LOGGER.error("Error while triggering Policy: {}", e.getMessage()))
.onErrorResume(
- e -> e instanceof DmaapException || e instanceof TimeoutException,
+ e -> e instanceof Exception,
e -> Mono.empty());
}
ControlLoopPublisherDmaapModel triggerEvent = ImmutableControlLoopPublisherDmaapModel.builder()
.closedLoopEventClient(DCAE_BBS_EVENT_PROCESSOR_MS_INSTANCE)
- .policyVersion(POLICY_VERSION)
+ .policyVersion(configuration.getPolicyVersion())
.policyName(POLICY_NAME)
.policyScope(configuration.getReRegistrationCloseLoopPolicyScope())
- .targetType(CLOSE_LOOP_TARGET_TYPE)
+ .targetType(configuration.getCloseLoopTargetType())
.aaiEnrichmentData(enrichmentData)
.closedLoopAlarmStart(Instant.now().getEpochSecond())
- .closedLoopEventStatus(CLOSED_LOOP_EVENT_STATUS)
+ .closedLoopEventStatus(configuration.getCloseLoopEventStatus())
.closedLoopControlName(configuration.getReRegistrationCloseLoopControlName())
- .version(CLOSE_LOOP_VERSION)
- .target(CLOSE_LOOP_TARGET)
+ .version(configuration.getCloseLoopVersion())
+ .target(configuration.getCloseLoopTarget())
.requestId(UUID.randomUUID().toString())
- .originator(FROM)
+ .originator(configuration.getCloseLoopOriginator())
.build();
LOGGER.debug("Processing Step: Publish for Policy");
LOGGER.trace("Trigger Policy event: ({})",triggerEvent);
}
@Override
- public void updateConfiguration(ApplicationConfiguration newConfiguration) {
- if (newConfiguration.getPipelinesPollingIntervalInSeconds() != currentPipelinesPollingInterval
- || newConfiguration.getCbsPollingInterval() != currentCbsPollingInterval) {
- configuration = newConfiguration;
- }
- if (newConfiguration.getPipelinesPollingIntervalInSeconds() != currentPipelinesPollingInterval) {
+ public void updateConfiguration() {
+ if (configuration.getPipelinesPollingIntervalInSeconds() != currentPipelinesPollingInterval) {
LOGGER.info("Pipelines Polling interval has changed. Re-scheduling processing pipelines");
cancelScheduledProcessingTasks();
reScheduleProcessingTasks();
}
- int newCbsPollingInterval = newConfiguration.getCbsPollingInterval();
+ int newCbsPollingInterval = configuration.getCbsPollingInterval();
if (newCbsPollingInterval != currentCbsPollingInterval) {
if (newCbsPollingInterval < DEFAULT_CBS_POLLING_INTERVAL) {
LOGGER.warn("CBS Polling interval is too small ({}). Will not re-schedule CBS job",
import org.onap.bbs.event.processor.model.PnfAaiObject;
import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
-import org.onap.bbs.event.processor.utilities.AaiReactiveClient;
import reactor.core.publisher.Mono;
Mono<PnfAaiObject> executePnfRetrieval(String taskName, String url);
Mono<ServiceInstanceAaiObject> executeServiceInstanceRetrieval(String taskName, String url);
-
- AaiReactiveClient resolveClient();
}
private final AaiReactiveClient reactiveClient;
@Autowired
- public AaiClientTaskImpl(AaiReactiveClient reactiveClient) {
+ AaiClientTaskImpl(AaiReactiveClient reactiveClient) {
this.reactiveClient = reactiveClient;
}
throw new AaiTaskException("Cannot invoke an A&AI client task with an invalid URL");
}
LOGGER.info("Executing task ({}) for retrieving PNF object", taskName);
- return resolveClient().getPnfObjectDataFor(url);
+ return reactiveClient.getPnfObjectDataFor(url);
}
@Override
throw new AaiTaskException("Cannot invoke an A&AI client task with an invalid URL");
}
LOGGER.info("Executing task ({}) for retrieving Service Instance object", taskName);
- return resolveClient().getServiceInstanceObjectDataFor(url);
- }
-
- @Override
- public AaiReactiveClient resolveClient() {
- return reactiveClient;
+ return reactiveClient.getServiceInstanceObjectDataFor(url);
}
}
import javax.net.ssl.SSLException;
import org.onap.bbs.event.processor.model.CpeAuthenticationConsumerDmaapModel;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
import reactor.core.publisher.Flux;
public interface DmaapCpeAuthenticationConsumerTask {
Flux<CpeAuthenticationConsumerDmaapModel> execute(String object) throws SSLException;
-
- DMaaPConsumerReactiveHttpClient resolveClient() throws SSLException;
}
package org.onap.bbs.event.processor.tasks;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
import javax.net.ssl.SSLException;
import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.config.ConfigurationChangeObserver;
import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
import org.onap.bbs.event.processor.model.CpeAuthenticationConsumerDmaapModel;
import org.onap.bbs.event.processor.utilities.CpeAuthenticationDmaapConsumerJsonParser;
import reactor.core.publisher.Mono;
@Component
-public class DmaapCpeAuthenticationConsumerTaskImpl implements DmaapCpeAuthenticationConsumerTask {
+public class DmaapCpeAuthenticationConsumerTaskImpl
+ implements DmaapCpeAuthenticationConsumerTask, ConfigurationChangeObserver {
private static final Logger LOGGER = LoggerFactory.getLogger(DmaapCpeAuthenticationConsumerTaskImpl.class);
- private final ApplicationConfiguration configuration;
+ private ApplicationConfiguration configuration;
private final CpeAuthenticationDmaapConsumerJsonParser cpeAuthenticationDmaapConsumerJsonParser;
private final ConsumerReactiveHttpClientFactory httpClientFactory;
+ private static final EmptyDmaapResponseException EMPTY_DMAAP_EXCEPTION =
+ new EmptyDmaapResponseException("CPE Authentication: Got an empty response from DMaaP");
+
+ private DMaaPConsumerReactiveHttpClient httpClient;
+
@Autowired
- public DmaapCpeAuthenticationConsumerTaskImpl(ApplicationConfiguration configuration) {
+ public DmaapCpeAuthenticationConsumerTaskImpl(ApplicationConfiguration configuration) throws SSLException {
this(configuration, new CpeAuthenticationDmaapConsumerJsonParser(),
new ConsumerReactiveHttpClientFactory(new DMaaPReactiveWebClientFactory()));
}
DmaapCpeAuthenticationConsumerTaskImpl(ApplicationConfiguration configuration,
CpeAuthenticationDmaapConsumerJsonParser
cpeAuthenticationDmaapConsumerJsonParser,
- ConsumerReactiveHttpClientFactory httpClientFactory) {
+ ConsumerReactiveHttpClientFactory httpClientFactory) throws SSLException {
this.configuration = configuration;
this.cpeAuthenticationDmaapConsumerJsonParser = cpeAuthenticationDmaapConsumerJsonParser;
this.httpClientFactory = httpClientFactory;
+
+ httpClient = httpClientFactory.create(this.configuration.getDmaapCpeAuthenticationConsumerConfiguration());
+ }
+
+ @PostConstruct
+ void registerForConfigChanges() {
+ configuration.register(this);
+ }
+
+ @PreDestroy
+ void unRegisterForConfigChanges() {
+ configuration.unRegister(this);
}
@Override
- public Flux<CpeAuthenticationConsumerDmaapModel> execute(String taskName) throws SSLException {
+ public synchronized void updateConfiguration() {
+ try {
+ LOGGER.info("DMaaP CPE authentication consumer update due to new application configuration");
+ httpClient = httpClientFactory.create(this.configuration.getDmaapCpeAuthenticationConsumerConfiguration());
+ } catch (SSLException e) {
+ LOGGER.error("Error while updating HTTP Client after a config update: SSL exception");
+ }
+ }
+
+ @Override
+ public Flux<CpeAuthenticationConsumerDmaapModel> execute(String taskName) {
LOGGER.debug("Executing task for CPE-Authentication with name \"{}\"", taskName);
- DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient = resolveClient();
- Mono<String> response = dmaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse();
+ DMaaPConsumerReactiveHttpClient httpClient = getHttpClient();
+ Mono<String> response = httpClient.getDMaaPConsumerResponse();
return cpeAuthenticationDmaapConsumerJsonParser.extractModelFromDmaap(response)
- .switchIfEmpty(Flux.error(
- new EmptyDmaapResponseException("CPE Authentication: Got an empty response from DMaaP")))
+ .switchIfEmpty(Flux.error(EMPTY_DMAAP_EXCEPTION))
.doOnError(e -> {
if (!(e instanceof EmptyDmaapResponseException)) {
LOGGER.error("DMaaP Consumption Exception: {}", e.getMessage());
});
}
- @Override
- public DMaaPConsumerReactiveHttpClient resolveClient() throws SSLException {
- return httpClientFactory.create(configuration.getDmaapCpeAuthenticationConsumerConfiguration());
+ private synchronized DMaaPConsumerReactiveHttpClient getHttpClient() {
+ return httpClient;
}
}
package org.onap.bbs.event.processor.tasks;
import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.producer.DMaaPPublisherReactiveHttpClient;
import org.springframework.http.ResponseEntity;
import reactor.core.publisher.Mono;
public interface DmaapPublisherTask {
Mono<ResponseEntity<String>> execute(ControlLoopPublisherDmaapModel controlLoopPublisherDmaapModel);
-
- DMaaPPublisherReactiveHttpClient resolveClient();
}
package org.onap.bbs.event.processor.tasks;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+
import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.config.ConfigurationChangeObserver;
import org.onap.bbs.event.processor.exceptions.DmaapException;
import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
import org.onap.bbs.event.processor.utilities.ControlLoopJsonBodyBuilder;
import reactor.core.publisher.Mono;
@Component
-public class DmaapPublisherTaskImpl implements DmaapPublisherTask {
+public class DmaapPublisherTaskImpl implements DmaapPublisherTask, ConfigurationChangeObserver {
private static final Logger LOGGER = LoggerFactory.getLogger(DmaapPublisherTaskImpl.class);
- private final ApplicationConfiguration configuration;
+ private ApplicationConfiguration configuration;
private final PublisherReactiveHttpClientFactory httpClientFactory;
+ private DMaaPPublisherReactiveHttpClient httpClient;
+
@Autowired
DmaapPublisherTaskImpl(ApplicationConfiguration configuration) {
this(configuration, new PublisherReactiveHttpClientFactory(new DmaaPRestTemplateFactory(),
PublisherReactiveHttpClientFactory httpClientFactory) {
this.configuration = configuration;
this.httpClientFactory = httpClientFactory;
+
+ httpClient = httpClientFactory.create(this.configuration.getDmaapPublisherConfiguration());
+ }
+
+ @PostConstruct
+ void registerForConfigChanges() {
+ configuration.register(this);
+ }
+
+ @PreDestroy
+ void unRegisterForConfigChanges() {
+ configuration.unRegister(this);
+ }
+
+ @Override
+ public synchronized void updateConfiguration() {
+ LOGGER.info("DMaaP Publisher update due to new application configuration");
+ httpClient = httpClientFactory.create(this.configuration.getDmaapPublisherConfiguration());
}
@Override
throw new DmaapException("Cannot invoke a DMaaP Publish task with a null message");
}
LOGGER.info("Executing task for publishing control loop message \n{}", controlLoopPublisherDmaapModel);
- DMaaPPublisherReactiveHttpClient dmaapPublisherReactiveHttpClient = resolveClient();
- return dmaapPublisherReactiveHttpClient.getDMaaPProducerResponse(controlLoopPublisherDmaapModel);
+ DMaaPPublisherReactiveHttpClient httpClient = getHttpClient();
+ return httpClient.getDMaaPProducerResponse(controlLoopPublisherDmaapModel);
}
- @Override
- public DMaaPPublisherReactiveHttpClient resolveClient() {
- return httpClientFactory.create(configuration.getDmaapPublisherConfiguration());
+ private synchronized DMaaPPublisherReactiveHttpClient getHttpClient() {
+ return httpClient;
}
}
import javax.net.ssl.SSLException;
import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
import reactor.core.publisher.Flux;
public interface DmaapReRegistrationConsumerTask {
Flux<ReRegistrationConsumerDmaapModel> execute(String taskName) throws SSLException;
-
- DMaaPConsumerReactiveHttpClient resolveClient() throws SSLException;
}
package org.onap.bbs.event.processor.tasks;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
import javax.net.ssl.SSLException;
import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.config.ConfigurationChangeObserver;
import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
import org.onap.bbs.event.processor.utilities.ReRegistrationDmaapConsumerJsonParser;
import reactor.core.publisher.Mono;
@Component
-public class DmaapReRegistrationConsumerTaskImpl implements DmaapReRegistrationConsumerTask {
+public class DmaapReRegistrationConsumerTaskImpl implements DmaapReRegistrationConsumerTask,
+ ConfigurationChangeObserver {
private static final Logger LOGGER = LoggerFactory.getLogger(DmaapReRegistrationConsumerTaskImpl.class);
- private final ApplicationConfiguration configuration;
+ private ApplicationConfiguration configuration;
private final ReRegistrationDmaapConsumerJsonParser reRegistrationDmaapConsumerJsonParser;
private final ConsumerReactiveHttpClientFactory httpClientFactory;
+ private static final EmptyDmaapResponseException EMPTY_DMAAP_EXCEPTION =
+ new EmptyDmaapResponseException("PNF Re-Registration: Got an empty response from DMaaP");
+
+ private DMaaPConsumerReactiveHttpClient httpClient;
+
@Autowired
- public DmaapReRegistrationConsumerTaskImpl(ApplicationConfiguration configuration) {
+ public DmaapReRegistrationConsumerTaskImpl(ApplicationConfiguration configuration) throws SSLException {
this(configuration, new ReRegistrationDmaapConsumerJsonParser(),
new ConsumerReactiveHttpClientFactory(new DMaaPReactiveWebClientFactory()));
}
DmaapReRegistrationConsumerTaskImpl(ApplicationConfiguration configuration,
ReRegistrationDmaapConsumerJsonParser reRegDmaapConsumerJsonParser,
- ConsumerReactiveHttpClientFactory httpClientFactory) {
+ ConsumerReactiveHttpClientFactory httpClientFactory)
+ throws SSLException {
this.configuration = configuration;
this.reRegistrationDmaapConsumerJsonParser = reRegDmaapConsumerJsonParser;
this.httpClientFactory = httpClientFactory;
+
+ httpClient = httpClientFactory.create(this.configuration.getDmaapReRegistrationConsumerConfiguration());
+ }
+
+ @PostConstruct
+ void registerForConfigChanges() {
+ configuration.register(this);
+ }
+
+ @PreDestroy
+ void unRegisterForConfigChanges() {
+ configuration.unRegister(this);
}
@Override
- public Flux<ReRegistrationConsumerDmaapModel> execute(String taskName) throws SSLException {
+ public synchronized void updateConfiguration() {
+ try {
+ LOGGER.info("DMaaP PNF reregistration consumer update due to new application configuration");
+ httpClient = httpClientFactory.create(this.configuration.getDmaapReRegistrationConsumerConfiguration());
+ } catch (SSLException e) {
+ LOGGER.error("Error while updating HTTP Client after a config update: SSL exception");
+ }
+ }
+
+ @Override
+ public Flux<ReRegistrationConsumerDmaapModel> execute(String taskName) {
LOGGER.debug("Executing task for Re-Registration with name \"{}\"", taskName);
- DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient = resolveClient();
- Mono<String> response = dmaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse();
+ DMaaPConsumerReactiveHttpClient httpClient = getHttpClient();
+ Mono<String> response = httpClient.getDMaaPConsumerResponse();
return reRegistrationDmaapConsumerJsonParser.extractModelFromDmaap(response)
- .switchIfEmpty(Flux.error(
- new EmptyDmaapResponseException("Re-Registration: Got an empty response from DMaaP")))
+ .switchIfEmpty(Flux.error(EMPTY_DMAAP_EXCEPTION))
.doOnError(e -> {
if (!(e instanceof EmptyDmaapResponseException)) {
LOGGER.error("DMaaP Consumption Exception: {}", e.getMessage());
});
}
- @Override
- public DMaaPConsumerReactiveHttpClient resolveClient() throws SSLException {
- return httpClientFactory.create(configuration.getDmaapReRegistrationConsumerConfiguration());
+ private synchronized DMaaPConsumerReactiveHttpClient getHttpClient() {
+ return httpClient;
}
}
this.gson = gson;
this.sslFactory = new SslFactory();
- aaiClientConfiguration = configuration.getAaiClientConfiguration();
+ aaiClientConfiguration = this.configuration.getAaiClientConfiguration();
setupWebClient();
}
}
@Override
- public void updateConfiguration(ApplicationConfiguration configuration) {
+ public void updateConfiguration() {
AaiClientConfiguration newConfiguration = configuration.getAaiClientConfiguration();
if (aaiClientConfiguration.equals(newConfiguration)) {
- LOGGER.debug("No Configuration changes necessary for AAI Reactive client");
+ LOGGER.info("No Configuration changes necessary for AAI Reactive client");
} else {
- LOGGER.debug("AAI Reactive client must be re-configured");
- aaiClientConfiguration = newConfiguration;
- try {
- setupWebClient();
- } catch (SSLException e) {
- LOGGER.error("AAI Reactive client error while re-configuring WebClient");
+ synchronized (this) {
+ LOGGER.info("AAI Reactive client must be re-configured");
+ aaiClientConfiguration = newConfiguration;
+ try {
+ setupWebClient();
+ } catch (SSLException e) {
+ LOGGER.error("AAI Reactive client error while re-configuring WebClient");
+ }
}
}
}
- private synchronized void setupWebClient() throws SSLException {
+ private void setupWebClient() throws SSLException {
SslContext sslContext = createSslContext();
ClientHttpConnector reactorClientHttpConnector = new ReactorClientHttpConnector(
return performReactiveHttpGet(url, ServiceInstanceAaiObject.class);
}
- private synchronized <T> Mono<T> performReactiveHttpGet(String url, Class<T> responseType) {
+ private <T> Mono<T> performReactiveHttpGet(String url, Class<T> responseType) {
LOGGER.debug("Will issue Reactive GET request to URL ({}) for object ({})", url, responseType.getName());
+ WebClient webClient = getWebClient();
return webClient
.get()
.uri(url)
}
return sslFactory.createInsecureContext();
}
+
+ private synchronized WebClient getWebClient() {
+ return webClient;
+ }
}
application:
pipelinesPollingIntervalSec: 30
pipelinesTimeoutSec: 15
+ policyVersion: 1.0.0.5
+ clTargetType: VM
+ clEventStatus: ONSET
+ clVersion: 1.0.2
+ clTarget: vserver.vserver-name
+ clOriginator: DCAE-bbs-event-processor
re-registration:
policyScope: policyScope
clControlName: controlName
"configs.security.enableAaiCertAuth=true",
"configs.application.pipelinesPollingIntervalSec=30",
"configs.application.pipelinesTimeoutSec=15",
+ "configs.application.policyVersion=1.0.0",
+ "configs.application.clTargetType=VM",
+ "configs.application.clEventStatus=ONSET",
+ "configs.application.clVersion=1.0.2",
+ "configs.application.clTarget=vserver.vserver-name",
+ "configs.application.clOriginator=DCAE-bbs-event-processor",
"configs.application.re-registration.policyScope=reRegPolicyScope",
"configs.application.re-registration.clControlName=reRegControlName",
"configs.application.cpe-authentication.policyScope=cpeAuthPolicyScope",
assertAll("Generic Application Properties",
() -> assertEquals(30, configuration.getPipelinesPollingIntervalInSeconds()),
() -> assertEquals(15, configuration.getPipelinesTimeoutInSeconds()),
+ () -> assertEquals("1.0.0", configuration.getPolicyVersion()),
+ () -> assertEquals("VM", configuration.getCloseLoopTargetType()),
+ () -> assertEquals("ONSET", configuration.getCloseLoopEventStatus()),
+ () -> assertEquals("1.0.2", configuration.getCloseLoopVersion()),
+ () -> assertEquals("vserver.vserver-name", configuration.getCloseLoopTarget()),
+ () -> assertEquals("DCAE-bbs-event-processor", configuration.getCloseLoopOriginator()),
() -> assertEquals("reRegPolicyScope", configuration.getReRegistrationCloseLoopPolicyScope()),
() -> assertEquals("cpeAuthPolicyScope", configuration.getCpeAuthenticationCloseLoopPolicyScope()),
() -> assertEquals("reRegControlName", configuration.getReRegistrationCloseLoopControlName()),
.pipelinesPollingIntervalSec(20)
.pipelinesTimeoutSec(20)
.cbsPollingIntervalSec(180)
+ .policyVersion("2.0.0")
+ .closeLoopTargetType("VM2")
+ .closeLoopEventStatus("ONSET-update")
+ .closeLoopVersion("2.0.2")
+ .closeLoopTarget("Target-update")
+ .closeLoopOriginator("Originator-update")
.reRegistrationPolicyScope("policyScope-update")
.reRegistrationClControlName("controlName-update")
.cpeAuthPolicyScope("policyScope-update")
() -> assertEquals(20, configuration.getPipelinesPollingIntervalInSeconds()),
() -> assertEquals(20, configuration.getPipelinesTimeoutInSeconds()),
() -> assertEquals(180, configuration.getCbsPollingInterval()),
+ () -> assertEquals("2.0.0", configuration.getPolicyVersion()),
+ () -> assertEquals("VM2", configuration.getCloseLoopTargetType()),
+ () -> assertEquals("ONSET-update", configuration.getCloseLoopEventStatus()),
+ () -> assertEquals("2.0.2", configuration.getCloseLoopVersion()),
+ () -> assertEquals("Target-update", configuration.getCloseLoopTarget()),
+ () -> assertEquals("Originator-update", configuration.getCloseLoopOriginator()),
() -> assertEquals("policyScope-update", configuration.getReRegistrationCloseLoopPolicyScope()),
() -> assertEquals("policyScope-update", configuration.getCpeAuthenticationCloseLoopPolicyScope()),
() -> assertEquals("controlName-update", configuration.getReRegistrationCloseLoopControlName()),
+ "\"application.reregistration.clControlName\": \"controlName\","
+ "\"application.cpe.authentication.policyScope\": \"policyScope\","
+ "\"application.cpe.authentication.clControlName\": \"controlName\","
+ + "\"application.policyVersion\": \"1.0\","
+ + "\"application.clTargetType\": \"VM\","
+ + "\"application.clEventStatus\": \"ONSET\","
+ + "\"application.clVersion\": \"1.0.2\","
+ + "\"application.clTarget\": \"vserver.vserver-name\","
+ + "\"application.clOriginator\": \"DCAE-bbs-event-processor\","
+ "\"application.reregistration.configKey\": \"config_key_2\","
+ "\"application.cpeAuth.configKey\": \"config_key_1\","
+ "\"application.closeLoop.configKey\": \"config_key_3\","
.reRegistrationClControlName("controlName")
.cpeAuthPolicyScope("policyScope")
.cpeAuthClControlName("controlName")
+ .policyVersion("1.0")
+ .closeLoopTargetType("VM")
+ .closeLoopEventStatus("ONSET")
+ .closeLoopVersion("1.0.2")
+ .closeLoopTarget("vserver.vserver-name")
+ .closeLoopOriginator("DCAE-bbs-event-processor")
.reRegConfigKey("config_key_2")
.cpeAuthConfigKey("config_key_1")
.closeLoopConfigKey("config_key_3")
@TestPropertySource(properties = {
"configs.application.pipelinesPollingIntervalSec=30",
"configs.application.pipelinesTimeoutSec=15",
+ "configs.application.policyVersion=1.0.0",
+ "configs.application.clTargetType=VM",
+ "configs.application.clEventStatus=ONSET",
+ "configs.application.clVersion=1.0.2",
+ "configs.application.clTarget=vserver.vserver-name",
+ "configs.application.clOriginator=DCAE-bbs-event-processor",
"configs.application.re-registration.policyScope=reRegPolicyScope",
"configs.application.re-registration.clControlName=reRegControlName",
"configs.application.cpe-authentication.policyScope=cpeAuthPolicyScope",
assertAll("Generic Application Properties",
() -> assertEquals(30, genericProperties.getPipelinesPollingIntervalSec()),
() -> assertEquals(15, genericProperties.getPipelinesTimeoutSec()),
+ () -> assertEquals("1.0.0", genericProperties.getPolicyVersion()),
+ () -> assertEquals("VM", genericProperties.getClTargetType()),
+ () -> assertEquals("ONSET", genericProperties.getClEventStatus()),
+ () -> assertEquals("1.0.2", genericProperties.getClVersion()),
+ () -> assertEquals("vserver.vserver-name", genericProperties.getClTarget()),
+ () -> assertEquals("DCAE-bbs-event-processor", genericProperties.getClOriginator()),
() -> assertEquals("reRegPolicyScope", genericProperties.getReRegistration().getPolicyScope()),
() -> assertEquals("cpeAuthPolicyScope",
genericProperties.getCpeAuthentication().getPolicyScope()),
.thenReturn("controlName");
when(configuration.getCpeAuthenticationCloseLoopPolicyScope())
.thenReturn("policyScope");
+ when(configuration.getPolicyVersion())
+ .thenReturn("1.0.0");
+ when(configuration.getCloseLoopTargetType())
+ .thenReturn("VM");
+ when(configuration.getCloseLoopEventStatus())
+ .thenReturn("ONSET");
+ when(configuration.getCloseLoopVersion())
+ .thenReturn("1.0.2");
+ when(configuration.getCloseLoopTarget())
+ .thenReturn("CL-Target");
+ when(configuration.getCloseLoopOriginator())
+ .thenReturn("DCAE-BBS-ep");
pipeline = new CpeAuthenticationPipeline(configuration, consumerTask,
publisherTask, aaiClientTask, new HashMap<>());
.thenReturn("controlName");
when(configuration.getReRegistrationCloseLoopPolicyScope())
.thenReturn("policyScope");
+ when(configuration.getPolicyVersion())
+ .thenReturn("1.0.0");
+ when(configuration.getCloseLoopTargetType())
+ .thenReturn("VM");
+ when(configuration.getCloseLoopEventStatus())
+ .thenReturn("ONSET");
+ when(configuration.getCloseLoopVersion())
+ .thenReturn("1.0.2");
+ when(configuration.getCloseLoopTarget())
+ .thenReturn("CL-Target");
+ when(configuration.getCloseLoopOriginator())
+ .thenReturn("DCAE-BBS-ep");
pipeline = new ReRegistrationPipeline(configuration, consumerTask,
publisherTask, aaiClientTask, new HashMap<>());
Code Review / dcaegen2 / services.git / commitdiff