Merge "Jakarta changes in slice-analysis-ms for IBN Cloud leased line update and CCVPN closed-loop. This commit contains work in sub-task: 1) AAI monitor thread; 2) bandwidth evaluator; 3) policy notification code."

23 files changed, 2435 insertions, 384 deletions

diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiInterface.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiInterface.java
index 87718a5c..582916aa 100644
--- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiInterface.java
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiInterface.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2021-2022 Wipro Limited.
+ *   Copyright (C) 2022 Huawei Canada Limited.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -33,4 +34,6 @@ public interface AaiInterface {
     public Map<String, String> fetchServiceDetails(String snssai);
 
     public Map<String, Integer> fetchCurrentConfigurationOfSlice(String snssai);
+
+    public Map<String, Integer> fetchMaxBandwidthOfService(String serviceId);
 }
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiService.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiService.java
index 6284ddfb..eb5d243c 100644
--- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiService.java
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/aai/AaiService.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2021-2022 Wipro Limited.
+ *   Copyright (C) 2022 Huawei Canada Limited.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -38,6 +39,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 
 /**
@@ -347,4 +349,35 @@ public class AaiService implements AaiInterface {
 
     }
 
+    /**
+     * Get network bandwidth attribute of an ethernet service. These data is inside a network policy whose
+     * etwork-policy-fqdn equals to provided service-instance-id
+     * @param serviceId target service instance id
+     * @return Map contains maxBandwidth value of given service-instance
+     */
+    public Map<String, Integer> fetchMaxBandwidthOfService(String serviceId){
+        log.info("Fetching max-bandwidth from AAI network-policy");
+        String networkPolicyUrl = aaiBaseUrl + "/network/network-policies" + "?network-policy-fqdn="
+                + serviceId;
+        Map<String, Integer> result = new HashMap<>();
+        try {
+            ResponseEntity<String> resp = restclient.sendGetRequest(networkPolicyUrl, new ParameterizedTypeReference<String>() {
+            });
+            if (resp.getStatusCodeValue() == 200){
+                String networkPolicy = resp.getBody();
+                JSONObject networkPolicyJson = new JSONObject(networkPolicy);
+                JSONArray networkPolicyList	= networkPolicyJson.optJSONArray("network-policy");
+                if (networkPolicyList != null){
+                    JSONObject networkPolicyOjb = networkPolicyList.getJSONObject(0);
+                    result.put("maxBandwidth", networkPolicyOjb.getInt("max-bandwidth"));
+                    return result;
+                }
+                log.info("Successfully fetched max bandwidth {}: {}", serviceId, result);
+            }
+        } catch (Exception e){
+            log.warn("Error encountered when fetching maxbandwidth: " + e);
+
+        }
+        return null;
+    }
 }
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/AaiEventNotificationCallback.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/AaiEventNotificationCallback.java
new file mode 100644
index 00000000..dc2cd775
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/AaiEventNotificationCallback.java
@@ -0,0 +1,96 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *  Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+
+package org.onap.slice.analysis.ms.dmaap;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import org.onap.slice.analysis.ms.models.Configuration;
+import org.onap.slice.analysis.ms.service.ccvpn.BandwidthEvaluator;
+import org.onap.slice.analysis.ms.service.ccvpn.Event;
+import org.onap.slice.analysis.ms.service.ccvpn.SimpleEvent;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * Handles AAI-EVENT from dmaap
+ */
+@Component
+public class AaiEventNotificationCallback implements NotificationCallback {
+
+    private static final String EVENT_HEADER = "event-header";
+    private static final String ACTION = "action";
+    private static final String ENTITY_TYPE = "entity-type";
+    private static final String SOURCE_NAME = "source-name";
+    private static final String ENTITY = "entity";
+    private final JsonParser parser = new JsonParser();
+    private Configuration configuration;
+    private String aaiNotifTargetAction;
+    private String aaiNotifTargetSource;
+    private String aaiNotifTargetEntity;
+
+    @Autowired
+    BandwidthEvaluator bandwidthEvaluator;
+
+    @PostConstruct
+    public void init(){
+        configuration = Configuration.getInstance();
+        aaiNotifTargetAction = configuration.getAaiNotifTargetAction();
+        aaiNotifTargetSource = configuration.getAaiNotifTargetSource();
+        aaiNotifTargetEntity = configuration.getAaiNotifTargetEntity();
+    }
+
+    @Override
+    public void activateCallBack(String msg) {
+        handleNotification(msg);
+    }
+
+    private void handleNotification(String msg) {
+        JsonElement jsonElement = parser.parse(msg);
+        if (jsonElement.isJsonObject()){
+            //handle a single AAI_EVENT
+            handleMsgJsonObject(jsonElement.getAsJsonObject());
+        } else if (jsonElement.isJsonArray()){
+            //handle a series of AAI_EVENT
+            JsonArray jsonArray = jsonElement.getAsJsonArray();
+            for (int i=0,e=jsonArray.size(); i<e; i++){
+                handleMsgJsonObject(jsonArray.get(i).getAsJsonObject());
+            }
+        }
+    }
+
+    private void handleMsgJsonObject(JsonObject jsonObject){
+        JsonObject header = jsonObject.get(EVENT_HEADER).getAsJsonObject();
+        if (header.has(ACTION) && header.get(ACTION).getAsString().equals(aaiNotifTargetAction)) {
+            if (header.has(ENTITY_TYPE) && header.get(ENTITY_TYPE).getAsString().equals(aaiNotifTargetEntity)){
+                if (header.has(SOURCE_NAME) && header.get(SOURCE_NAME).getAsString().equals(aaiNotifTargetSource)) {
+                    JsonObject body = jsonObject.get(ENTITY).getAsJsonObject();
+                    Event event = new SimpleEvent<>(SimpleEvent.Type.ONDEMAND_CHECK, body);
+                    bandwidthEvaluator.post(event);
+                }
+            }
+        }
+    }
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/DmaapClient.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/DmaapClient.java
index 6e0f4f27..ad5941a4 100644
--- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/DmaapClient.java
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/DmaapClient.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020 Wipro Limited.
+ *   Copyright (C) 2022 Huawei Canada Limited.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -38,20 +39,27 @@ import org.springframework.stereotype.Component;
 import com.att.nsa.cambria.client.CambriaConsumer;
 
 /**
- * This class initializes and starts the dmaap client 
+ * This class initializes and starts the dmaap client
  * to listen on application required dmaap events
  */
 @Component
 public class DmaapClient {
 
+    private static final String AAI_SUBSCRIBER = "aai_subscriber";
     private Configuration configuration;
     private static Logger log = LoggerFactory.getLogger(DmaapClient.class);
 
     private DmaapUtils dmaapUtils;
-    
+
     @Autowired
     private IntelligentSlicingCallback intelligentSlicingCallback;
 
+    @Autowired
+    private VesNotificationCallback vesNotificationCallback;
+
+    @Autowired
+    private AaiEventNotificationCallback aaiEventNotificationCallback;
+
     /**
      * init dmaap client.
      */
@@ -74,28 +82,37 @@ public class DmaapClient {
     public synchronized void startClient() {
 
         Map<String, Object> streamSubscribes = configuration.getStreamsSubscribes();
-       
+
         String pmTopicUrl = ((Map<String, String>) ((Map<String, Object>) streamSubscribes
                 .get("performance_management_topic")).get("dmaap_info")).get("topic_url");
         String[] pmTopicSplit = pmTopicUrl.split("\\/");
         String pmTopic = pmTopicSplit[pmTopicSplit.length - 1];
         log.debug("pm topic : {}", pmTopic);
-        
+
         String policyResponseTopicUrl = ((Map<String, String>) ((Map<String, Object>) streamSubscribes
                 .get("dcae_cl_response_topic")).get("dmaap_info")).get("topic_url");
         String[] policyResponseTopicUrlSplit = policyResponseTopicUrl.split("\\/");
         String policyResponseTopic = policyResponseTopicUrlSplit[policyResponseTopicUrlSplit.length - 1];
         log.debug("policyResponse Topic : {}", policyResponseTopic);
-        
+
         String intelligentSlicingTopicUrl = ((Map<String, String>) ((Map<String, Object>) streamSubscribes
                 .get("intelligent_slicing_topic")).get("dmaap_info")).get("topic_url");
         String[] intelligentSlicingTopicSplit = intelligentSlicingTopicUrl.split("\\/");
         String intelligentSlicingTopic = intelligentSlicingTopicSplit[intelligentSlicingTopicSplit.length - 1];
         log.debug("intelligent slicing topic : {}", pmTopic);
-        
+
+        // Parsing ccvpn notification topic
+        String ccvpnNotiTopicUrl = ((Map<String, String>) ((Map<String, Object>) streamSubscribes
+                .get("ves_ccvpn_notification_topic")).get("dmaap_info")).get("topic_url");
+        String[] ccvpnNotiTopicSplit = ccvpnNotiTopicUrl.split("\\/");
+        String ccvpnNotiTopic = ccvpnNotiTopicSplit[ccvpnNotiTopicSplit.length - 1];
+        log.debug("ccvpn notification topic : {}", ccvpnNotiTopic);
+
         CambriaConsumer pmNotifCambriaConsumer = dmaapUtils.buildConsumer(configuration, pmTopic);
         CambriaConsumer policyResponseCambriaConsumer = dmaapUtils.buildConsumer(configuration, policyResponseTopic);
         CambriaConsumer intelligentSlicingCambriaConsumer = dmaapUtils.buildConsumer(configuration, intelligentSlicingTopic);
+        // Creating ccvpn notification cambriaconsumer
+        CambriaConsumer ccvpnNotiCambriaConsumer = dmaapUtils.buildConsumer(configuration, ccvpnNotiTopic);
 
         ScheduledExecutorService executorPool;
 
@@ -106,22 +123,33 @@ public class DmaapClient {
         executorPool = Executors.newScheduledThreadPool(10);
         executorPool.scheduleAtFixedRate(pmNotificationConsumer, 0, configuration.getPollingInterval(),
                 TimeUnit.SECONDS);
-        
+
         // create notification consumers for Policy
- 		NotificationConsumer policyNotificationConsumer = new NotificationConsumer(policyResponseCambriaConsumer,
- 				new PolicyNotificationCallback());
- 		// start policy notification consumer threads
- 		executorPool = Executors.newScheduledThreadPool(10);
- 		executorPool.scheduleAtFixedRate(policyNotificationConsumer, 0, configuration.getPollingInterval(),
- 				TimeUnit.SECONDS);
- 		
-		// create notification consumers for ML MS
- 		NotificationConsumer intelligentSlicingConsumer = new NotificationConsumer(intelligentSlicingCambriaConsumer,
- 				intelligentSlicingCallback);
- 		// start intelligent Slicing notification consumer threads
- 		executorPool = Executors.newScheduledThreadPool(10);
- 		executorPool.scheduleAtFixedRate(intelligentSlicingConsumer, 0, configuration.getPollingInterval(),
- 				TimeUnit.SECONDS);
+        NotificationConsumer policyNotificationConsumer = new NotificationConsumer(policyResponseCambriaConsumer,
+                new PolicyNotificationCallback());
+        // start policy notification consumer threads
+        executorPool = Executors.newScheduledThreadPool(10);
+        executorPool.scheduleAtFixedRate(policyNotificationConsumer, 0, configuration.getPollingInterval(),
+                TimeUnit.SECONDS);
+
+        // create notification consumers for ML MS
+        NotificationConsumer intelligentSlicingConsumer = new NotificationConsumer(intelligentSlicingCambriaConsumer,
+                intelligentSlicingCallback);
+        // start intelligent Slicing notification consumer threads
+        executorPool = Executors.newScheduledThreadPool(10);
+        executorPool.scheduleAtFixedRate(intelligentSlicingConsumer, 0, configuration.getPollingInterval(),
+                TimeUnit.SECONDS);
+
+        // create notification consumers for ccvpn close-loop PM
+        NotificationConsumer ccvpnNotiConsumer = new NotificationConsumer(ccvpnNotiCambriaConsumer,
+                vesNotificationCallback);
+        executorPool = Executors.newScheduledThreadPool(1);
+        executorPool.scheduleWithFixedDelay(ccvpnNotiConsumer, 0, configuration.getVesNotifPollingInterval(),
+                TimeUnit.SECONDS);
+
+        // start AAI-EVENT dmaap topic monitor
+        MRTopicMonitor mrTopicMonitor = new MRTopicMonitor(AAI_SUBSCRIBER, aaiEventNotificationCallback);
+        mrTopicMonitor.start();
     }
 
 }
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/MRTopicMonitor.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/MRTopicMonitor.java
new file mode 100644
index 00000000..0c1ac604
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/MRTopicMonitor.java
@@ -0,0 +1,283 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ *  Copyright (C) 2022 Huawei Canada Limited.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.slice.analysis.ms.dmaap;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import lombok.Getter;
+import lombok.NonNull;
+import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
+import org.onap.dmaap.mr.client.response.MRConsumerResponse;
+import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
+import org.onap.slice.analysis.ms.models.Configuration;
+import org.onap.slice.analysis.ms.dmaap.MRTopicParams;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This is a Dmaap message-router topic monitor.
+ * It takes advantage of AT&T's dmaap-client's long-polling implementation, this monitor constantly fetch/refetch target msg topic.
+ * So that new msg can be notified almost immediately. This is the major different from previous implementation.
+ */
+public class MRTopicMonitor implements Runnable {
+
+    private final String name;
+    private volatile boolean running = false;
+    private Configuration configuration;
+    private static Logger logger = LoggerFactory.getLogger(MRTopicMonitor.class);
+    private static int DEFAULT_TIMEOUT_MS_FETCH = 15000;
+    private MRConsumerWrapper consumerWrapper;
+    private NotificationCallback callback;
+
+    /**
+     * Constructor
+     * @param name name of topic subscriber in config
+     * @param callback callbackfunction for received message
+     */
+    public MRTopicMonitor(String name, NotificationCallback callback){
+        this.name = name;
+        this.callback = callback;
+    }
+
+    /**
+     * Start the monitoring thread
+     */
+    public void start(){
+        logger.info("Starting Dmaap Bus Monitor");
+        configuration = Configuration.getInstance();
+
+        Map<String, Object> streamSubscribes = configuration.getStreamsSubscribes();
+        Map<String, Object> topicParamsJson = (Map<String, Object>) streamSubscribes.get(name);
+        JsonObject jsonObject = (new Gson()).toJsonTree(topicParamsJson).getAsJsonObject();
+        consumerWrapper = buildConsumerWrapper(jsonObject);
+        running = true;
+        Executor executor = Executors.newSingleThreadExecutor();
+        executor.execute(this);
+    }
+
+    /**
+     * Main loop that keep fetching and processing
+     */
+    @Override
+    public void run(){
+        while (running){
+            try {
+                Iterable<String> dmaapMsgs = consumerWrapper.fetch();
+                for (String msg : dmaapMsgs){
+                    process(msg);
+                }
+            } catch (IOException | RuntimeException e){
+                logger.error("fetchMessage encountered error: {}", e);
+            }
+        }
+        logger.info("{}: exiting thread", this);
+    }
+
+    /**
+     * Stop the monitor
+     */
+    public void stop(){
+        logger.info("{}: exiting", this);
+        running = false;
+        this.consumerWrapper.close();
+        this.consumerWrapper = null;
+    }
+
+    private void process(String msg) {
+        try {
+            callback.activateCallBack(msg);
+        } catch (Exception e){
+            logger.error("process message encountered error: {}", e);
+        }
+    }
+
+    private Iterable<String> fetch() throws IOException {
+        return this.consumerWrapper.fetch();
+    }
+
+    private MRConsumerWrapper buildConsumerWrapper(@NonNull JsonObject topicParamsJson )
+            throws IllegalArgumentException {
+        MRTopicParams topicParams = MRTopicParams.builder().buildFromConfigJson(topicParamsJson).build();
+        return new MRConsumerWrapper(topicParams);
+    }
+
+    /**
+     * Wrapper class of DmaapClient (package org.onap.dmaap.mr.client)
+     * A polling fashion dmaap  consumer, whose #fetch() sleep a certain time when connection fails, otherwise keep retryiny.
+     * It supports both https and http protocols.
+     */
+    private class MRConsumerWrapper {
+        /**
+         * Name of the "protocol" property.
+         */
+        protected static final String PROTOCOL_PROP = "Protocol";
+        /**
+         * Fetch timeout.
+         */
+        protected int fetchTimeout;
+
+        /**
+         * Time to sleep on a fetch failure.
+         */
+        @Getter
+        private final int sleepTime;
+
+        /**
+         * Counted down when {@link #close()} is invoked.
+         */
+        private final CountDownLatch closeCondition = new CountDownLatch(1);
+
+        /**
+         * MR Consumer.
+         */
+        protected MRConsumerImpl consumer;
+
+        /**
+         * Constructs the object.
+         *
+         * @param MRTopicParams parameters for the bus topic
+         */
+        protected MRConsumerWrapper(MRTopicParams MRTopicParams) {
+            this.fetchTimeout = MRTopicParams.getFetchTimeout();
+
+            if (this.fetchTimeout <= 0) {
+                this.sleepTime = DEFAULT_TIMEOUT_MS_FETCH;
+            } else {
+                // don't sleep too long, even if fetch timeout is large
+                this.sleepTime = Math.min(this.fetchTimeout, DEFAULT_TIMEOUT_MS_FETCH);
+            }
+
+            if (MRTopicParams.isTopicInvalid()) {
+                throw new IllegalArgumentException("No topic for DMaaP");
+            }
+
+            if (MRTopicParams.isServersInvalid()) {
+                throw new IllegalArgumentException("Must provide at least one host for HTTP AAF");
+            }
+
+            try{
+                this.consumer = new MRConsumerImpl.MRConsumerImplBuilder()
+                        .setHostPart(MRTopicParams.getServers())
+                        .setTopic(MRTopicParams.getTopic())
+                        .setConsumerGroup(MRTopicParams.getConsumerGroup())
+                        .setConsumerId(MRTopicParams.getConsumerInstance())
+                        .setTimeoutMs(MRTopicParams.getFetchTimeout())
+                        .setLimit(MRTopicParams.getFetchLimit())
+                        .setApiKey(MRTopicParams.getApiKey())
+                        .setApiSecret(MRTopicParams.getApiSecret())
+                        .createMRConsumerImpl();
+            } catch (MalformedURLException e) {
+                throw new IllegalArgumentException("Illegal MrConsumer parameters");
+            }
+
+
+            this.consumer.setUsername(MRTopicParams.getUserName());
+            this.consumer.setPassword(MRTopicParams.getPassword());
+
+            if(MRTopicParams.isUserNameValid() && MRTopicParams.isPasswordValid()){
+                this.consumer.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+            } else {
+                this.consumer.setProtocolFlag(ProtocolTypeConstants.HTTPNOAUTH.getValue());
+            }
+
+            Properties props = new Properties();
+
+            if (MRTopicParams.isUseHttps()) {
+                props.setProperty(PROTOCOL_PROP, "https");
+                this.consumer.setHost(MRTopicParams.getServers().get(0) + ":3905");
+
+            } else {
+                props.setProperty(PROTOCOL_PROP, "http");
+                this.consumer.setHost(MRTopicParams.getServers().get(0) + ":3904");
+            }
+
+            this.consumer.setProps(props);
+        }
+
+        /**
+         * Try fetch new message. But backoff for some sleepTime when connection fails.
+         * @return
+         * @throws IOException
+         */
+        public Iterable<String> fetch() throws IOException {
+            final MRConsumerResponse response = this.consumer.fetchWithReturnConsumerResponse();
+            if (response == null) {
+                logger.warn("{}: DMaaP NULL response received", this);
+
+                sleepAfterFetchFailure();
+                return new ArrayList<>();
+            } else {
+                logger.debug("DMaaP consumer received {} : {}", response.getResponseCode(),
+                        response.getResponseMessage());
+
+                if (!"200".equals(response.getResponseCode())) {
+
+                    logger.error("DMaaP consumer received: {} : {}", response.getResponseCode(),
+                            response.getResponseMessage());
+
+                    sleepAfterFetchFailure();
+                }
+            }
+
+            if (response.getActualMessages() == null) {
+                return new ArrayList<>();
+            } else {
+                return response.getActualMessages();
+            }
+        }
+
+        /**
+         * Causes the thread to sleep; invoked after fetch() fails.  If the consumer is closed,
+         * or the thread is interrupted, then this will return immediately.
+         */
+        protected void sleepAfterFetchFailure() {
+            try {
+                logger.info("{}: backoff for {}ms", this, sleepTime);
+                if (this.closeCondition.await(this.sleepTime, TimeUnit.MILLISECONDS)) {
+                    logger.info("{}: closed while handling fetch error", this);
+                }
+
+            } catch (InterruptedException e) {
+                logger.warn("{}: interrupted while handling fetch error", this, e);
+                Thread.currentThread().interrupt();
+            }
+        }
+
+        /**
+         * Close the dmaap client and this thread
+         */
+        public void close() {
+            this.closeCondition.countDown();
+            this.consumer.close();
+        }
+    }
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/MRTopicParams.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/MRTopicParams.java
new file mode 100644
index 00000000..66c27413
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/MRTopicParams.java
@@ -0,0 +1,380 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2018 Samsung Electronics Co., Ltd. All rights reserved.
+ * Modifications Copyright (C) 2018-2019, 2021 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2019 Nordix Foundation.
+ * Modifications Copyright (C) 2022 Huawei Canada Limited.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.slice.analysis.ms.dmaap;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Partially copied from Onap Policy
+ * policy/common/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParams.java
+ * Modified to fit this project.
+ * Member variables of this Params class are as follows.
+ *
+ * <p>servers DMaaP servers
+ * topic DMaaP Topic to be monitored
+ * apiKey DMaaP API Key (optional)
+ * apiSecret DMaaP API Secret (optional)
+ * consumerGroup DMaaP Reader Consumer Group
+ * consumerInstance DMaaP Reader Instance
+ * fetchTimeout DMaaP fetch timeout
+ * fetchLimit DMaaP fetch limit
+ * environment DME2 Environment
+ * aftEnvironment DME2 AFT Environment
+ * partner DME2 Partner
+ * latitude DME2 Latitude
+ * longitude DME2 Longitude
+ * additionalProps Additional properties to pass to DME2
+ * useHttps does connection use HTTPS?
+ * allowSelfSignedCerts are self-signed certificates allow
+ */
+@Getter
+@Setter
+public class MRTopicParams {
+
+    private int port;
+    private List<String> servers;
+    private Map<String, String> additionalProps;
+    private String topic;
+    private String effectiveTopic;
+    private String apiKey;
+    private String apiSecret;
+    private String consumerGroup;
+    private String consumerInstance;
+    private int fetchTimeout;
+    private int fetchLimit;
+    private boolean useHttps;
+    private boolean allowSelfSignedCerts;
+    private boolean managed;
+
+    private String userName;
+    private String password;
+    private String environment;
+    private String aftEnvironment;
+    private String partner;
+    private String latitude;
+    private String longitude;
+    private String partitionId;
+    private String clientName;
+    private String hostname;
+    private String basePath;
+    @Getter
+    private String serializationProvider;
+
+    public static TopicParamsBuilder builder() {
+        return new TopicParamsBuilder();
+    }
+
+    /**
+     * Methods to Check if the property is INVALID.
+     */
+
+    public boolean isEnvironmentInvalid() {
+        return StringUtils.isBlank(environment);
+    }
+
+    public boolean isAftEnvironmentInvalid() {
+        return StringUtils.isBlank(aftEnvironment);
+    }
+
+    public boolean isLatitudeInvalid() {
+        return StringUtils.isBlank(latitude);
+    }
+
+    public boolean isLongitudeInvalid() {
+        return StringUtils.isBlank(longitude);
+    }
+
+    public boolean isConsumerInstanceInvalid() {
+        return StringUtils.isBlank(consumerInstance);
+    }
+
+    public boolean isConsumerGroupInvalid() {
+        return StringUtils.isBlank(consumerGroup);
+    }
+
+    public boolean isClientNameInvalid() {
+        return StringUtils.isBlank(clientName);
+    }
+
+    public boolean isPartnerInvalid() {
+        return StringUtils.isBlank(partner);
+    }
+
+    public boolean isServersInvalid() {
+        return (servers == null || servers.isEmpty()
+                || (servers.size() == 1 && ("".equals(servers.get(0)))));
+    }
+
+    public boolean isTopicInvalid() {
+        return StringUtils.isBlank(topic);
+    }
+
+    public boolean isPartitionIdInvalid() {
+        return StringUtils.isBlank(partitionId);
+    }
+
+    public boolean isHostnameInvalid() {
+        return StringUtils.isBlank(hostname);
+    }
+
+    public boolean isPortInvalid() {
+        return  (port <= 0 || port >= 65535);
+    }
+
+    /**
+     * Methods to Check if the property is Valid.
+     */
+
+    public boolean isApiKeyValid() {
+        return StringUtils.isNotBlank(apiKey);
+    }
+
+    public boolean isApiSecretValid() {
+        return StringUtils.isNotBlank(apiSecret);
+    }
+
+    public boolean isUserNameValid() {
+        return StringUtils.isNotBlank(userName);
+    }
+
+    public boolean isPasswordValid() {
+        return StringUtils.isNotBlank(password);
+    }
+
+    public boolean isAdditionalPropsValid() {
+        return additionalProps != null;
+    }
+
+    @NoArgsConstructor(access = AccessLevel.PRIVATE)
+    public static class TopicParamsBuilder {
+
+        final MRTopicParams params = new MRTopicParams();
+
+        public TopicParamsBuilder servers(List<String> servers) {
+            this.params.servers = servers;
+            return this;
+        }
+
+        public TopicParamsBuilder topic(String topic) {
+            this.params.topic = topic;
+            return this;
+        }
+
+        public TopicParamsBuilder effectiveTopic(String effectiveTopic) {
+            this.params.effectiveTopic = effectiveTopic;
+            return this;
+        }
+
+        public TopicParamsBuilder apiKey(String apiKey) {
+            this.params.apiKey = apiKey;
+            return this;
+        }
+
+        public TopicParamsBuilder apiSecret(String apiSecret) {
+            this.params.apiSecret = apiSecret;
+            return this;
+        }
+
+        public TopicParamsBuilder consumerGroup(String consumerGroup) {
+            this.params.consumerGroup = consumerGroup;
+            return this;
+        }
+
+        public TopicParamsBuilder consumerInstance(String consumerInstance) {
+            this.params.consumerInstance = consumerInstance;
+            return this;
+        }
+
+        public TopicParamsBuilder fetchTimeout(int fetchTimeout) {
+            this.params.fetchTimeout = fetchTimeout;
+            return this;
+        }
+
+        public TopicParamsBuilder fetchLimit(int fetchLimit) {
+            this.params.fetchLimit = fetchLimit;
+            return this;
+        }
+
+        public TopicParamsBuilder useHttps(boolean useHttps) {
+            this.params.useHttps = useHttps;
+            return this;
+        }
+
+        public TopicParamsBuilder allowSelfSignedCerts(boolean allowSelfSignedCerts) {
+            this.params.allowSelfSignedCerts = allowSelfSignedCerts;
+            return this;
+        }
+
+        public TopicParamsBuilder userName(String userName) {
+            this.params.userName = userName;
+            return this;
+        }
+
+        public TopicParamsBuilder password(String password) {
+            this.params.password = password;
+            return this;
+        }
+
+        public TopicParamsBuilder environment(String environment) {
+            this.params.environment = environment;
+            return this;
+        }
+
+        public TopicParamsBuilder aftEnvironment(String aftEnvironment) {
+            this.params.aftEnvironment = aftEnvironment;
+            return this;
+        }
+
+        public TopicParamsBuilder partner(String partner) {
+            this.params.partner = partner;
+            return this;
+        }
+
+        public TopicParamsBuilder latitude(String latitude) {
+            this.params.latitude = latitude;
+            return this;
+        }
+
+        public TopicParamsBuilder longitude(String longitude) {
+            this.params.longitude = longitude;
+            return this;
+        }
+
+        public TopicParamsBuilder additionalProps(Map<String, String> additionalProps) {
+            this.params.additionalProps = additionalProps;
+            return this;
+        }
+
+        public TopicParamsBuilder partitionId(String partitionId) {
+            this.params.partitionId = partitionId;
+            return this;
+        }
+
+        public MRTopicParams build() {
+            return params;
+        }
+
+        public TopicParamsBuilder buildFromConfigJson(JsonObject jsonObject) {
+            String consumerGroup = null;
+            String consumerInstance = null;
+            String aafUsername = null;
+            String aafPassword = null;
+            List<String> servers = new ArrayList<>();
+            String topic = null;
+            boolean useHttps = false;
+            int fetchTimeout = -1;
+            int fetchLimit = -1;
+
+            if (jsonObject.has("consumer_group") && !jsonObject.get("consumer_group").isJsonNull()) {
+                consumerGroup = jsonObject.get("consumer_group").getAsString();
+            }
+            if (jsonObject.has("consumer_instance") && !jsonObject.get("consumer_instance").isJsonNull()) {
+                consumerInstance = jsonObject.get("consumer_instance").getAsString();
+            }
+            if (jsonObject.has("aaf_username") && !jsonObject.get("aaf_username").isJsonNull()) {
+                aafUsername = jsonObject.get("aaf_username").getAsString();
+            }
+            if (jsonObject.has("aaf_password") && !jsonObject.get("aaf_password").isJsonNull()) {
+                aafPassword = jsonObject.get("aaf_password").getAsString();
+            }
+            if (jsonObject.has("fetch_timeout") && !jsonObject.get("fetch_timeout").isJsonNull()) {
+                fetchTimeout = jsonObject.get("fetch_timeout").getAsInt();
+            }
+            if (jsonObject.has("fetch_limit") && !jsonObject.get("fetch_limit").isJsonNull()) {
+                fetchLimit = jsonObject.get("fetch_limit").getAsInt();
+            }
+            if (jsonObject.has("servers") && !jsonObject.get("servers").isJsonNull()) {
+                JsonArray jsonArray = jsonObject.get("servers").getAsJsonArray();
+                servers = new ArrayList<>();
+                for (int i=0, e=jsonArray.size(); i<e; i++){
+                    servers.add(jsonArray.get(i).getAsString());
+                }
+            }
+            if (jsonObject.has("servers") && !jsonObject.get("servers").isJsonNull()) {
+                JsonArray jsonArray = jsonObject.get("servers").getAsJsonArray();
+                for (int i=0, e=jsonArray.size(); i<e; i++){
+                    servers.add(jsonArray.get(i).getAsString());
+                }
+            }
+            String topicUrl = jsonObject.get("dmaap_info").getAsJsonObject().get("topic_url").getAsString();
+            if (topicUrl.startsWith("https")){
+                useHttps = true;
+            }
+            String[] pmTopicSplit = topicUrl.split("\\/");
+            topic = pmTopicSplit[pmTopicSplit.length - 1];
+
+            this.params.topic = topic;
+            this.params.servers = servers;
+            this.params.consumerGroup = consumerGroup;
+            this.params.consumerInstance = consumerInstance;
+            this.params.password = aafPassword;
+            this.params.userName = aafUsername;
+            this.params.fetchTimeout = fetchTimeout;
+            this.params.fetchLimit = fetchLimit;
+            this.params.useHttps = useHttps;
+            return this;
+        }
+
+        public TopicParamsBuilder managed(boolean managed) {
+            this.params.managed = managed;
+            return this;
+        }
+
+        public TopicParamsBuilder hostname(String hostname) {
+            this.params.hostname = hostname;
+            return this;
+        }
+
+        public TopicParamsBuilder clientName(String clientName) {
+            this.params.clientName = clientName;
+            return this;
+        }
+
+        public TopicParamsBuilder port(int port) {
+            this.params.port = port;
+            return this;
+        }
+
+        public TopicParamsBuilder basePath(String basePath) {
+            this.params.basePath = basePath;
+            return this;
+        }
+
+        public TopicParamsBuilder serializationProvider(String serializationProvider) {
+            this.params.serializationProvider = serializationProvider;
+            return this;
+        }
+
+    }
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/VesNotificationCallback.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/VesNotificationCallback.java
new file mode 100644
index 00000000..83bfcdbe
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/dmaap/VesNotificationCallback.java
@@ -0,0 +1,112 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *  Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+
+package org.onap.slice.analysis.ms.dmaap;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.onap.slice.analysis.ms.models.Configuration;
+import org.onap.slice.analysis.ms.models.vesnotification.NotificationFields;
+
+import org.onap.slice.analysis.ms.service.ccvpn.CCVPNPmDatastore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.io.IOException;
+
+/**
+ * Handles Notification on dmaap for ves notification events
+ */
+@Component
+public class VesNotificationCallback implements NotificationCallback {
+
+    private Configuration configuration;
+    private String NOTIFICATIONFIELDS = "notificationFields";
+    private String EVENT = "event";
+    private String vesNotifChangeIdentifier;
+    private String vesNotfiChangeType;
+
+    @Autowired
+    CCVPNPmDatastore ccvpnPmDatastore;
+
+    private static Logger log = LoggerFactory.getLogger(VesNotificationCallback.class);
+
+    /**
+     * init ves callback; load configuration.
+     */
+    @PostConstruct
+    public void init(){
+        configuration = Configuration.getInstance();
+        vesNotifChangeIdentifier = configuration.getVesNotifChangeIdentifier();
+        vesNotfiChangeType = configuration.getVesNotifChangeType();
+    }
+
+    /**
+     * Triggers on handleNofitication method
+     * @param msg incoming message
+     */
+    @Override
+    public void activateCallBack(String msg) {
+        handleNotification(msg);
+    }
+
+    /**
+     * Parse Performance dmaap notification and save to DB 
+     * @param msg incoming message
+     */
+    private void handleNotification(String msg) {
+        log.info("Message received from VES : {}" ,msg);
+        ObjectMapper obj = new ObjectMapper();
+        NotificationFields output = null;
+        String notifChangeIdentifier = "";
+        String notifChangeType = "";
+        String cllId = null;
+        String uniId = null;
+        String bw = null;
+        try {
+            JsonNode node = obj.readTree(msg);
+            JsonNode notificationNode = node.get(EVENT).get(NOTIFICATIONFIELDS);
+            output = obj.treeToValue(notificationNode, NotificationFields.class);
+
+            //Filter out target notification changeIdentifier and changeType
+            notifChangeIdentifier = output.getChangeIdentifier();
+            notifChangeType = output.getChangeType();
+            if (notifChangeType.equals(vesNotfiChangeType)
+            && notifChangeIdentifier.equals(vesNotifChangeIdentifier)) {
+                cllId = output.getArrayOfNamedHashMap().get(0).getHashMap().getCllId();
+                uniId = output.getArrayOfNamedHashMap().get(0).getHashMap().getUniId();
+                bw = output.getArrayOfNamedHashMap().get(0).getHashMap().getBandwidthValue();
+            }
+        }
+        catch (IOException e) {
+            log.error("Error converting VES msg to object, {}", e.getMessage());
+        }
+        if (cllId != null && uniId != null && bw != null){
+            ccvpnPmDatastore.addUsedBwToEndpoint(cllId, uniId, bw);
+        }
+
+    }
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/Configuration.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/Configuration.java
index 2f3f6c2a..d5aeb7b4 100644
--- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/Configuration.java
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/Configuration.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020-2022 Wipro Limited.
+ *   Copyright (C) 2022 Huawei Canada Limited.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -31,12 +32,16 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
+import lombok.Getter;
+import lombok.Setter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * Model class for the application Configuration
  */
+@Getter
+@Setter
 public class Configuration {
     private static Logger log = LoggerFactory.getLogger(Configuration.class);
 
@@ -65,48 +70,22 @@ public class Configuration {
     private String desUrl;
     private int pmDataDurationInWeeks;
 
+    private int vesNotifPollingInterval;
+    private String vesNotifChangeIdentifier;
+    private String vesNotifChangeType;
+    private int ccvpnEvalInterval;
+    private double ccvpnEvalThreshold;
+    private double ccvpnEvalPrecision;
+    private String aaiNotifTargetAction;
+    private String aaiNotifTargetSource;
+    private String aaiNotifTargetEntity;
+    private boolean ccvpnEvalPeriodicCheckOn;
+    private boolean ccvpnEvalOnDemandCheckOn;
+
     /**
-     * Check if topic is secure.
+     * No args constructor
      */
-    public boolean isSecured() {
-        return (aafUsername != null);
-
-    }
-
-    public String getAafUsername() {
-        return aafUsername;
-    }
-
-    public void setAafUsername(String aafUsername) {
-        this.aafUsername = aafUsername;
-    }
-
-    public String getAafPassword() {
-        return aafPassword;
-    }
-
-    public void setAafPassword(String aafPassword) {
-        this.aafPassword = aafPassword;
-    }
-
-    public Map<String, Object> getStreamsSubscribes() {
-        return streamsSubscribes;
-    }
-
-    public void setStreamsSubscribes(Map<String, Object> streamsSubscribes) {
-        this.streamsSubscribes = streamsSubscribes;
-    }
-
-    public Map<String, Object> getStreamsPublishes() {
-        return streamsPublishes;
-    }
-
-    public void setStreamsPublishes(Map<String, Object> streamsPublishes) {
-        this.streamsPublishes = streamsPublishes;
-    }
-
     protected Configuration() {
-
     }
 
     /**
@@ -119,174 +98,11 @@ public class Configuration {
         return instance;
     }
 
-    public String getCg() {
-        return cg;
-    }
-
-    public void setCg(String cg) {
-        this.cg = cg;
-    }
-
-    public String getCid() {
-        return cid;
-    }
-
-    public void setCid(String cid) {
-        this.cid = cid;
-    }
-
-    public int getPollingInterval() {
-        return pollingInterval;
-    }
-
-    public void setPollingInterval(int pollingInterval) {
-        this.pollingInterval = pollingInterval;
-    }
-
-    public int getPollingTimeout() {
-        return pollingTimeout;
-    }
-
-    public void setPollingTimeout(int pollingTimeout) {
-        this.pollingTimeout = pollingTimeout;
-    }
-
-    public String getPgHost() {
-        return pgHost;
-    }
-
-    public void setPgHost(String pgHost) {
-        this.pgHost = pgHost;
-    }
-
-    public int getPgPort() {
-        return pgPort;
-    }
-
-    public void setPgPort(int pgPort) {
-        this.pgPort = pgPort;
-    }
-
-    public String getPgUsername() {
-        return pgUsername;
-    }
-
-    public void setPgUsername(String pgUsername) {
-        this.pgUsername = pgUsername;
-    }
-
-    public String getPgPassword() {
-        return pgPassword;
-    }
-
-    public void setPgPassword(String pgPassword) {
-        this.pgPassword = pgPassword;
-    }
-
-    public List<String> getDmaapServers() {
-        return dmaapServers;
-    }
-
-    public void setDmaapServers(List<String> dmaapServers) {
-        this.dmaapServers = dmaapServers;
-    }
-
-    public String getConfigDbService() {
-        return configDbService;
-    }
-
-    public void setConfigDbService(String configDbService) {
-        this.configDbService = configDbService;
-    }
-
-    public String getCpsUrl() {
-        return cpsUrl;
-    }
-
-    public void setCpsUrl(String cpsUrl) {
-        this.cpsUrl = cpsUrl;
-    }
-
-    public String getAaiUrl() {
-        return aaiUrl;
-    }
-
-    public void setAaiUrl(String aaiUrl) {
-        this.aaiUrl = aaiUrl;
-    }
-
-    public Boolean getConfigDbEnabled() {
-        return configDbEnabled;
-    }
-
-    public void setConfigDbEnabled(Boolean configDbEnabled) {
-        this.configDbEnabled = configDbEnabled;
-    }
-
-    public int getSamples() {
-        return samples;
-    }
-
-    public void setSamples(int samples) {
-        this.samples = samples;
-    }
-
-    public int getMinPercentageChange() {
-        return minPercentageChange;
-    }
-
-    public void setMinPercentageChange(int minPercentageChange) {
-        this.minPercentageChange = minPercentageChange;
-    }
-
-    public long getInitialDelaySeconds() {
-        return initialDelaySeconds;
-    }
-
-    public void setInitialDelaySeconds(long initialDelaySeconds) {
-        this.initialDelaySeconds = initialDelaySeconds;
-    }
-
-    /**
-     * Get RannfnssiDetails TemplateId from Configuration
-     */
-    public String getRannfnssiDetailsTemplateId() {
-        return rannfnssiDetailsTemplateId;
-    }
-
-    /**
-     * Set RannfnssiDetails TemplateId
-     */
-    public void setRannfnssiDetailsTemplateId(String rannfnssiDetailsTemplateId) {
-        this.rannfnssiDetailsTemplateId = rannfnssiDetailsTemplateId;
-    }
-
-    /**
-     * Get Data Extraction Service Url
-     */
-    public String getDesUrl() {
-        return desUrl;
-    }
-
-    /**
-     * Set Data Extraction Service Url
-     */
-    public void setDesUrl(String desUrl) {
-        this.desUrl = desUrl;
-    }
-
-    /**
-     * Get duration for which PM data is to be fetched from DES
-     */
-    public int getPmDataDurationInWeeks() {
-        return pmDataDurationInWeeks;
-    }
-
     /**
-     * Set duration for which PM data is to be fetched from DES
+     * Check if topic is secure.
      */
-    public void setPmDataDurationInWeeks(int pmDataDurationInWeeks) {
-        this.pmDataDurationInWeeks = pmDataDurationInWeeks;
+    public boolean isSecured() {
+        return (aafUsername != null);
     }
 
     @Override
@@ -340,6 +156,19 @@ public class Configuration {
         desUrl = jsonObject.get("sliceanalysisms.desUrl").getAsString();
         pmDataDurationInWeeks = jsonObject.get("sliceanalysisms.pmDataDurationInWeeks").getAsInt();
 
+        vesNotifChangeIdentifier = jsonObject.get("sliceanalysisms.vesNotifChangeIdentifier").getAsString();
+        vesNotifChangeType = jsonObject.get("sliceanalysisms.vesNotifChangeType").getAsString();
+        vesNotifPollingInterval = jsonObject.get("sliceanalysisms.vesNotifPollingInterval").getAsInt();
+
+        aaiNotifTargetAction = jsonObject.get("sliceanalysisms.aaiNotif.targetAction").getAsString();
+        aaiNotifTargetSource = jsonObject.get("sliceanalysisms.aaiNotif.targetSource").getAsString();
+        aaiNotifTargetSource = jsonObject.get("sliceanalysisms.aaiNotif.targetEntity").getAsString();
+        ccvpnEvalInterval = jsonObject.get("sliceanalysisms.ccvpnEvalInterval").getAsInt();
+        ccvpnEvalThreshold = jsonObject.get("sliceanalysisms.ccvpnEvalThreshold").getAsDouble();
+        ccvpnEvalPrecision = jsonObject.get("sliceanalysisms.ccvpnEvalPrecision").getAsDouble();
+        ccvpnEvalPeriodicCheckOn = jsonObject.get("sliceanalysisms.ccvpnEvalPeriodicCheckOn").getAsBoolean();
+        ccvpnEvalOnDemandCheckOn = jsonObject.get("sliceanalysisms.ccvpnEvalOnDemandCheckOn").getAsBoolean();
+
         if (Objects.isNull(jsonObject.get("aafUsername"))) {
             aafUsername = null;
         } else {
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/AdditionalProperties.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/AdditionalProperties.java
index 24c1dd2f..30d1b357 100644
--- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/AdditionalProperties.java
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/AdditionalProperties.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020 Wipro Limited.
+ *   Copyright (C) 2022 Huawei Canada Limited.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -21,54 +22,76 @@
 
 package org.onap.slice.analysis.ms.models.policy;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
-/** 
- * Model class for the AdditionalProperties Object 
+/**
+ * Model class for the AdditionalProperties Object
  */
+@JsonInclude(JsonInclude.Include.NON_NULL)
 public class AdditionalProperties<T> {
-	private String modifyAction;
-	private List<String> snssaiList;
-	private String sliceProfileId;
-	private T resourceConfig;
-	private Map<String, String> nsiInfo;
-	private String scriptName;
-	public String getModifyAction() {
-		return modifyAction;
-	}
-	public void setModifyAction(String modifyAction) {
-		this.modifyAction = modifyAction;
-	}
-	public List<String> getSnssaiList() {
-		return snssaiList;
-	}
-	public void setSnssaiList(List<String> snssaiList) {
-		this.snssaiList = snssaiList;
-	}
-	public String getSliceProfileId() {
-		return sliceProfileId;
-	}
-	public void setSliceProfileId(String sliceProfileId) {
-		this.sliceProfileId = sliceProfileId;
-	}
-	public T getResourceConfig() {
-		return resourceConfig;
-	}
-	public void setResourceConfig(T resourceConfig) {
-		this.resourceConfig = resourceConfig;
-	}
-	public Map<String, String> getNsiInfo() {
-		return nsiInfo;
-	}
-	public void setNsiInfo(Map<String, String> nsiInfo) {
-		this.nsiInfo = nsiInfo;
-	}
-	public String getScriptName() {
-		return scriptName;
-	}
-	public void setScriptName(String scriptName) {
-		this.scriptName = scriptName;
-	}
+    private String modifyAction;
+    private List<String> snssaiList;
+    private String sliceProfileId;
+    private T resourceConfig;
+    private Map<String, String> nsiInfo;
+    private String scriptName;
+    // Extra attributes for CCVPN CloseLoop
+    private String enableSdnc;
+    private List<TransportNetwork> transportNetworks;
 
+    public String getModifyAction() {
+        return modifyAction;
+    }
+    public void setModifyAction(String modifyAction) {
+        this.modifyAction = modifyAction;
+    }
+    public List<String> getSnssaiList() {
+        return snssaiList;
+    }
+    public void setSnssaiList(List<String> snssaiList) {
+        this.snssaiList = snssaiList;
+    }
+    public String getSliceProfileId() {
+        return sliceProfileId;
+    }
+    public void setSliceProfileId(String sliceProfileId) {
+        this.sliceProfileId = sliceProfileId;
+    }
+    public T getResourceConfig() {
+        return resourceConfig;
+    }
+    public void setResourceConfig(T resourceConfig) {
+        this.resourceConfig = resourceConfig;
+    }
+    public Map<String, String> getNsiInfo() {
+        return nsiInfo;
+    }
+    public void setNsiInfo(Map<String, String> nsiInfo) {
+        this.nsiInfo = nsiInfo;
+    }
+    public String getScriptName() {
+        return scriptName;
+    }
+    public void setScriptName(String scriptName) {
+        this.scriptName = scriptName;
+    }
+    // Extra attributes setter/getter for CCVPN CloseLoop
+    public String getEnableSdnc() {
+        return enableSdnc;
+    }
+
+    public void setEnableSdnc(String enableSdnc) {
+        this.enableSdnc = enableSdnc;
+    }
+
+    public List<TransportNetwork> getTransportNetworks() {
+        return transportNetworks;
+    }
+
+    public void setTransportNetworks(List<TransportNetwork> transportNetworks) {
+        this.transportNetworks = transportNetworks;
+    }
 }
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/Payload.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/Payload.java
index d96850cb..e3351541 100644
--- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/Payload.java
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/Payload.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020 Wipro Limited.
+ *   Copyright (C) 2022 Huawei Canada Limited.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -20,64 +21,73 @@
  *******************************************************************************/
 package org.onap.slice.analysis.ms.models.policy;
 
-/** 
- * Model class for the Paylaod Object 
+/**
+ * Model class for the Paylaod Object
  */
 public class Payload {
 
-	private String name;
-	private String serviceInstanceID;
-	private String globalSubscriberId;
-	private String subscriptionServiceType;
-	private String networkType;
-	private AdditionalProperties<?> additionalProperties;
+    private String name;
+    private String serviceInstanceID;
+    private String globalSubscriberId;
+    private String subscriptionServiceType;
+    private String networkType;
+    private AdditionalProperties<?> additionalProperties;
+    private String serviceType;
 
-	public String getName() {
-		return name;
-	}
+    public String getName() {
+        return name;
+    }
 
-	public void setName(String name) {
-		this.name = name;
-	}
+    public void setName(String name) {
+        this.name = name;
+    }
 
-	public String getServiceInstanceID() {
-		return serviceInstanceID;
-	}
+    public String getServiceInstanceID() {
+        return serviceInstanceID;
+    }
 
-	public void setServiceInstanceID(String serviceInstanceId) {
-		this.serviceInstanceID = serviceInstanceId;
-	}
+    public void setServiceInstanceID(String serviceInstanceId) {
+        this.serviceInstanceID = serviceInstanceId;
+    }
 
-	public String getGlobalSubscriberId() {
-		return globalSubscriberId;
-	}
+    public String getGlobalSubscriberId() {
+        return globalSubscriberId;
+    }
 
-	public void setGlobalSubscriberId(String globalSubscriberId) {
-		this.globalSubscriberId = globalSubscriberId;
-	}
+    public void setGlobalSubscriberId(String globalSubscriberId) {
+        this.globalSubscriberId = globalSubscriberId;
+    }
 
-	public String getSubscriptionServiceType() {
-		return subscriptionServiceType;
-	}
+    public String getSubscriptionServiceType() {
+        return subscriptionServiceType;
+    }
 
-	public void setSubscriptionServiceType(String subscriptionServiceType) {
-		this.subscriptionServiceType = subscriptionServiceType;
-	}
+    public void setSubscriptionServiceType(String subscriptionServiceType) {
+        this.subscriptionServiceType = subscriptionServiceType;
+    }
 
-	public String getNetworkType() {
-		return networkType;
-	}
+    public String getNetworkType() {
+        return networkType;
+    }
 
-	public void setNetworkType(String networkType) {
-		this.networkType = networkType;
-	}
+    public void setNetworkType(String networkType) {
+        this.networkType = networkType;
+    }
 
-	public AdditionalProperties<?> getAdditionalProperties() {
-		return additionalProperties;
-	}
+    public AdditionalProperties<?> getAdditionalProperties() {
+        return additionalProperties;
+    }
 
-	public void setAdditionalProperties(AdditionalProperties<?> additionalProperties) {
-		this.additionalProperties = additionalProperties;
-	}
+    public void setAdditionalProperties(AdditionalProperties<?> additionalProperties) {
+        this.additionalProperties = additionalProperties;
+    }
+
+    public String getServiceType() {
+        return serviceType;
+    }
+
+    public void setServiceType(String serviceType) {
+        this.serviceType = serviceType;
+    }
 
 }
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/Sla.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/Sla.java
new file mode 100644
index 00000000..2d6451aa
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/Sla.java
@@ -0,0 +1,66 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *   ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+
+package org.onap.slice.analysis.ms.models.policy;
+
+/**
+ * Model class for payload.additionalProperties.transportNetworks.sla
+ */
+public class Sla {
+
+    private int latency;
+    private int maxBandwidth;
+
+    /**
+     * No args constructor for use in serialization
+     *
+     */
+    public Sla() {
+    }
+
+    /**
+     *
+     * @param maxBandwidth
+     * @param latency
+     */
+    public Sla(int latency, int maxBandwidth) {
+        super();
+        this.latency = latency;
+        this.maxBandwidth = maxBandwidth;
+    }
+
+    public int getLatency() {
+        return latency;
+    }
+
+    public void setLatency(int latency) {
+        this.latency = latency;
+    }
+
+    public int getMaxBandwidth() {
+        return maxBandwidth;
+    }
+
+    public void setMaxBandwidth(int maxBandwidth) {
+        this.maxBandwidth = maxBandwidth;
+    }
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/TransportNetwork.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/TransportNetwork.java
new file mode 100644
index 00000000..46881806
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/policy/TransportNetwork.java
@@ -0,0 +1,65 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *   ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+
+package org.onap.slice.analysis.ms.models.policy;
+
+/**
+ * Model class TransportNetwork inside payload.additionalProperties for CCVPN closeLoop
+ */
+public class TransportNetwork {
+
+    private String id;
+    private Sla sla;
+
+    /**
+     * No args constructor for use in serialization
+     *
+     */
+    public TransportNetwork() {
+    }
+
+    /**
+     *
+     * @param sla
+     * @param id
+     */
+    public TransportNetwork(String id, Sla sla) {
+        this.id = id;
+        this.sla = sla;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public Sla getSla() {
+        return sla;
+    }
+
+    public void setSla(Sla sla) {
+        this.sla = sla;
+    }
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/ArrayOfNamedHashMap.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/ArrayOfNamedHashMap.java
new file mode 100644
index 00000000..fcf7eb09
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/ArrayOfNamedHashMap.java
@@ -0,0 +1,50 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *   ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.models.vesnotification;
+
+import javax.annotation.Generated;
+
+/**
+ * Model class for Ves ccvpnNotification.arrayOfNamedHashMap
+ */
+@Generated("jsonschema2pojo")
+public class ArrayOfNamedHashMap {
+
+    private String name;
+    private HashMap hashMap;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public HashMap getHashMap() {
+        return hashMap;
+    }
+
+    public void setHashMap(HashMap hashMap) {
+        this.hashMap = hashMap;
+    }
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/HashMap.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/HashMap.java
new file mode 100644
index 00000000..e26f402c
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/HashMap.java
@@ -0,0 +1,68 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *   ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.models.vesnotification;
+
+import javax.annotation.Generated;
+
+/**
+ * Model class for Ves ccvpnNotification.arrayOfNamedHashMap.hashMap
+ */
+@Generated("jsonschema2pojo")
+public class HashMap {
+
+    private String cllId;
+    private String uniId;
+    private String bandwidthValue;
+    private String time;
+
+    public String getCllId() {
+        return cllId;
+    }
+
+    public void setCllId(String cllId) {
+        this.cllId = cllId;
+    }
+
+    public String getUniId() {
+        return uniId;
+    }
+
+    public void setUniId(String uniId) {
+        this.uniId = uniId;
+    }
+
+    public String getBandwidthValue() {
+        return bandwidthValue;
+    }
+
+    public void setBandwidthValue(String bandwidthValue) {
+        this.bandwidthValue = bandwidthValue;
+    }
+
+    public String getTime() {
+        return time;
+    }
+
+    public void setTime(String time) {
+        this.time = time;
+    }
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/NotificationFields.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/NotificationFields.java
new file mode 100644
index 00000000..28509a6b
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/models/vesnotification/NotificationFields.java
@@ -0,0 +1,70 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *   ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.models.vesnotification;
+
+import java.util.ArrayList;
+import java.util.List;
+import javax.annotation.Generated;
+
+/**
+ * Model class for Ves notification message
+ */
+@Generated("jsonschema2pojo")
+public class NotificationFields {
+
+    private String changeIdentifier;
+    private String changeType;
+    private String notificationFieldsVersion;
+    private List<ArrayOfNamedHashMap> arrayOfNamedHashMap = new ArrayList<ArrayOfNamedHashMap>();
+
+    public String getChangeIdentifier() {
+        return changeIdentifier;
+    }
+
+    public void setChangeIdentifier(String changeIdentifier) {
+        this.changeIdentifier = changeIdentifier;
+    }
+
+    public String getChangeType() {
+        return changeType;
+    }
+
+    public void setChangeType(String changeType) {
+        this.changeType = changeType;
+    }
+
+    public String getNotificationFieldsVersion() {
+        return notificationFieldsVersion;
+    }
+
+    public void setNotificationFieldsVersion(String notificationFieldsVersion) {
+        this.notificationFieldsVersion = notificationFieldsVersion;
+    }
+
+    public List<ArrayOfNamedHashMap> getArrayOfNamedHashMap() {
+        return arrayOfNamedHashMap;
+    }
+
+    public void setArrayOfNamedHashMap(List<ArrayOfNamedHashMap> arrayOfNamedHashMap) {
+        this.arrayOfNamedHashMap = arrayOfNamedHashMap;
+    }
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PolicyService.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PolicyService.java
index 2d3a2df0..297683b4 100644
--- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PolicyService.java
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PolicyService.java
@@ -3,6 +3,7 @@
  *  slice-analysis-ms
  *  ================================================================================
  *   Copyright (C) 2020-2021 Wipro Limited.
+ *   Copyright (C) 2022 Huawei Canada Limited.
  *   ==============================================================================
  *     Licensed under the Apache License, Version 2.0 (the "License");
  *     you may not use this file except in compliance with the License.
@@ -35,6 +36,9 @@ import org.onap.slice.analysis.ms.models.policy.AAI;
 import org.onap.slice.analysis.ms.models.policy.AdditionalProperties;
 import org.onap.slice.analysis.ms.models.policy.OnsetMessage;
 import org.onap.slice.analysis.ms.models.policy.Payload;
+import org.onap.slice.analysis.ms.models.policy.Sla;
+import org.onap.slice.analysis.ms.models.policy.TransportNetwork;
+import org.onap.slice.analysis.ms.service.ccvpn.RequestOwner;
 import org.onap.slice.analysis.ms.utils.DmaapUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,77 +46,149 @@ import org.springframework.stereotype.Component;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+/**
+ * Serivce to generate and publish onsetMessage to ONAP/Policy
+ */
 @Component
 public class PolicyService {
-	private PolicyDmaapClient policyDmaapClient;
-	private static Logger log = LoggerFactory.getLogger(PolicyService.class);
-	private ObjectMapper objectMapper = new ObjectMapper();
-
-	/**
-	 * Initialization
-	 */
-	@PostConstruct
-	public void init() {
-		Configuration configuration = Configuration.getInstance();
-		policyDmaapClient = new PolicyDmaapClient(new DmaapUtils(), configuration);
-	}
-	
-	protected <T> OnsetMessage formPolicyOnsetMessage(String snssai, AdditionalProperties<T> addProps, Map<String, String> serviceDetails) {
-		OnsetMessage onsetmsg = new OnsetMessage();
-		Payload payload = new Payload();
-		payload.setGlobalSubscriberId(serviceDetails.get("globalSubscriberId"));
-		payload.setSubscriptionServiceType(serviceDetails.get("subscriptionServiceType"));
-		payload.setNetworkType("AN");
-		payload.setName(serviceDetails.get("ranNFNSSIId"));
-		payload.setServiceInstanceID(serviceDetails.get("ranNFNSSIId"));
-
-		addProps.setModifyAction("");
-		Map<String, String> nsiInfo = new HashMap<>();
-		nsiInfo.put("nsiId", UUID.randomUUID().toString());
-		nsiInfo.put("nsiName", "");
-		addProps.setNsiInfo(nsiInfo);
-		addProps.setScriptName("AN");
-		addProps.setSliceProfileId(serviceDetails.get("sliceProfileId"));
-		addProps.setModifyAction("reconfigure");
-		List<String> snssaiList = new ArrayList<>();
-		snssaiList.add(snssai);
-		addProps.setSnssaiList(snssaiList);
-
-		payload.setAdditionalProperties(addProps);
-		try {
-			onsetmsg.setPayload(objectMapper.writeValueAsString(payload));
-		} catch (Exception e) {
-			log.error("Error while mapping payload as string , {}",e.getMessage());
-		}
-
-		onsetmsg.setClosedLoopControlName("ControlLoop-Slicing-116d7b00-dbeb-4d03-8719-d0a658fa735b");
-		onsetmsg.setClosedLoopAlarmStart(System.currentTimeMillis());
-		onsetmsg.setClosedLoopEventClient("microservice.sliceAnalysisMS");
-		onsetmsg.setClosedLoopEventStatus("ONSET");
-		onsetmsg.setRequestID(UUID.randomUUID().toString());
-		onsetmsg.setTarget("generic-vnf.vnf-id");
-		onsetmsg.setTargetType("VNF");
-		onsetmsg.setFrom("DCAE");
-		onsetmsg.setVersion("1.0.2");
-		AAI aai = new AAI();
-		aai.setVserverIsClosedLoopDisabled("false");
-		aai.setVserverProvStatus("ACTIVE");
-		aai.setvServerVNFId(serviceDetails.get("ranNFNSSIId"));
-		onsetmsg.setAai(aai); 
-		return onsetmsg;
-	}
-
-	protected <T> void sendOnsetMessageToPolicy(String snssai, AdditionalProperties<T> addProps, Map<String, String> serviceDetails) {
-		OnsetMessage onsetMessage = formPolicyOnsetMessage(snssai, addProps, serviceDetails);
-		String msg =  "";
-		try { 
-			msg = objectMapper.writeValueAsString(onsetMessage);
-			log.info("Policy onset message for S-NSSAI: {} is {}", snssai, msg);
-			policyDmaapClient.sendNotificationToPolicy(msg);
-		} 
-		catch (Exception e) { 
-			log.error("Error sending notification to policy, {}",e.getMessage());
-		} 		
-	}
+    private PolicyDmaapClient policyDmaapClient;
+    private static Logger log = LoggerFactory.getLogger(PolicyService.class);
+    private ObjectMapper objectMapper = new ObjectMapper();
 
+    /**
+     * Initialization
+     */
+    @PostConstruct
+    public void init() {
+        Configuration configuration = Configuration.getInstance();
+        policyDmaapClient = new PolicyDmaapClient(new DmaapUtils(), configuration);
+    }
+
+    protected <T> OnsetMessage formPolicyOnsetMessage(String snssai, AdditionalProperties<T> addProps, Map<String, String> serviceDetails) {
+        OnsetMessage onsetmsg = new OnsetMessage();
+        Payload payload = new Payload();
+        payload.setGlobalSubscriberId(serviceDetails.get("globalSubscriberId"));
+        payload.setSubscriptionServiceType(serviceDetails.get("subscriptionServiceType"));
+        payload.setNetworkType("AN");
+        payload.setName(serviceDetails.get("ranNFNSSIId"));
+        payload.setServiceInstanceID(serviceDetails.get("ranNFNSSIId"));
+
+        addProps.setModifyAction("");
+        Map<String, String> nsiInfo = new HashMap<>();
+        nsiInfo.put("nsiId", UUID.randomUUID().toString());
+        nsiInfo.put("nsiName", "");
+        addProps.setNsiInfo(nsiInfo);
+        addProps.setScriptName("AN");
+        addProps.setSliceProfileId(serviceDetails.get("sliceProfileId"));
+        addProps.setModifyAction("reconfigure");
+        List<String> snssaiList = new ArrayList<>();
+        snssaiList.add(snssai);
+        addProps.setSnssaiList(snssaiList);
+
+        payload.setAdditionalProperties(addProps);
+        try {
+            onsetmsg.setPayload(objectMapper.writeValueAsString(payload));
+        } catch (Exception e) {
+            log.error("Error while mapping payload as string , {}",e.getMessage());
+        }
+
+        onsetmsg.setClosedLoopControlName("ControlLoop-Slicing-116d7b00-dbeb-4d03-8719-d0a658fa735b");
+        onsetmsg.setClosedLoopAlarmStart(System.currentTimeMillis());
+        onsetmsg.setClosedLoopEventClient("microservice.sliceAnalysisMS");
+        onsetmsg.setClosedLoopEventStatus("ONSET");
+        onsetmsg.setRequestID(UUID.randomUUID().toString());
+        onsetmsg.setTarget("generic-vnf.vnf-id");
+        onsetmsg.setTargetType("VNF");
+        onsetmsg.setFrom("DCAE");
+        onsetmsg.setVersion("1.0.2");
+        AAI aai = new AAI();
+        aai.setVserverIsClosedLoopDisabled("false");
+        aai.setVserverProvStatus("ACTIVE");
+        aai.setvServerVNFId(serviceDetails.get("ranNFNSSIId"));
+        onsetmsg.setAai(aai);
+        return onsetmsg;
+    }
+
+    protected <T> void sendOnsetMessageToPolicy(String snssai, AdditionalProperties<T> addProps, Map<String, String> serviceDetails) {
+        OnsetMessage onsetMessage = formPolicyOnsetMessage(snssai, addProps, serviceDetails);
+        String msg =  "";
+        try {
+            msg = objectMapper.writeValueAsString(onsetMessage);
+            log.info("Policy onset message for S-NSSAI: {} is {}", snssai, msg);
+            policyDmaapClient.sendNotificationToPolicy(msg);
+        }
+        catch (Exception e) {
+            log.error("Error sending notification to policy, {}",e.getMessage());
+        }
+    }
+
+    /**
+     * Generate onsetMessage for ccvpn service update operation
+     * @param cllId cloud leased line Id (ethernet service id)
+     * @param newBw new bandwidth value for bandwidth adjustment
+     * @param <T> type for additionalPropert
+     *           ies, can be omitted
+     * @return OnsetMessage result
+     */
+    public <T> OnsetMessage formPolicyOnsetMessageForCCVPN(String cllId, Integer newBw, RequestOwner owner) {
+        Sla sla = new Sla(2, newBw);
+        String transportNetworkId = cllId;
+        if (owner == RequestOwner.UUI) {
+            transportNetworkId += "-network-001";
+        } else if (owner == RequestOwner.DCAE) {
+            transportNetworkId += "-network-002";
+        }
+        TransportNetwork transportNetwork = new TransportNetwork(transportNetworkId, sla);
+        AdditionalProperties additionalProperties = new AdditionalProperties();
+        additionalProperties.setModifyAction("bandwidth");
+        additionalProperties.setEnableSdnc("true");
+        List<TransportNetwork> transportNetworks = new ArrayList();
+        transportNetworks.add(transportNetwork);
+        additionalProperties.setTransportNetworks(transportNetworks);
+
+        Payload payload = new Payload();
+        payload.setGlobalSubscriberId("IBNCustomer");
+        payload.setSubscriptionServiceType("IBN");
+        payload.setServiceType("CLL");
+        payload.setName("cloud-leased-line-101");
+        payload.setServiceInstanceID(cllId);
+        payload.setAdditionalProperties(additionalProperties);
+
+        OnsetMessage onsetmsg = new OnsetMessage();
+        try {
+            onsetmsg.setPayload(objectMapper.writeValueAsString(payload));
+        } catch (Exception e) {
+            log.error("Error while mapping payload as string , {}",e.getMessage());
+        }
+        onsetmsg.setClosedLoopControlName("ControlLoop-CCVPN-CLL-227e8b00-dbeb-4d03-8719-d0a658fb846c");
+        onsetmsg.setClosedLoopAlarmStart(System.currentTimeMillis());
+        onsetmsg.setClosedLoopEventClient("microservice.sliceAnalysisMS");
+        onsetmsg.setClosedLoopEventStatus("ONSET");
+        onsetmsg.setRequestID(UUID.randomUUID().toString());
+        onsetmsg.setTarget("generic-vnf.vnf-id");
+        onsetmsg.setTargetType("VNF");
+        onsetmsg.setFrom("DCAE");
+        onsetmsg.setVersion("1.0.2");
+        AAI aai = new AAI();
+        aai.setVserverIsClosedLoopDisabled("true");
+        onsetmsg.setAai(aai);
+        return onsetmsg;
+    }
+
+    /**
+     * Sending the onsetMessage to Onap-Policy through PolicyDmaapClient
+     * @param onsetMessage the onsetMessage about to send
+     * @param <T> type inherent from previous implementation can be omitted
+     */
+    public <T> void sendOnsetMessageToPolicy(OnsetMessage onsetMessage){
+        String msg =  "";
+        try {
+            msg = objectMapper.writeValueAsString(onsetMessage);
+            log.info("Policy onset message for ControlLoop-CCVPN-CLL is {}", msg);
+            policyDmaapClient.sendNotificationToPolicy(msg);
+        }
+        catch (Exception e) {
+            log.error("Error sending notification to policy, {}",e.getMessage());
+        }
+    }
 }
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/BandwidthEvaluator.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/BandwidthEvaluator.java
new file mode 100644
index 00000000..a7847b0e
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/BandwidthEvaluator.java
@@ -0,0 +1,324 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.service.ccvpn;
+
+import com.google.gson.JsonObject;
+import lombok.NonNull;
+import org.onap.slice.analysis.ms.aai.AaiService;
+
+import org.onap.slice.analysis.ms.models.Configuration;
+import org.onap.slice.analysis.ms.service.PolicyService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+
+/**
+ * This class implements the CCVPN PM Closed-loop logical function.
+ * A simple actor model design is implemented here.
+ */
+@Component
+public class BandwidthEvaluator {
+    private static Logger log = LoggerFactory.getLogger(BandwidthEvaluator.class);
+    private Configuration configuration;
+
+    @Autowired
+    AaiService aaiService;
+
+    @Autowired
+    CCVPNPmDatastore ccvpnPmDatastore;
+
+    @Autowired
+    PolicyService policyService;
+
+    private Loop evaluationEventLoop;
+    private Loop aaiEventLoop;
+
+    private static final Event KILL_PILL = new SimpleEvent(null, 0);
+    private static final int DEFAULT_EVAL_INTERVAL = 5;
+    private static final String SERVICE_INSTANCE_LOCATION_ID = "service-instance-location-id";
+    private static final String BANDWIDTH_TOTAL = "bandwidth-total";
+
+    /**
+     * Interval of each round of evaluation, defined in config_all.json
+     */
+    private static int evaluationInterval;
+
+    /**
+     * Percentage threshold of bandwidth adjustment.
+     */
+    private static double threshold;
+
+    /**
+     * Precision of bandwidth evaluation and adjustment.
+     */
+    private static double precision; // in Mbps;
+    private final ScheduledExecutorService executorPool = Executors.newScheduledThreadPool(1);
+
+    /**
+     * Initialize and start the bandwidth evaluator process, schedule a periodic service bandwidth usage check
+     */
+    @PostConstruct
+    public void init() {
+        loadConfig();
+        /**
+         * Evalution main loop
+         */
+        evaluationEventLoop = new Loop("EvaluationLoop"){
+            @Override
+            public void process(Event event) {
+                if (event.type() == SimpleEvent.Type.PERIODIC_CHECK && isPeriodicCheckOn()){
+                    log.info("Received new periodic check request: {}", event.time());
+                    Map<Endpointkey, CCVPNPmDatastore.EvictingQueue<Integer>> usedBwMap = ccvpnPmDatastore.getUsedBwMap();
+                    Map<String, Integer> candidate = new TreeMap<>();
+                    for(Map.Entry<Endpointkey, CCVPNPmDatastore.EvictingQueue<Integer>> entry: usedBwMap.entrySet()) {
+                        String serviceId = entry.getKey().getCllId();
+                        Object[] usedBws = entry.getValue().tryReadToArray();
+
+                        if (usedBws == null) {
+                            // No enough data for evaluating
+                            continue;
+                        }
+                        if (ccvpnPmDatastore.getMaxBwOfSvc(serviceId) == 0) {
+                            // Max bandwidth not cached yet
+                            post(new SimpleEvent(SimpleEvent.Type.AAI_BW_REQ, serviceId));
+                            continue;
+                        }
+                        double avg = Arrays.stream(usedBws)
+                                .mapToInt(o -> (int) o)
+                                .summaryStatistics()
+                                .getAverage();
+                        if (needAdjust(serviceId, avg, ccvpnPmDatastore.getMaxBwOfSvc(serviceId))) {
+                            int newBw = (int) (Math.ceil((avg / threshold) * 1.2 / precision) * precision);
+                            candidate.put(serviceId, Math.max(candidate.getOrDefault(serviceId, 0), newBw));
+                        }
+                    }
+                    for(Map.Entry<String, Integer> entry: candidate.entrySet()) {
+                        if (isServiceUnderMaintenance(entry.getKey())) {
+                            post(new SimpleEvent(SimpleEvent.Type.AAI_BW_REQ, entry.getKey()));
+                            continue;
+                        }
+                        ccvpnPmDatastore.updateSvcState(entry.getKey(), ServiceState.UNDER_MAINTENANCE);
+                        sendModifyRequest(entry.getKey(), entry.getValue(), RequestOwner.DCAE);
+                    }
+
+                } else if (event.type() == SimpleEvent.Type.ONDEMAND_CHECK && isOnDemandCheckOn()) {
+                    log.info("Received new on-demand check request: {}", event.time());
+                    JsonObject payload = (JsonObject) event.subject();
+                    String serviceId = payload.get(SERVICE_INSTANCE_LOCATION_ID).getAsString();
+                    if (!isServiceUnderMaintenance(serviceId)){
+                        int newBandwidth = payload.get(BANDWIDTH_TOTAL).getAsInt();
+                        Map<String, Integer> maxBandwidthData = aaiService.fetchMaxBandwidthOfService(serviceId);
+                        int oldBandwidth = maxBandwidthData.get("maxBandwidth");
+                        if (newBandwidth != oldBandwidth) {
+                            ccvpnPmDatastore.updateSvcState(serviceId, ServiceState.UNDER_MAINTENANCE);
+                            sendModifyRequest(serviceId, newBandwidth, RequestOwner.UUI);
+                        }
+                    }
+                }
+            }
+
+            private void sendModifyRequest(String cllId, Integer newBandwidth, RequestOwner owner) {
+                policyService.sendOnsetMessageToPolicy(
+                        policyService.formPolicyOnsetMessageForCCVPN(cllId, newBandwidth, owner)
+                );
+            }
+
+            private boolean needAdjust(String serivceId, double currentAverageUsage, int maxBandwidth){
+                return currentAverageUsage > threshold * maxBandwidth;
+            }
+
+            private boolean isServiceUnderMaintenance(String serivceId) {
+                return ccvpnPmDatastore.getStatusOfSvc(serivceId) == ServiceState.UNDER_MAINTENANCE;
+            }
+        };
+
+        /**
+         * AAI data consumer loop
+         */
+        aaiEventLoop = new Loop("AAIEventLoop"){
+            @Override
+            public void process(Event event) {
+                if (event.type() == SimpleEvent.Type.AAI_BW_REQ){
+                    log.info("Received new AAI network policy query at: {}", event.time());
+                    String serviceId = (String) event.subject();
+                    Map<String, Integer> maxBandwidthData = aaiService.fetchMaxBandwidthOfService(serviceId);
+                    int bwVal = maxBandwidthData.get("maxBandwidth");
+                    if (maxBandwidthData != null){
+                        if (ccvpnPmDatastore.getMaxBwOfSvc(serviceId) == 0){
+                            ccvpnPmDatastore.updateMaxBw(serviceId, bwVal, true);
+                        } else if (ccvpnPmDatastore.getMaxBwOfSvc(serviceId) != bwVal) {
+                            ccvpnPmDatastore.updateMaxBw(serviceId, bwVal, true);
+                            ccvpnPmDatastore.updateSvcState(serviceId, ServiceState.RUNNING);
+                        }
+                    }
+                }
+            }
+        };
+        scheduleEvaluation();
+    }
+
+    /**
+     * Stop the bandwidth evaluator process including two actors and periodic usage check
+     */
+    @PreDestroy
+    public void stop(){
+        stopScheduleEvaluation();
+        aaiEventLoop.stop();
+        evaluationEventLoop.stop();
+    }
+
+    /**
+     * Start to schedule periodic usage check at fixed rate
+     */
+    private void scheduleEvaluation(){
+        executorPool.scheduleAtFixedRate(new Runnable() {
+                @Override
+                public void run() {
+                    post(new SimpleEvent(SimpleEvent.Type.PERIODIC_CHECK, 1));
+                }
+            }, 0, (evaluationInterval == 0? DEFAULT_EVAL_INTERVAL : evaluationInterval), TimeUnit.SECONDS);
+    }
+
+    /**
+     * Stop periodic bandwidth usage check
+     */
+    private void stopScheduleEvaluation(){
+        executorPool.shutdownNow();
+    }
+
+    /**
+     * Post/broadcast event between Loops
+     * @param event event object
+     */
+    public void post(@NonNull Event event){
+        if (event.type() == SimpleEvent.Type.AAI_BW_REQ) {
+            aaiEventLoop.add(event);
+        } else if (event.type() == SimpleEvent.Type.PERIODIC_CHECK) {
+            evaluationEventLoop.add(event);
+        } else if (event.type() == SimpleEvent.Type.ONDEMAND_CHECK) {
+            evaluationEventLoop.add(event);
+        }
+    }
+
+    private void loadConfig() {
+        configuration = Configuration.getInstance();
+        evaluationInterval = configuration.getCcvpnEvalInterval();
+        threshold = configuration.getCcvpnEvalThreshold();
+        precision = configuration.getCcvpnEvalPrecision(); // in Mbps;
+    }
+
+    private boolean isPeriodicCheckOn() {
+        configuration = Configuration.getInstance();
+        return configuration.isCcvpnEvalPeriodicCheckOn();
+    }
+
+    private boolean isOnDemandCheckOn() {
+        configuration = Configuration.getInstance();
+        return configuration.isCcvpnEvalOnDemandCheckOn();
+    }
+
+    /**
+     * Inner loop implementation. Each loop acts like an actor.
+     */
+    private abstract class Loop implements Runnable {
+        private final String name;
+        private volatile boolean running;
+        private final BlockingQueue<Event> eventsQueue;
+        private final ExecutorService executor;
+        private volatile Future<?> dispatchFuture;
+
+        /**
+         * Constructor that accepts a loop name
+         * @param name name of this loop
+         */
+        Loop(String name){
+            this.name = name;
+            executor = Executors.newSingleThreadExecutor();
+            eventsQueue = new LinkedBlockingQueue<>();
+            dispatchFuture = executor.submit(this);
+        }
+
+        /**
+         * Add new event to this loop
+         * @param evt Event
+         * @return true
+         */
+        public boolean add(Event evt) {
+            return eventsQueue.add(evt);
+        }
+
+        /**
+         * Running loop that process event accordingly
+         */
+        @Override
+        public void run(){
+            running = true;
+            log.info("BandwidthEvaluator -- {} initiated", this.name);
+            while (running){
+                try{
+                    Event event = eventsQueue.take();
+                    if (event == KILL_PILL){
+                        break;
+                    }
+                    process(event);
+                } catch (InterruptedException e){
+                    log.warn("Process loop interrupted");
+                } catch (Exception | Error e){
+                    log.warn("Process loop hit an error {}", e.getMessage());
+                }
+            }
+        }
+
+        /**
+         * Operation defined by subclass for different event processing
+         * @param event incoming event
+         */
+        abstract public void process(Event event);
+
+        /**
+         * Stop this loop
+         */
+        public void stop(){
+            running = false;
+            add(KILL_PILL);
+        }
+    }
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/CCVPNPmDatastore.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/CCVPNPmDatastore.java
new file mode 100644
index 00000000..9c86f6e7
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/CCVPNPmDatastore.java
@@ -0,0 +1,263 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.service.ccvpn;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayDeque;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+/**
+ * This class represents the data structure for storing the CCVPN pm data;
+ */
+@Component
+public class CCVPNPmDatastore {
+
+    private static Logger log = LoggerFactory.getLogger(CCVPNPmDatastore.class);
+    private static final Pattern pattern = Pattern.compile("([0-9.]+)\\s*(kb|Kb|mb|Mb|Gb|gb)*");
+    private static final int WINDOW_SIZE = 5;
+    private final ConcurrentMap<String, ServiceState> svcStatus = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, Integer> endpointToMaxBw = new ConcurrentHashMap<>();
+    private final ConcurrentMap<Endpointkey, EvictingQueue<Integer>> endpointToUsedBw = new ConcurrentHashMap<>();
+
+    /**
+     * Given a cllId, return a map between Endpointkey and their corresponding UsedBw Queue.
+     * All Endpoints belongs to this same service
+     * @param cllId target cll instance id
+     * @return a filtered map contains used bandwidth data of endpointkeys whose cllId equals to the given one.
+     */
+    public Map<Endpointkey, EvictingQueue<Integer>> getUsedBwOfSvc(String cllId){
+        return endpointToUsedBw.entrySet().stream()
+                .filter(map -> map.getKey().getCllId() == cllId)
+                .collect(Collectors.toMap(map -> map.getKey(), map -> map.getValue()));
+    }
+
+    /**
+     * Return the complete used bandwidth map.
+     * @return a complete endpoint to bandwidth data map
+     */
+    public Map<Endpointkey, EvictingQueue<Integer>> getUsedBwMap(){
+        return endpointToUsedBw;
+    }
+
+    /**
+     * Return max bandwidth of cll service. If max bandwidth is null or missing, return 0;
+     * @param cllId target cll instance id
+     * @return Integer bandwidth value
+     */
+    public Integer getMaxBwOfSvc(String cllId){
+        return endpointToMaxBw.getOrDefault(cllId, 0);
+    }
+
+    /**
+     * Get Service status of this cll service
+     * @param cllId target cll instance id
+     * @return ServiceState of this cll
+     */
+    public ServiceState getStatusOfSvc(String cllId){
+        return svcStatus.getOrDefault(cllId, ServiceState.UNKNOWN);
+    }
+
+    /**
+     * return the complete map of cll service status
+     * @return complete map of serviceStatusMap
+     */
+    public ConcurrentMap<String, ServiceState> getSvcStatusMap(){
+        return svcStatus;
+    }
+
+    /**
+     * Override the service status to provided state
+     * @param cllId target cll instance id
+     * @param state new state
+     */
+    public void updateSvcState(String cllId, ServiceState state){
+        svcStatus.put(cllId, state);
+    }
+
+    /**
+     * Update max bandwidth value to given bandwidth string
+     * @param cllId target cll instance id
+     * @param bw new bandwidth
+     */
+    public void updateMaxBw(String cllId, String bw){
+        double bwvvaldb = Double.parseDouble(bw);
+        int bwvval = (int) bwvvaldb;
+        updateMaxBw(cllId, bwvval, false);
+    }
+
+    /**
+     * Update max bandwidth to given bandwidth value;
+     * if @param{override} is false, only write the bandwidth if it is absent.
+     * Otherwise override the old value no matter if it exists or not
+     * Also, when @param{override} is true, compare the provided value with the old value, if equals, return false;
+     * otherwise, return true;
+     * @param cllId target cll instance id
+     * @param bw new bandwidth int value in Mbps
+     * @param override override old value or not
+     * @return whether bandwidth value is changed or not.
+     */
+    public boolean updateMaxBw(String cllId, int bw, boolean override){
+        ;
+        if ( endpointToMaxBw.putIfAbsent(cllId, bw) == null || !override){
+            return true;
+        } else {
+            if (endpointToMaxBw.get(cllId) == bw){
+                return false;
+            } else {
+                endpointToMaxBw.replace(cllId, bw);
+                return true;
+            }
+        }
+    }
+
+    /**
+     * Append the latest bandwidth data to associated endpoint
+     * @param cllId target cll instance id
+     * @param uniId target uni id
+     * @param bw latest bandwidth usage data
+     */
+    public void addUsedBwToEndpoint(String cllId, String uniId, String bw){
+        Endpointkey enk = new Endpointkey(cllId, uniId);
+        Matcher matcher = pattern.matcher(bw.trim());
+        //Default input bw unit is kbps;
+        String unit = null;
+        // Bw in Mbps;
+        int result = 0;
+        if (matcher.find()) {
+            unit = matcher.group(2);
+            if (unit == null || unit.isEmpty() || unit.toLowerCase().equals("kb")) {
+                double val = Double.parseDouble(matcher.group(1));
+                result = (int) Math.ceil((double) val / (int) 1000 ) ;
+            } else if (unit.toLowerCase().equals("mb")){
+                double val = Double.parseDouble(matcher.group(1));
+                result = (int) val ;
+            } else if (unit.toLowerCase().equals("gb")){
+                double val = Double.parseDouble(matcher.group(1));
+                result = (int) val * (int) 1000;
+            }
+        } else {
+            log.warn("Illigal bw string: " + bw);
+        }
+
+        EvictingQueue<Integer> dataq = new EvictingQueue<Integer>(WINDOW_SIZE);
+        dataq.offer(result);
+        EvictingQueue q = endpointToUsedBw.putIfAbsent(enk, dataq);
+        if (q != null) {
+            q.offer(result);
+        }
+
+    }
+
+    /**
+     * Copy the used bandwidth queue of specified cllId:uniId to an array and return;
+     * @param cllId target cll id
+     * @param uniId target uni id
+     * @return Object[] contains all the used bandwidth data
+     */
+    public Object[] readToArray(String cllId, String uniId){
+        return endpointToUsedBw.get(new Endpointkey(cllId, uniId)).tryReadToArray();
+    }
+
+    /**
+     * Inner data structure is logically similar to circular buffer, thread-safe through blocking
+     * @param <E> Generic type of data
+     */
+    public class EvictingQueue<E> {
+        private final Queue<E> delegate;
+        final int maxSize;
+
+        /**
+         * Constructor accept a maxsize param
+         * @param maxSize max size
+         */
+        EvictingQueue(int maxSize){
+            if (maxSize < 0){
+                throw new IllegalArgumentException("Invalid maxsize for initializing EvictingQueue");
+            }
+            this.delegate = new ArrayDeque<>(maxSize);
+            this.maxSize = maxSize;
+        }
+
+        /**
+         * Adding new data to this queue
+         * @param e new data
+         * @return true
+         */
+        public synchronized boolean offer(E e){
+            return add(e);
+        }
+
+        /**
+         * Try copy data to an array and return, only if data has filled up the whole queue
+         * Otherwise, return null
+         * @return the data array
+         */
+        public synchronized Object[] tryReadToArray(){
+            if (remainingCapacity() > 0){
+                return null;
+            }
+            return toArray();
+        }
+
+        /**
+         * Return the size of this queue, and number of data added. It is no larger than the max capacity.
+         * @return int value of output
+         */
+        public int size(){
+            return delegate.size();
+        }
+
+        /**
+         * return the remaining capacity of this queue
+         * @return int value of output
+         */
+        public int remainingCapacity(){
+            return maxSize - size();
+        }
+
+        private Object[] toArray(){
+            return delegate.toArray();
+        }
+
+        private boolean add(E e){
+            if(null == e){
+                throw new IllegalArgumentException("Invalid new item in add method");
+            }
+            if (maxSize == 0){
+                return true;
+            }
+            if (size() == maxSize){
+                delegate.remove();
+            }
+            delegate.add(e);
+            return true;
+        }
+    }
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/Endpointkey.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/Endpointkey.java
new file mode 100644
index 00000000..f6c1a8c1
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/Endpointkey.java
@@ -0,0 +1,75 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.service.ccvpn;
+
+import java.util.Objects;
+
+/**
+ * Endpoint key class represents each uni associated with each cll
+ */
+public class Endpointkey {
+
+    private final String cllId;
+    private final String uniId;
+
+    /**
+     * Constructor accpets cllId and uniId.
+     * @param cllId String cll instance id
+     * @param uniId String uni id
+     */
+    public Endpointkey(String cllId, String uniId){
+        this.cllId = cllId;
+        this.uniId = uniId;
+    }
+
+    /**
+     * Return cllId
+     * @return String cllId
+     */
+    public String getCllId() {
+        return cllId;
+    }
+
+    /**
+     * Return uniId
+     * @return String uni id
+     */
+    public String getUniId() {
+        return uniId;
+    }
+
+    @Override
+    public int hashCode() { return Objects.hash(cllId, uniId); }
+
+    @Override
+    public boolean equals(Object obj){
+        if (this == obj){
+            return true;
+        }
+        if (obj instanceof Endpointkey){
+            final Endpointkey other = (Endpointkey) obj;
+            return Objects.equals(this.cllId, other.cllId) &&
+                    Objects.equals(this.uniId, other.uniId);
+        }
+        return false;
+    }
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/Event.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/Event.java
new file mode 100644
index 00000000..c36e4d86
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/Event.java
@@ -0,0 +1,37 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.service.ccvpn;
+
+/**
+ * Event (CCVPN) interface;
+ * It is the message entity inside CCVPN Closed-loop.
+ * @param <T> message type
+ * @param <S> message paylaod
+ */
+public interface Event<T extends Enum, S> {
+
+    long time();
+
+    T type();
+
+    S subject();
+
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/RequestOwner.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/RequestOwner.java
new file mode 100644
index 00000000..319032ad
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/RequestOwner.java
@@ -0,0 +1,30 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *  Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+
+package org.onap.slice.analysis.ms.service.ccvpn;
+
+/**
+ * Modification request owner
+ */
+public enum RequestOwner {
+    DCAE,
+    UUI
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/ServiceState.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/ServiceState.java
new file mode 100644
index 00000000..b4d358f5
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/ServiceState.java
@@ -0,0 +1,32 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *  Copyright (C) 2022 Huawei Canada Limited.
+ *  ==============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *     ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.service.ccvpn;
+
+/**
+ * Enum class represents possible state of a Ethernet service (designed for CCVPN CLL serivce)
+ */
+public enum ServiceState {
+    RUNNING,
+    UNDER_MAINTENANCE,
+    ERROR,
+    DELETED,
+    UNKNOWN
+}
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/SimpleEvent.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/SimpleEvent.java
new file mode 100644
index 00000000..8277380d
--- /dev/null
+++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ccvpn/SimpleEvent.java
@@ -0,0 +1,98 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  slice-analysis-ms
+ *  ================================================================================
+ *   Copyright (C) 2022 Huawei Canada Limited.
+ *  =============================================================================
+ *     Licensed under the Apache License, Version 2.0 (the "License");
+ *     you may not use this file except in compliance with the License.
+ *     You may obtain a copy of the License at
+ *
+ *          http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *     Unless required by applicable law or agreed to in writing, software
+ *     distributed under the License is distributed on an "AS IS" BASIS,
+ *     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *     See the License for the specific language governing permissions and
+ *     limitations under the License.
+ *  ============LICENSE_END=========================================================
+ *
+ *******************************************************************************/
+package org.onap.slice.analysis.ms.service.ccvpn;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+
+/**
+ * SimpleEvent a generic class implements Event (CCVPN) interface;
+ * It is the message entity inside CCVPN Closed-loop
+ *
+ * @param <T> message type
+ * @param <S> message paylaod
+ */
+public class SimpleEvent<T extends Enum, S> implements Event<T, S> {
+    private final T type;
+    private final S subject;
+    private final long time;
+
+    /**
+     * All the event types
+     */
+    public enum Type {
+        PERIODIC_CHECK,
+        ONDEMAND_CHECK,
+        AAI_BW_REQ
+    }
+
+    /**
+     * Event contructor
+     * @param type event type
+     * @param subject event content
+     */
+    public SimpleEvent(T type, S subject) {
+        this.type = type;
+        this.subject = subject;
+        this.time = System.currentTimeMillis();
+    }
+
+    /**
+     * Return the epoch time of this event happened.
+     * @return long value of epoch time
+     */
+    @Override
+    public long time() {
+        return time;
+    }
+
+    /**
+     * Return the type of this event.
+     * @return event type
+     */
+    @Override
+    public T type() {
+        return type;
+    }
+
+    /**
+     * Return the subject of this event
+     * @return event content
+     */
+    @Override
+    public S subject() {
+        return subject;
+    }
+
+    /**
+     * toString method
+     * @return toString
+     */
+    @Override
+    public String toString() {
+        return new StringBuilder()
+                .append("time " + LocalDateTime.ofInstant(Instant.ofEpochMilli(time), ZoneId.systemDefault()))
+                .append("type " + type())
+                .append("subject " + subject())
+                .toString();
+    }
+}