diff options
Diffstat (limited to 'components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service')
7 files changed, 117 insertions, 71 deletions
diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/AverageCalculator.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/AverageCalculator.java index a003e9c0..52670d53 100644 --- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/AverageCalculator.java +++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/AverageCalculator.java @@ -50,19 +50,22 @@ public class AverageCalculator { * Find average of samples */ public List<MeasurementObject> findAverageOfSamples(List<List<MeasurementObject>> samples) { + log.debug("find average for samples {}", samples); int numOfSamples = samples.size(); List<MeasurementObject> result = new ArrayList<>(); - samples.forEach(sample -> - sample.forEach(cellMeasObj -> { - int index = result.indexOf(cellMeasObj); - if(index != -1) { - result.set(index, findSum(result.get(index), cellMeasObj)); + if(!samples.isEmpty()) { + for(List<MeasurementObject> sample : samples) { + for(MeasurementObject cellMeasObj : sample) { + int index = MeasurementObject.findIndex(cellMeasObj.getMeasurementObjectId(), result); + if(index != -1) { + result.set(index, findSum(result.get(index), cellMeasObj)); + } + else { + result.add(cellMeasObj); + } + } } - else { - result.add(cellMeasObj); - } - }) - ); + } return findAvg(result, numOfSamples); } @@ -81,13 +84,15 @@ public class AverageCalculator { * Calculate the average */ public List<MeasurementObject> findAvg(List<MeasurementObject> result, int numOfSamples) { - result.forEach(cellMeasObj -> - pmNames.forEach(pmName -> { - int value = (cellMeasObj.getPmData().get(pmName))/numOfSamples; - cellMeasObj.getPmData().put(pmName, value); - }) - ); - log.debug("Average of measurement data samples {}",result); + if(!result.isEmpty()) { + result.forEach(cellMeasObj -> + pmNames.forEach(pmName -> { + int value = (cellMeasObj.getPmData().get(pmName))/numOfSamples; + cellMeasObj.getPmData().put(pmName, value); + }) + ); + log.debug("Average of measurement data samples {}",result); + } return result; } } diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ConsumerThread.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ConsumerThread.java index 39235cd5..6868bc46 100644 --- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ConsumerThread.java +++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/ConsumerThread.java @@ -21,6 +21,8 @@ package org.onap.slice.analysis.ms.service; +import java.util.List; + import org.onap.slice.analysis.ms.configdb.IConfigDbService; import org.onap.slice.analysis.ms.models.Configuration; import org.onap.slice.analysis.ms.utils.BeanUtil; @@ -31,7 +33,7 @@ import org.slf4j.LoggerFactory; * This Thread class consumes message from pm data queue and sends onset message to policy */ public class ConsumerThread extends Thread { - private static Logger log = LoggerFactory.getLogger(PmThread.class); + private static Logger log = LoggerFactory.getLogger(ConsumerThread.class); private PmDataQueue pmDataQueue; private IConfigDbService configDbService; private SnssaiSamplesProcessor snssaiSamplesProcessor; @@ -55,16 +57,31 @@ public class ConsumerThread extends Thread { public void run() { boolean done = false; String snssai = ""; + boolean result = false; + List<String> nfs = null; while (!done) { try { Thread.sleep(initialDelaySec); snssai = pmDataQueue.getSnnsaiFromQueue(); if (!snssai.equals("")) { - log.info("Consumer thread started for s-nssai {}",snssai); - snssaiSamplesProcessor.processSamplesOfSnnsai(snssai, configDbService.fetchNetworkFunctionsOfSnssai(snssai)); + log.info("Consumer thread processing data for s-nssai {}",snssai); + try { + nfs = configDbService.fetchNetworkFunctionsOfSnssai(snssai); + } + catch(Exception e) { + pmDataQueue.putSnssaiToQueue(snssai); + log.error("Exception caught while fetching nfs of snssai {}, {}", snssai, e.getMessage()); + } + if(nfs != null) { + result = snssaiSamplesProcessor.processSamplesOfSnnsai(snssai, nfs); + if(!result) { + log.info("Not enough samples to process for {}",snssai); + pmDataQueue.putSnssaiToQueue(snssai); + } + } } } catch (Exception e) { - log.error("Exception in Consumer Thread ", e); + log.error("Exception in Consumer Thread, {}", e.getMessage()); done = true; } } diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmDataQueue.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmDataQueue.java index d907bfed..4f196979 100644 --- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmDataQueue.java +++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmDataQueue.java @@ -31,6 +31,8 @@ import java.util.concurrent.LinkedBlockingQueue; import org.onap.slice.analysis.ms.models.MeasurementObject; import org.onap.slice.analysis.ms.models.SubCounter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; /** @@ -38,6 +40,8 @@ import org.springframework.stereotype.Component; */ @Component public class PmDataQueue { + private static Logger log = LoggerFactory.getLogger(PmDataQueue.class); + private Map<SubCounter, Queue<List<MeasurementObject>>> subCounterMap = Collections.synchronizedMap(new LinkedHashMap<SubCounter, Queue<List<MeasurementObject>>>()); private Queue<String> snssaiList = new LinkedBlockingQueue<>(); @@ -54,6 +58,7 @@ public class PmDataQueue { measQueue.add(measurementObjectData); subCounterMap.put(subCounter, measQueue); } + log.debug("Queue: {}", subCounterMap); } /** @@ -61,12 +66,15 @@ public class PmDataQueue { * returns the specified number of samples */ public List<List<MeasurementObject>> getSamplesFromQueue(SubCounter subCounter, int samples) { - List<List<MeasurementObject>> sampleList = new LinkedList<>(); + List<List<MeasurementObject>> sampleList = null; if (subCounterMap.containsKey(subCounter)){ Queue<List<MeasurementObject>> measQueue = subCounterMap.get(subCounter); - while(samples > 0) { - sampleList.add(measQueue.remove()); - samples --; + if(measQueue.size() >= samples) { + sampleList = new LinkedList<>(); + while(samples > 0) { + sampleList.add(measQueue.remove()); + samples --; + } } } return sampleList; @@ -86,10 +94,12 @@ public class PmDataQueue { public String getSnnsaiFromQueue() { String snssai = ""; try { - snssai = snssaiList.remove(); + if(!snssaiList.isEmpty()){ + snssai = snssaiList.remove(); + } } catch(Exception e) { - + log.error("Problem fetching from the Queue, {}", e.getMessage()); } return snssai; } diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmEventProcessor.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmEventProcessor.java index 99c24c8a..51a04660 100644 --- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmEventProcessor.java +++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmEventProcessor.java @@ -31,6 +31,8 @@ import org.onap.slice.analysis.ms.models.pmnotification.Event; import org.onap.slice.analysis.ms.models.pmnotification.MeasInfoList; import org.onap.slice.analysis.ms.models.pmnotification.MeasResult; import org.onap.slice.analysis.ms.models.pmnotification.MeasValuesList; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Scope; import org.springframework.stereotype.Component; @@ -41,7 +43,7 @@ import org.springframework.stereotype.Component; @Scope("prototype") public class PmEventProcessor implements IPmEventProcessor{ protected Map<String, List<MeasurementObject>> instanceMap = new HashMap<>(); - + private static Logger log = LoggerFactory.getLogger(PmEventProcessor.class); /** * Process the PM event @@ -53,6 +55,7 @@ public class PmEventProcessor implements IPmEventProcessor{ List<MeasValuesList> subCounterMeasurements = measurement.getMeasValuesList(); subCounterMeasurements.forEach(subCounterMeasurement -> processMeasurementObjectData(collectedSubCounters, subCounterMeasurement)); }); + log.info("Processed Event: {}", instanceMap); return instanceMap; } @@ -65,26 +68,28 @@ public class PmEventProcessor implements IPmEventProcessor{ measResultList.forEach(measResult -> { String pmName = collectedSubCounters.get(measResult.getP()-1); Integer pmValue = Integer.valueOf(measResult.getsValue()); - Map<String,String> pmMapping = getMapKey(pmName); - String snssai = pmMapping.get("snssai"); - String pm = pmMapping.get("pm"); - Map<String, Integer> pmData = new HashMap<>(); - pmData.put(pm, pmValue); - if (instanceMap.containsKey(snssai)) { - int index = instanceMap.get(snssai).indexOf(new MeasurementObject(measObjId)); - if (index == -1) { - instanceMap.get(snssai).add(new MeasurementObject(measObjId,pmData)); + if(pmName.contains("PrbUsedUl") || pmName.contains("PrbUsedDl")) { + Map<String,String> pmMapping = getMapKey(pmName); + String snssai = pmMapping.get("snssai"); + String pm = pmMapping.get("pm"); + Map<String, Integer> pmData = new HashMap<>(); + pmData.put(pm, pmValue); + if (instanceMap.containsKey(snssai)) { + int index = MeasurementObject.findIndex(measObjId, instanceMap.get(snssai)); + if (index == -1) { + instanceMap.get(snssai).add(new MeasurementObject(measObjId,pmData)); + } + else { + instanceMap.get(snssai).get(index).getPmData().put(pm, pmValue); + } } else { - instanceMap.get(snssai).get(index).getPmData().put(pmName, pmValue); - } - } - else { - List<MeasurementObject> l = new LinkedList<>(); - l.add(new MeasurementObject(measObjId,pmData)); - instanceMap.put(snssai, l); + List<MeasurementObject> l = new LinkedList<>(); + l.add(new MeasurementObject(measObjId,pmData)); + instanceMap.put(snssai, l); + } } - }); + }); } /** @@ -92,14 +97,10 @@ public class PmEventProcessor implements IPmEventProcessor{ */ public Map<String, String> getMapKey(String pmName) { String [] pmNameArr = pmName.split("\\."); - String snssai = ""; String pm = pmNameArr[1]; Map<String, String> result = new HashMap<>(); result.put("pm", pm); - if ((pm.equalsIgnoreCase("PrbUsedDl")) || (pm.equalsIgnoreCase("PrbUsedUl"))){ - snssai = pmNameArr[2]; - } - result.put("snssai", snssai); + result.put("snssai", pmNameArr[2]); return result; } }
\ No newline at end of file diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmThread.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmThread.java index d8bcbf64..e2903ae1 100644 --- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmThread.java +++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/PmThread.java @@ -52,7 +52,6 @@ public class PmThread extends Thread { super(); this.newPmNotification = BeanUtil.getBean(NewPmNotification.class); this.performanceNotificationsRepository = BeanUtil.getBean(PerformanceNotificationsRepository.class); - this.pmEventProcessor = BeanUtil.getBean(IPmEventProcessor.class); this.pmDataQueue = BeanUtil.getBean(PmDataQueue.class); } @@ -68,12 +67,13 @@ public class PmThread extends Thread { while (!done) { try { Thread.sleep(1000); - if (newPmNotification.getNewNotif()) { - log.info("New PM notification from Dmaap"); + if (newPmNotification.getNewNotif()) { String pmNotificationString = performanceNotificationsRepository.getPerformanceNotificationFromQueue(); if(pmNotificationString != null) { + log.info("New PM notification"); ObjectMapper mapper = new ObjectMapper(); pmNotification = mapper.readValue(pmNotificationString, PmNotification.class); + this.pmEventProcessor = BeanUtil.getBean(IPmEventProcessor.class); processedData = pmEventProcessor.processEvent(pmNotification.getEvent()); String networkFunction = pmNotification.getEvent().getPerf3gppFields().getMeasDataCollection().getMeasuredEntityDn(); processedData.forEach((key,value) -> { 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 80063398..165bef90 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 @@ -99,8 +99,8 @@ public class PolicyService { ObjectMapper obj = new ObjectMapper(); String msg = ""; try { - log.debug("Policy onset message for S-NSSAI: {} is {}", snssai, msg); msg = obj.writeValueAsString(onsetMessage); + log.info("Policy onset message for S-NSSAI: {} is {}", snssai, msg); policyDmaapClient.sendNotificationToPolicy(msg); } catch (Exception e) { diff --git a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/SnssaiSamplesProcessor.java b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/SnssaiSamplesProcessor.java index 16fb41ad..e8d5de3e 100644 --- a/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/SnssaiSamplesProcessor.java +++ b/components/slice-analysis-ms/src/main/java/org/onap/slice/analysis/ms/service/SnssaiSamplesProcessor.java @@ -63,7 +63,7 @@ public class SnssaiSamplesProcessor { private Map<String, List<String>> ricToCellMapping = new HashMap<>(); private Map<String, Map<String, Integer>> ricToPrbsMapping = new HashMap<>(); private Map<String, Map<String, Integer>> ricToThroughputMapping = new HashMap<>(); - private int samples; + private int noOfSamples; private List<String> pmsToCompute; private Map<String, String> prbThroughputMapping = new HashMap<>(); private int minPercentageChange; @@ -71,7 +71,7 @@ public class SnssaiSamplesProcessor { @PostConstruct public void init() { Configuration configuration = Configuration.getInstance(); - samples = configuration.getSamples(); + noOfSamples = configuration.getSamples(); pmsToCompute = new ArrayList<>(); pmsToCompute.add("PrbUsedDl"); pmsToCompute.add("PrbUsedUl"); @@ -84,22 +84,34 @@ public class SnssaiSamplesProcessor { /** * process the measurement data of an S-NSSAI */ - public void processSamplesOfSnnsai(String snssai, List<String> networkFunctions) { - networkFunctions.forEach(nf -> { + public boolean processSamplesOfSnnsai(String snssai, List<String> networkFunctions) { + List<MeasurementObject> sample = null; + List<List<MeasurementObject>> samples = null; + log.info("Network Functions {} of snssai {}", networkFunctions, snssai); + for(String nf : networkFunctions) { log.debug("Average of samples for {}:", snssai); - addToMeasurementList(averageCalculator.findAverageOfSamples(pmDataQueue.getSamplesFromQueue(new SubCounter(nf, snssai), samples))); - }); + samples = pmDataQueue.getSamplesFromQueue(new SubCounter(nf, snssai), noOfSamples); + if(samples != null) { + sample = averageCalculator.findAverageOfSamples(samples); + addToMeasurementList(sample); + } + else { + log.info("Not enough samples present for nf {}", nf); + return false; + } + } + log.info("snssai measurement list {}", snssaiMeasurementList); ricToCellMapping = configDbService.fetchRICsOfSnssai(snssai); - log.debug("RIC to Cell Mapping for {} S-NSSAI: {}", snssai, ricToCellMapping); - Map<String, Map<String, Integer>> ricConfiguration = configDbService.fetchCurrentConfigurationOfRIC(snssai); + log.info("RIC to Cell Mapping for {} S-NSSAI: {}", snssai, ricToCellMapping); + Map<String, Map<String, Object>> ricConfiguration = configDbService.fetchCurrentConfigurationOfRIC(snssai); Map<String, Integer> sliceConfiguration = configDbService.fetchCurrentConfigurationOfSlice(snssai); - log.debug("RIC Configuration: {}", ricConfiguration); - log.debug("Slice Configuration: {}", sliceConfiguration); + log.info("RIC Configuration {} and Slice Configuration {}", ricConfiguration, sliceConfiguration); pmsToCompute.forEach(pm -> { + log.debug("processing for pm {}", pm); sumOfPrbsAcrossCells(pm); int sum = computeSum(pm); computeThroughput(sliceConfiguration, sum, pm); - calculatePercentageChange(ricConfiguration, pm); + calculatePercentageChange(ricConfiguration, prbThroughputMapping.get(pm)); }); updateConfiguration(); if(ricToThroughputMapping.size() > 0) { @@ -107,7 +119,7 @@ public class SnssaiSamplesProcessor { addProps.setResourceConfig(ricToThroughputMapping); policyService.sendOnsetMessageToPolicy(snssai, addProps, configDbService.fetchServiceDetails(snssai)); } - + return true; } /** @@ -132,17 +144,18 @@ public class SnssaiSamplesProcessor { * Calculate the change in the configuration value and keep the configuration only if it is greater than a * specific limit */ - protected void calculatePercentageChange(Map<String, Map<String, Integer>> ricConfiguration, String pm) { + protected void calculatePercentageChange(Map<String, Map<String, Object>> ricConfiguration, String pm) { Iterator<Map.Entry<String, Map<String,Integer>>> it = ricToThroughputMapping.entrySet().iterator(); Map.Entry<String, Map<String,Integer>> entry = null; float existing = 0; float change = 0; while(it.hasNext()) { entry = it.next(); - existing = ricConfiguration.get(entry.getKey()).get(pm); + existing = (float)((int)ricConfiguration.get(entry.getKey()).get(pm)); change = ((Math.abs(entry.getValue().get(pm) - existing))/existing)*100; if (change <= minPercentageChange) { ricToThroughputMapping.get(entry.getKey()).remove(pm); + log.info("Removing pm data {} for RIC {}", pm, entry.getKey()); } } } @@ -151,7 +164,7 @@ public class SnssaiSamplesProcessor { ricToCellMapping.forEach((ric,cells) -> { int sumOfPrbs = 0; for(String cell : cells) { - int index = snssaiMeasurementList.indexOf(new MeasurementObject(cell)); + int index = MeasurementObject.findIndex(cell, snssaiMeasurementList); sumOfPrbs += snssaiMeasurementList.get(index).getPmData().get(pmName); } if(ricToPrbsMapping.containsKey(ric)) { @@ -163,6 +176,7 @@ public class SnssaiSamplesProcessor { ricToPrbsMapping.put(ric, pmToPrbMapping); } }); + log.info("PRBs sum computed for RIC {}", ricToPrbsMapping); } protected Integer computeSum(String pm) { @@ -188,7 +202,6 @@ public class SnssaiSamplesProcessor { ricToThroughputMapping.put(ric, throughtputMap); } } - + log.info("Throughput computed for RIC {}", ricToThroughputMapping); } - } |