diff options
Diffstat (limited to 'components/slice-analysis-ms/src/main/java/org')
9 files changed, 131 insertions, 32 deletions
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 289fedfd..ab67c031 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 @@ -367,21 +367,24 @@ public class AaiService implements AaiInterface { try { ResponseEntity<String> resp = restclient.sendGetRequest(networkPolicyUrl, new ParameterizedTypeReference<String>() { }); + log.debug("AaiService received {} : {}", resp.getStatusCodeValue(), resp.getBody()); if (resp.getStatusCodeValue() == 200){ String networkPolicy = resp.getBody(); JSONObject networkPolicyJson = new JSONObject(networkPolicy); JSONArray networkPolicyList = networkPolicyJson.optJSONArray("network-policy"); - if (networkPolicyList != null){ + if (networkPolicyList != null && networkPolicyList.length() > 0){ JSONObject networkPolicyOjb = networkPolicyList.getJSONObject(0); result.put("maxBandwidth", networkPolicyOjb.getInt("max-bandwidth")); + log.info("Successfully retrieved max bandwidth for service {}: {}", + serviceId, result.get("maxBandwidth")); return result; } - log.info("Successfully fetched max bandwidth {}: {}", serviceId, result); } + log.warn("Failed to retrieve max bandwidth for service {}, no such network-policy or no valid max-bandwidth " + + "associated", serviceId); } catch (Exception e){ log.warn("Error encountered when fetching maxbandwidth: " + e); - } - return null; + return new HashMap<String, Integer>(); } } 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 index dc2cd775..0259f130 100644 --- 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 @@ -29,6 +29,8 @@ 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.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @@ -40,6 +42,7 @@ import javax.annotation.PostConstruct; @Component public class AaiEventNotificationCallback implements NotificationCallback { + private static Logger logger = LoggerFactory.getLogger(AaiEventNotificationCallback.class); private static final String EVENT_HEADER = "event-header"; private static final String ACTION = "action"; private static final String ENTITY_TYPE = "entity-type"; @@ -71,26 +74,60 @@ public class AaiEventNotificationCallback implements NotificationCallback { JsonElement jsonElement = parser.parse(msg); if (jsonElement.isJsonObject()){ //handle a single AAI_EVENT + logger.debug("Handle a single aai-event"); handleMsgJsonObject(jsonElement.getAsJsonObject()); } else if (jsonElement.isJsonArray()){ //handle a series of AAI_EVENT + logger.debug("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()); + if (jsonArray.get(i).isJsonPrimitive()){ + // Deal with a batch of event message + handleNotification(jsonArray.get(i).getAsString()); + } else { + 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); + if (!header.has(ACTION) || !header.get(ACTION).getAsString().equals(aaiNotifTargetAction)){ + return; + } + if (!header.has(ENTITY_TYPE) || !header.get(ENTITY_TYPE).getAsString().equals(aaiNotifTargetEntity)){ + return; + } + if (!header.has(SOURCE_NAME) || !header.get(SOURCE_NAME).getAsString().equals(aaiNotifTargetSource)){ + return; + } + JsonObject entity = jsonObject.get(ENTITY).getAsJsonObject(); + JsonObject body = getNestedJsonObject(entity, aaiNotifTargetEntity); + logger.debug("AAI-EVENT entity object {}", body); + if (body == null){ + return; + } + Event event = new SimpleEvent<>(SimpleEvent.Type.ONDEMAND_CHECK, body); + bandwidthEvaluator.post(event); + } + + private JsonObject getNestedJsonObject(JsonObject obj, String target){ + for (String k: obj.keySet()){ + if (k.equals(target)){ + //Found it; + return obj.getAsJsonArray(k).get(0).getAsJsonObject(); + } + if (obj.get(k).isJsonObject()) { + return getNestedJsonObject(obj.getAsJsonObject(k), target); + } else if (obj.get(k).isJsonArray()){ + JsonElement tmp = obj.getAsJsonArray(k).get(0); + if (tmp.isJsonObject()){ + return getNestedJsonObject(tmp.getAsJsonObject(), target); } } } + return null; } } 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 index 0c1ac604..aa1bc964 100644 --- 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 @@ -91,8 +91,11 @@ public class MRTopicMonitor implements Runnable { public void run(){ while (running){ try { + logger.debug("Topic: {} getting new msg...", name); Iterable<String> dmaapMsgs = consumerWrapper.fetch(); for (String msg : dmaapMsgs){ + logger.debug("Received message: {}" + + "\r\n and processing start", msg); process(msg); } } catch (IOException | RuntimeException e){ 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 index 66c27413..e5aaa1e3 100644 --- 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 @@ -321,12 +321,7 @@ public class MRTopicParams { 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; 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 index 83bfcdbe..584da7b7 100644 --- 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 @@ -104,6 +104,8 @@ public class VesNotificationCallback implements NotificationCallback { log.error("Error converting VES msg to object, {}", e.getMessage()); } if (cllId != null && uniId != null && bw != null){ + log.info("Saving new CCVPN service usage data into ccvpnPmDatastore"); + log.debug("new bandwidth data -- serviceId: {}, uniId: {}, bw: {}", cllId, uniId, bw); 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 d5aeb7b4..2a509aa1 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 @@ -162,7 +162,7 @@ public class Configuration { aaiNotifTargetAction = jsonObject.get("sliceanalysisms.aaiNotif.targetAction").getAsString(); aaiNotifTargetSource = jsonObject.get("sliceanalysisms.aaiNotif.targetSource").getAsString(); - aaiNotifTargetSource = jsonObject.get("sliceanalysisms.aaiNotif.targetEntity").getAsString(); + aaiNotifTargetEntity = jsonObject.get("sliceanalysisms.aaiNotif.targetEntity").getAsString(); ccvpnEvalInterval = jsonObject.get("sliceanalysisms.ccvpnEvalInterval").getAsInt(); ccvpnEvalThreshold = jsonObject.get("sliceanalysisms.ccvpnEvalThreshold").getAsDouble(); ccvpnEvalPrecision = jsonObject.get("sliceanalysisms.ccvpnEvalPrecision").getAsDouble(); 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 ceaeb730..e0460407 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 @@ -22,13 +22,11 @@ package org.onap.slice.analysis.ms.models.policy; import com.fasterxml.jackson.annotation.JsonInclude; -import com.fasterxml.jackson.annotation.JsonInclude.Include; -import com.fasterxml.jackson.annotation.JsonProperty; /** * Model class for the Paylaod Object */ -@JsonInclude(Include.NON_NULL) +@JsonInclude(JsonInclude.Include.NON_NULL) public class Payload { private String name; @@ -38,6 +36,8 @@ public class Payload { private String networkType; private AdditionalProperties<?> additionalProperties; private String serviceType; + private String modelInvariantUuid; + private String modelUuid; public String getName() { return name; @@ -95,4 +95,19 @@ public class Payload { this.serviceType = serviceType; } + public String getModelInvariantUuid() { + return modelInvariantUuid; + } + + public void setModelInvariantUuid(String modelInvariantUuid) { + this.modelInvariantUuid = modelInvariantUuid; + } + + public String getModelUuid() { + return modelUuid; + } + + public void setModelUuid(String modelUuid) { + this.modelUuid = modelUuid; + } } 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 297683b4..01e2886a 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 @@ -153,6 +153,8 @@ public class PolicyService { payload.setName("cloud-leased-line-101"); payload.setServiceInstanceID(cllId); payload.setAdditionalProperties(additionalProperties); + payload.setModelInvariantUuid("6790ab0e-034f-11eb-adc1-0242ac120002"); + payload.setModelUuid("6790ab0e-034f-11eb-adc1-0242ac120002"); OnsetMessage onsetmsg = new OnsetMessage(); try { @@ -184,7 +186,7 @@ public class PolicyService { String msg = ""; try { msg = objectMapper.writeValueAsString(onsetMessage); - log.info("Policy onset message for ControlLoop-CCVPN-CLL is {}", msg); + log.info("Sending onset message to Onap/Policy for ControlLoop-CCVPN-CLL, the msg: {}", msg); policyDmaapClient.sendNotificationToPolicy(msg); } catch (Exception e) { 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 index a7847b0e..c66122c1 100644 --- 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 @@ -46,6 +46,7 @@ import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import static java.util.concurrent.Executors.newSingleThreadExecutor; @@ -105,7 +106,7 @@ public class BandwidthEvaluator { @Override public void process(Event event) { if (event.type() == SimpleEvent.Type.PERIODIC_CHECK && isPeriodicCheckOn()){ - log.info("Received new periodic check request: {}", event.time()); + log.info("=== Processing 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()) { @@ -114,10 +115,12 @@ public class BandwidthEvaluator { if (usedBws == null) { // No enough data for evaluating + log.debug("CCVPN Evaluator Output: service {}, not enough data to evaluate", serviceId); continue; } if (ccvpnPmDatastore.getMaxBwOfSvc(serviceId) == 0) { // Max bandwidth not cached yet + log.debug("CCVPN Evaluator Output: service {}, max bandwidth not cached, wait for next round", serviceId); post(new SimpleEvent(SimpleEvent.Type.AAI_BW_REQ, serviceId)); continue; } @@ -126,48 +129,81 @@ public class BandwidthEvaluator { .summaryStatistics() .getAverage(); if (needAdjust(serviceId, avg, ccvpnPmDatastore.getMaxBwOfSvc(serviceId))) { + log.debug("CCVPN Evaluator Output: service {}, need adjustment, putting into candidate list", serviceId); int newBw = (int) (Math.ceil((avg / threshold) * 1.2 / precision) * precision); candidate.put(serviceId, Math.max(candidate.getOrDefault(serviceId, 0), newBw)); } } + // check svc under maintenance + Map<String , ServiceState> svcUnderMaintenance = getServicesUnderMaintenance(); + for (Map.Entry<String, ServiceState> entry: svcUnderMaintenance.entrySet()){ + candidate.putIfAbsent(entry.getKey(), 0); + } + // fetch the maxbandwidth info if underMaintenance; otherwise send modification request for(Map.Entry<String, Integer> entry: candidate.entrySet()) { if (isServiceUnderMaintenance(entry.getKey())) { + if (entry.getValue() == 0){ + log.debug("CCVPN Evaluator Output: service {}," + + " are in maintenance state, fetching bandwidth info from AAI", entry.getKey()); + } else { + log.debug("CCVPN Evaluator Output: candidate {}," + + " need adjustment, but skipped due to maintenance state", entry.getKey()); + } post(new SimpleEvent(SimpleEvent.Type.AAI_BW_REQ, entry.getKey())); continue; } + log.debug("CCVPN Evaluator Output: candidate {}," + + " need adjustment, sending request to policy", entry.getKey()); ccvpnPmDatastore.updateSvcState(entry.getKey(), ServiceState.UNDER_MAINTENANCE); sendModifyRequest(entry.getKey(), entry.getValue(), RequestOwner.DCAE); } + log.info("=== Processing periodic check complete ==="); } else if (event.type() == SimpleEvent.Type.ONDEMAND_CHECK && isOnDemandCheckOn()) { - log.info("Received new on-demand check request: {}", event.time()); + log.info("=== Processing 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) { + if (maxBandwidthData.get("maxBandwidth") != null + && maxBandwidthData.get("maxBandwidth") != newBandwidth){ + log.debug("CCVPN Evaluator Output: on-demand adjustment request for service: {} processed," + + " sending request to policy", serviceId); ccvpnPmDatastore.updateSvcState(serviceId, ServiceState.UNDER_MAINTENANCE); sendModifyRequest(serviceId, newBandwidth, RequestOwner.UUI); } + } else { + log.debug("CCVPN Evaluator Output: service {}," + + " received on-demand request, but skipped due to maintenance state", serviceId); } + log.info("=== Processing on-demand check complete ==="); } } private void sendModifyRequest(String cllId, Integer newBandwidth, RequestOwner owner) { + log.info("Sending modification request to policy. RequestOwner: {} - Service: {} change to bw: {}", + owner, cllId, newBandwidth); policyService.sendOnsetMessageToPolicy( policyService.formPolicyOnsetMessageForCCVPN(cllId, newBandwidth, owner) ); } private boolean needAdjust(String serivceId, double currentAverageUsage, int maxBandwidth){ + log.debug("CCVPN Service Usage Analysis: usage: {}, threshold: {}, maxbw {}", currentAverageUsage, threshold, maxBandwidth); return currentAverageUsage > threshold * maxBandwidth; } private boolean isServiceUnderMaintenance(String serivceId) { return ccvpnPmDatastore.getStatusOfSvc(serivceId) == ServiceState.UNDER_MAINTENANCE; } + + private Map<String, ServiceState> getServicesUnderMaintenance(){ + return ccvpnPmDatastore.getSvcStatusMap().entrySet() + .stream() + .filter(e -> e.getValue() == ServiceState.UNDER_MAINTENANCE) + .collect(Collectors.toMap(p -> p.getKey(), p -> p.getValue())); + } }; /** @@ -177,18 +213,22 @@ public class BandwidthEvaluator { @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()); + log.info("=== Processing 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 (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.getMaxBwOfSvc(serviceId) == 0){ - ccvpnPmDatastore.updateMaxBw(serviceId, bwVal, true); - } else if (ccvpnPmDatastore.getMaxBwOfSvc(serviceId) != bwVal) { - ccvpnPmDatastore.updateMaxBw(serviceId, bwVal, true); + ccvpnPmDatastore.updateMaxBw(serviceId, bwValue, true); + } else if (ccvpnPmDatastore.getMaxBwOfSvc(serviceId) != bwValue) { + log.debug("Service modification complete; serviceId: {} with new bandwidth: {}", serviceId, bwValue); + ccvpnPmDatastore.updateMaxBw(serviceId, bwValue, true); ccvpnPmDatastore.updateSvcState(serviceId, ServiceState.RUNNING); } } + log.info("=== Processing AAI network policy query complete ==="); } } }; @@ -229,6 +269,8 @@ public class BandwidthEvaluator { * @param event event object */ public void post(@NonNull Event event){ + log.debug("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) { |