Break the circular dependency by seperating networkpolicymonitor from bandwidthEvaluator
Issue-ID: DCAEGEN2-3270
Signed-off-by: decheng zhang <decheng.zhang@huawei.com>
Change-Id: I2632e5ea500121b8c223bdbcda16bdc24dc36895
Signed-off-by: decheng zhang <decheng.zhang@huawei.com>
- [DCAEGEN2-3273](https://jira.onap.org/browse/DCAEGEN2-3273) - Fix bug that config thread hang up when cbs policy is empty
- - [DCAEGEN2-3270](https://jira.onap.org/browse/DCAEGEN2-3270) - Dcae slicems startup error
+ - [DCAEGEN2-3270](https://jira.onap.org/browse/DCAEGEN2-3270) - Dcae slicems startup error, breaking the circular dependency.
## [1.1.4] - 2022/07/28
- [DCAEGEN2-3120](https://jira.onap.org/browse/DCAEGEN2-3120) - Enhance sliceanalysis MS to use DCAE SDK dmaap-client lib
private static Logger log = LoggerFactory.getLogger(BandwidthEvaluator.class);
private Configuration configuration;
- @Autowired
- AaiService aaiService;
-
- @Autowired
- CCVPNPmDatastore ccvpnPmDatastore;
-
@Autowired
StrategyFactory strategyFactory;
private Loop evaluationEventLoop;
- private Loop aaiEventLoop;
private static final Event KILL_PILL = new SimpleEvent(null, 0);
private static final int DEFAULT_EVAL_INTERVAL = 5;
}
};
- /**
- * AAI data consumer loop
- */
- aaiEventLoop = new Loop("AAIEventLoop"){
- @Override
- public void process(Event event) {
- if (event.type() == SimpleEvent.Type.AAI_BW_REQ){
- log.debug("=== Processing new AAI network policy query at: {} ===", event.time());
- String serviceId = (String) event.subject();
- Map<String, Integer> maxBandwidthData = aaiService.fetchMaxBandwidthOfService(serviceId);
- if (maxBandwidthData.get("maxBandwidth") != null){
- log.debug("Successfully retrieved bandwidth info from AAI; service: {}, bandwidth: {}",
- serviceId, maxBandwidthData.get("maxBandwidth"));
- int bwValue = maxBandwidthData.get("maxBandwidth").intValue();
- if (ccvpnPmDatastore.getProvBwOfSvc(serviceId) == 0){
- ccvpnPmDatastore.updateProvBw(serviceId, bwValue, true);
- log.debug("Provision bw of cll {} updated from 0 to {}, max bw is {}", serviceId, ccvpnPmDatastore.getProvBwOfSvc(serviceId), bwValue);
- } else if (ccvpnPmDatastore.getProvBwOfSvc(serviceId) != bwValue) {
- log.debug("Service modification complete; serviceId: {} update prov bw from {} to {}", serviceId, ccvpnPmDatastore.getProvBwOfSvc(serviceId), bwValue);
- ccvpnPmDatastore.updateProvBw(serviceId, bwValue, true);
- ccvpnPmDatastore.updateSvcState(serviceId, ServiceState.RUNNING);
- log.debug("Service state of {} is changed to running, {}", serviceId, ccvpnPmDatastore.getStatusOfSvc(serviceId));
- }
- }
- log.debug("=== Processing AAI network policy query complete ===");
- }
- }
- };
scheduleEvaluation();
}
@PreDestroy
public void stop(){
stopScheduleEvaluation();
- aaiEventLoop.stop();
evaluationEventLoop.stop();
}
public void post(@NonNull Event event){
log.info("A new event triggered, type: {}, subject: {}, at time: {}",
event.type(), event.subject(), event.time());
- if (event.type() == SimpleEvent.Type.AAI_BW_REQ) {
- aaiEventLoop.add(event);
- } else if (event.type() == SimpleEvent.Type.PERIODIC_CHECK) {
+ if (event.type() == SimpleEvent.Type.PERIODIC_CHECK) {
evaluationEventLoop.add(event);
} else if (event.type() == SimpleEvent.Type.ONDEMAND_CHECK) {
evaluationEventLoop.add(event);
private static double precision; // in Mbps;
@Autowired
- BandwidthEvaluator bandwidthEvaluator;
+ NetworkPolicyMonitor networkPolicyMonitor;
@Autowired
CCVPNPmDatastore ccvpnPmDatastore;
}
/**
- * Post/broadcast event to the BandwidthEvaluator
+ * Post/broadcast event to the networkPolicyMonitor
* @param event event object
*/
private void post(Event event){
- bandwidthEvaluator.post(event);
+ networkPolicyMonitor.post(event);
}
private void loadConfig() {
*/
@Component
public class FlexibleThresholdStrategy implements EvaluationStrategy{
- private static Logger log = LoggerFactory.getLogger(FixedUpperBoundStrategy.class);
+ private static Logger log = LoggerFactory.getLogger(FlexibleThresholdStrategy.class);
private Configuration configuration;
private static final String TYPE_NAME = "FlexibleThresholdStrategy";
private static final String SERVICE_INSTANCE_LOCATION_ID = "service-instance-location-id";
private static double precision; // in Mbps;
@Autowired
- BandwidthEvaluator bandwidthEvaluator;
+ NetworkPolicyMonitor networkPolicyMonitor;
@Autowired
CCVPNPmDatastore ccvpnPmDatastore;
}
/**
- * Post/broadcast event to the BandwidthEvaluator
+ * Post/broadcast event to the networkPolicyMonitor
* @param event event object
*/
private void post(Event event){
- bandwidthEvaluator.post(event);
+ networkPolicyMonitor.post(event);
}
private void loadConfig() {
--- /dev/null
+/*******************************************************************************
+ * ============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 lombok.NonNull;
+import org.onap.slice.analysis.ms.aai.AaiService;
+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.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Actor that processes aai network-policy query request
+ */
+@Component
+public class NetworkPolicyMonitor {
+ private static Logger log = LoggerFactory.getLogger(NetworkPolicyMonitor.class);
+ private Loop aaiEventLoop;
+ private static final Event KILL_PILL = new SimpleEvent(null, 0);
+
+ @Autowired
+ AaiService aaiService;
+
+ @Autowired
+ CCVPNPmDatastore ccvpnPmDatastore;
+
+ /**
+ * Initialize and start the NetworkPolicyMonitor.
+ */
+ @PostConstruct
+ public void init() {
+ /**
+ * AAI data consumer loop
+ */
+ aaiEventLoop = new Loop("AAIEventLoop"){
+ @Override
+ public void process(Event event) {
+ if (event.type() == SimpleEvent.Type.AAI_BW_REQ){
+ log.debug("=== Processing new AAI network policy query at: {} ===", event.time());
+ String serviceId = (String) event.subject();
+ Map<String, Integer> maxBandwidthData = aaiService.fetchMaxBandwidthOfService(serviceId);
+ if (maxBandwidthData.get("maxBandwidth") != null){
+ log.debug("Successfully retrieved bandwidth info from AAI; service: {}, bandwidth: {}",
+ serviceId, maxBandwidthData.get("maxBandwidth"));
+ int bwValue = maxBandwidthData.get("maxBandwidth").intValue();
+ if (ccvpnPmDatastore.getProvBwOfSvc(serviceId) == 0){
+ ccvpnPmDatastore.updateProvBw(serviceId, bwValue, true);
+ log.debug("Provision bw of cll {} updated from 0 to {}, max bw is {}", serviceId, ccvpnPmDatastore.getProvBwOfSvc(serviceId), bwValue);
+ } else if (ccvpnPmDatastore.getProvBwOfSvc(serviceId) != bwValue) {
+ log.debug("Service modification complete; serviceId: {} update prov bw from {} to {}", serviceId, ccvpnPmDatastore.getProvBwOfSvc(serviceId), bwValue);
+ ccvpnPmDatastore.updateProvBw(serviceId, bwValue, true);
+ ccvpnPmDatastore.updateSvcState(serviceId, ServiceState.RUNNING);
+ log.debug("Service state of {} is changed to running, {}", serviceId, ccvpnPmDatastore.getStatusOfSvc(serviceId));
+ }
+ }
+ log.debug("=== Processing AAI network policy query complete ===");
+ }
+ }
+ };
+ }
+
+ /**
+ * 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);
+ }
+ }
+ /**
+ * 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("NetworkPolicyMonitor -- {} 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);
+ }
+ }
+}
private static Logger log = LoggerFactory.getLogger(StrategyFactory.class);
@Autowired
- List<EvaluationStrategy> strategies;
+ public List<EvaluationStrategy> strategies;
private StrategyFactory() {}
--- /dev/null
+/*******************************************************************************
+ * ============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.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mockito;
+import org.mockito.Spy;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = BandwidthEvaluatorTest.class)
+public class NetworkPolicyMonitorTest {
+
+ @Spy
+ @InjectMocks
+ NetworkPolicyMonitor NetworkPolicyMonitor;
+ @Test
+ public void initTest() {
+ NetworkPolicyMonitor.init();
+ Mockito.verify(NetworkPolicyMonitor, Mockito.atLeastOnce()).init();
+ }
+
+ @Test
+ public void postTest() {
+ Event evt = new SimpleEvent(null, "{}");
+ NetworkPolicyMonitor.post(evt);
+ Mockito.verify(NetworkPolicyMonitor, Mockito.atLeastOnce()).post(Mockito.any(Event.class));
+ }
+}
Code Review / dcaegen2 / services.git / commitdiff