Notify all metric reporters when new types of system metric added

Change-Id: I307be0cb68bdc7fc3c75212c3fac4390fb9391a9

apps/cpman/app/src/main/java/org/onosproject/cpman/impl/SystemMetricsAggregator.java

@@ -34,7 +34,7 @@ import static org.slf4j.LoggerFactory.getLogger;
 /**
  * Aggregate system metrics.
  */
-public class SystemMetricsAggregator {
+public final class SystemMetricsAggregator {
 
     private final Logger log = getLogger(getClass());
 
@@ -42,40 +42,108 @@ public class SystemMetricsAggregator {
     private static final String DEFAULT_METER_SUFFIX = "rate";
     private static final String DISK_RESOURCE_TYPE = "disk";
     private static final String NETWORK_RESOURCE_TYPE = "network";
-    private final Map<ControlMetricType, Meter> meterMap = Maps.newHashMap();
+    private final Map<ControlMetricType, Meter> systemMap = Maps.newHashMap();
-    private final Set<ControlMetricType> metricTypeSet = Sets.newHashSet();
+    private final Map<String, Map<ControlMetricType, Meter>> diskMap = Maps.newHashMap();
+    private final Map<String, Map<ControlMetricType, Meter>> networkMap = Maps.newHashMap();
 
-    public SystemMetricsAggregator(MetricsService metricsService, Optional<String> resName, String resType) {
+    private MetricsService metricsService;
-        String resourceName = resName.isPresent() ? resName.get() : DEFAULT_RESOURCE_NAME;
-        MetricsComponent mc = metricsService.registerComponent(resourceName);
 
-        if (resName.isPresent()) {
+    public static SystemMetricsAggregator getInstance() {
-            if (DISK_RESOURCE_TYPE.equals(resType)) {
+        return SingletonHelper.INSTANCE;
-                metricTypeSet.addAll(ControlResource.DISK_METRICS);
-            } else if (NETWORK_RESOURCE_TYPE.equals(resType)) {
-                metricTypeSet.addAll(ControlResource.NETWORK_METRICS);
-            } else {
-                log.warn("Not valid resource type {}", resType);
-            }
-        } else {
-            metricTypeSet.addAll(ControlResource.MEMORY_METRICS);
-            metricTypeSet.addAll(ControlResource.CPU_METRICS);
-        }
-
-        metricTypeSet.forEach(type -> {
-            MetricsFeature metricsFeature = mc.registerFeature(type.toString());
-            Meter meter = metricsService.createMeter(mc, metricsFeature, DEFAULT_METER_SUFFIX);
-            meterMap.putIfAbsent(type, meter);
-        });
     }
 
     /**
-     * Increments metric value.
+     * Configures metric services.
      *
-     * @param type metric type
+     * @param service metrics service
+     */
+    public void setMetricsService(MetricsService service) {
+
+        metricsService = service;
+    }
+
+    /**
+     * Increments system metric value.
+     *
+     * @param type  metric type
      * @param value metric value
      */
     public void increment(ControlMetricType type, long value) {
-        meterMap.get(type).mark(value);
+        systemMap.get(type).mark(value);
+    }
+
+    /**
+     * Increments disk or network metric value.
+     *
+     * @param resourceName resource name
+     * @param resourceType resource type
+     * @param type         control metric type
+     * @param value        metric value
+     */
+    public void increment(String resourceName, String resourceType, ControlMetricType type, long value) {
+        if (DISK_RESOURCE_TYPE.equals(resourceType) && diskMap.containsKey(resourceName)) {
+            diskMap.get(resourceName).get(type).mark(value);
+        }
+
+        if (NETWORK_RESOURCE_TYPE.equals(resourceType) && networkMap.containsKey(resourceName)) {
+            networkMap.get(resourceName).get(type).mark(value);
+        }
+    }
+
+    /**
+     * Adds a set of new monitoring metric types.
+     *
+     * @param optResourceName optional resource name, null denotes system metric
+     * @param resType         resource type
+     */
+    public void addMetrics(Optional<String> optResourceName, String resType) {
+        Set<ControlMetricType> metricTypeSet = Sets.newHashSet();
+        String resourceName = optResourceName.isPresent() ?
+                optResourceName.get() : DEFAULT_RESOURCE_NAME;
+
+        MetricsComponent metricsComponent = metricsService.registerComponent(resourceName);
+
+        if (optResourceName.isPresent()) {
+            if (!diskMap.containsKey(resourceName) && DISK_RESOURCE_TYPE.equals(resType)) {
+                metricTypeSet.addAll(ControlResource.DISK_METRICS);
+                diskMap.putIfAbsent(resourceName,
+                        getMeterMap(metricTypeSet, metricsComponent, metricsService));
+                metricsService.notifyReporters();
+            } else if (!networkMap.containsKey(resourceName) && NETWORK_RESOURCE_TYPE.equals(resType)) {
+                metricTypeSet.addAll(ControlResource.NETWORK_METRICS);
+                networkMap.putIfAbsent(resourceName,
+                        getMeterMap(metricTypeSet, metricsComponent, metricsService));
+                metricsService.notifyReporters();
+            } else {
+                return;
+            }
+        } else {
+            if (systemMap.isEmpty()) {
+                metricTypeSet.addAll(ControlResource.MEMORY_METRICS);
+                metricTypeSet.addAll(ControlResource.CPU_METRICS);
+
+                systemMap.putAll(getMeterMap(metricTypeSet, metricsComponent, metricsService));
+                metricsService.notifyReporters();
+            }
+        }
+    }
+
+    private Map<ControlMetricType, Meter> getMeterMap(Set<ControlMetricType> types,
+                                                      MetricsComponent component,
+                                                      MetricsService service) {
+        Map<ControlMetricType, Meter> meterMap = Maps.newHashMap();
+        types.forEach(type -> {
+            MetricsFeature metricsFeature = component.registerFeature(type.toString());
+            Meter meter = service.createMeter(component, metricsFeature, DEFAULT_METER_SUFFIX);
+            meterMap.putIfAbsent(type, meter);
+        });
+        return meterMap;
+    }
+
+    private SystemMetricsAggregator() {
+    }
+
+    private static class SingletonHelper {
+        private static final SystemMetricsAggregator INSTANCE = new SystemMetricsAggregator();
     }
 }

apps/cpman/app/src/main/java/org/onosproject/cpman/rest/SystemMetricsCollectorWebResource.java

@@ -63,6 +63,9 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
     private static final String INVALID_RESOURCE_NAME = "Invalid resource name";
     private static final String INVALID_REQUEST = "Invalid request";
     private static final int PERCENT_CONSTANT = 100;
+    private static final String SYSTEM_TYPE = "system";
+    private static final String DISK_TYPE = "disk";
+    private static final String NETWORK_TYPE = "network";
 
     private static final Set<String> MEMORY_FIELD_SET = ControlResource.MEMORY_METRICS
             .stream().map(type -> toCamelCase(type.toString(), true))
@@ -72,8 +75,7 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
             .stream().map(type -> toCamelCase(type.toString(), true))
             .collect(Collectors.toSet());
 
-    private SystemMetricsAggregator systemAggr =
+    private SystemMetricsAggregator aggregator = SystemMetricsAggregator.getInstance();
-            new SystemMetricsAggregator(metricsService, Optional.ofNullable(null), "system");
 
     /**
      * Collects CPU metrics.
@@ -103,30 +105,33 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
             long userCpuTime = nullIsIllegal(jsonTree.get("userCpuTime").asLong(), INVALID_REQUEST);
             long cpuIdleTime = nullIsIllegal(jsonTree.get("cpuIdleTime").asLong(), INVALID_REQUEST);
 
+            aggregator.setMetricsService(metricsService);
+            aggregator.addMetrics(Optional.ofNullable(null), SYSTEM_TYPE);
+
             cm = new ControlMetric(ControlMetricType.CPU_LOAD,
                     new MetricValue.Builder().load(cpuLoad).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.CPU_LOAD, cpuLoad);
+            aggregator.increment(ControlMetricType.CPU_LOAD, cpuLoad);
 
             cm = new ControlMetric(ControlMetricType.TOTAL_CPU_TIME,
                     new MetricValue.Builder().load(totalCpuTime).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.TOTAL_CPU_TIME, totalCpuTime);
+            aggregator.increment(ControlMetricType.TOTAL_CPU_TIME, totalCpuTime);
 
             cm = new ControlMetric(ControlMetricType.SYS_CPU_TIME,
                     new MetricValue.Builder().load(sysCpuTime).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.SYS_CPU_TIME, sysCpuTime);
+            aggregator.increment(ControlMetricType.SYS_CPU_TIME, sysCpuTime);
 
             cm = new ControlMetric(ControlMetricType.USER_CPU_TIME,
                     new MetricValue.Builder().load(userCpuTime).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.USER_CPU_TIME, userCpuTime);
+            aggregator.increment(ControlMetricType.USER_CPU_TIME, userCpuTime);
 
             cm = new ControlMetric(ControlMetricType.CPU_IDLE_TIME,
                     new MetricValue.Builder().load(cpuIdleTime).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.CPU_IDLE_TIME, cpuIdleTime);
+            aggregator.increment(ControlMetricType.CPU_IDLE_TIME, cpuIdleTime);
 
         } catch (IOException e) {
             throw new IllegalArgumentException(e.getMessage());
@@ -161,25 +166,28 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
             long memUsedRatio = memTotal == 0L ? 0L : (memUsed * PERCENT_CONSTANT) / memTotal;
             long memFreeRatio = memTotal == 0L ? 0L : (memFree * PERCENT_CONSTANT) / memTotal;
 
+            aggregator.setMetricsService(metricsService);
+            aggregator.addMetrics(Optional.ofNullable(null), SYSTEM_TYPE);
+
             cm = new ControlMetric(ControlMetricType.MEMORY_USED_RATIO,
                     new MetricValue.Builder().load(memUsedRatio).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.MEMORY_USED_RATIO, memUsedRatio);
+            aggregator.increment(ControlMetricType.MEMORY_USED_RATIO, memUsedRatio);
 
             cm = new ControlMetric(ControlMetricType.MEMORY_FREE_RATIO,
                     new MetricValue.Builder().load(memFreeRatio).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.MEMORY_FREE_RATIO, memFreeRatio);
+            aggregator.increment(ControlMetricType.MEMORY_FREE_RATIO, memFreeRatio);
 
             cm = new ControlMetric(ControlMetricType.MEMORY_USED,
                     new MetricValue.Builder().load(memUsed).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.MEMORY_USED, memUsed);
+            aggregator.increment(ControlMetricType.MEMORY_USED, memUsed);
 
             cm = new ControlMetric(ControlMetricType.MEMORY_FREE,
                     new MetricValue.Builder().load(memFree).add());
             monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
-            systemAggr.increment(ControlMetricType.MEMORY_FREE, memFree);
+            aggregator.increment(ControlMetricType.MEMORY_FREE, memFree);
 
         } catch (IOException e) {
             throw new IllegalArgumentException(e.getMessage());
@@ -204,14 +212,15 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
         try {
             ObjectNode jsonTree = (ObjectNode) mapper().readTree(stream);
             ArrayNode diskRes =
-                    jsonTree.get("disks") == null ? mapper().createArrayNode() : (ArrayNode) jsonTree.get("disks");
+                    jsonTree.get("disks") == null ?
+                            mapper().createArrayNode() : (ArrayNode) jsonTree.get("disks");
 
             for (JsonNode node : diskRes) {
                 JsonNode resourceName = node.get("resourceName");
                 nullIsIllegal(resourceName, INVALID_RESOURCE_NAME);
 
-                SystemMetricsAggregator diskAggr = new SystemMetricsAggregator(metricsService,
+                aggregator.setMetricsService(metricsService);
-                        Optional.of(resourceName.asText()), "disk");
+                aggregator.addMetrics(Optional.of(resourceName.asText()), DISK_TYPE);
 
                 long readBytes = nullIsIllegal(node.get("readBytes").asLong(), INVALID_REQUEST);
                 long writeBytes = nullIsIllegal(node.get("writeBytes").asLong(), INVALID_REQUEST);
@@ -219,12 +228,13 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
                 cm = new ControlMetric(ControlMetricType.DISK_READ_BYTES,
                         new MetricValue.Builder().load(readBytes).add());
                 monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, resourceName.asText());
-                diskAggr.increment(ControlMetricType.DISK_READ_BYTES, readBytes);
+                aggregator.increment(resourceName.asText(), DISK_TYPE,
-
+                        ControlMetricType.DISK_READ_BYTES, readBytes);
                 cm = new ControlMetric(ControlMetricType.DISK_WRITE_BYTES,
                         new MetricValue.Builder().load(writeBytes).add());
                 monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, resourceName.asText());
-                diskAggr.increment(ControlMetricType.DISK_WRITE_BYTES, writeBytes);
+                aggregator.increment(resourceName.asText(), DISK_TYPE,
+                        ControlMetricType.DISK_WRITE_BYTES, writeBytes);
             }
         } catch (IOException e) {
             throw new IllegalArgumentException(e.getMessage());
@@ -256,8 +266,8 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
                 JsonNode resourceName = node.get("resourceName");
                 nullIsIllegal(resourceName, INVALID_RESOURCE_NAME);
 
-                SystemMetricsAggregator networkAggr = new SystemMetricsAggregator(metricsService,
+                aggregator.setMetricsService(metricsService);
-                        Optional.of(resourceName.asText()), "network");
+                aggregator.addMetrics(Optional.of(resourceName.asText()), NETWORK_TYPE);
 
                 long inBytes = nullIsIllegal(node.get("incomingBytes").asLong(), INVALID_REQUEST);
                 long outBytes = nullIsIllegal(node.get("outgoingBytes").asLong(), INVALID_REQUEST);
@@ -267,22 +277,26 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
                 cm = new ControlMetric(ControlMetricType.NW_INCOMING_BYTES,
                         new MetricValue.Builder().load(inBytes).add());
                 monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, resourceName.asText());
-                networkAggr.increment(ControlMetricType.NW_INCOMING_BYTES, inBytes);
+                aggregator.increment(resourceName.asText(), NETWORK_TYPE,
+                        ControlMetricType.NW_INCOMING_BYTES, inBytes);
 
                 cm = new ControlMetric(ControlMetricType.NW_OUTGOING_BYTES,
                         new MetricValue.Builder().load(outBytes).add());
                 monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, resourceName.asText());
-                networkAggr.increment(ControlMetricType.NW_OUTGOING_BYTES, outBytes);
+                aggregator.increment(resourceName.asText(), NETWORK_TYPE,
+                        ControlMetricType.NW_OUTGOING_BYTES, outBytes);
 
                 cm = new ControlMetric(ControlMetricType.NW_INCOMING_PACKETS,
                         new MetricValue.Builder().load(inPackets).add());
                 monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, resourceName.asText());
-                networkAggr.increment(ControlMetricType.NW_INCOMING_PACKETS, inPackets);
+                aggregator.increment(resourceName.asText(), NETWORK_TYPE,
+                        ControlMetricType.NW_INCOMING_PACKETS, inPackets);
 
                 cm = new ControlMetric(ControlMetricType.NW_OUTGOING_PACKETS,
                         new MetricValue.Builder().load(outPackets).add());
                 monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, resourceName.asText());
-                networkAggr.increment(ControlMetricType.NW_OUTGOING_PACKETS, outPackets);
+                aggregator.increment(resourceName.asText(), NETWORK_TYPE,
+                        ControlMetricType.NW_OUTGOING_PACKETS, outPackets);
             }
         } catch (IOException e) {
             throw new IllegalArgumentException(e.getMessage());

apps/cpman/app/src/test/java/org/onosproject/cpman/rest/ControlMetricsCollectorResourceTest.java

@@ -36,7 +36,6 @@ import org.onlab.rest.BaseResource;
 import org.onosproject.cpman.ControlPlaneMonitorService;
 import org.onosproject.cpman.SystemInfo;
 import org.onosproject.cpman.impl.SystemInfoFactory;
-import org.onosproject.cpman.impl.SystemMetricsAggregator;
 import org.onosproject.net.DeviceId;
 import org.onosproject.rest.resources.ResourceTest;
 
@@ -51,7 +50,6 @@ import java.util.Map;
 import java.util.Optional;
 
 import static org.easymock.EasyMock.anyInt;
-import static org.easymock.EasyMock.anyLong;
 import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.anyString;
 import static org.easymock.EasyMock.createMock;
@@ -69,9 +67,6 @@ public class ControlMetricsCollectorResourceTest extends ResourceTest {
     final ControlPlaneMonitorService mockControlPlaneMonitorService =
                                      createMock(ControlPlaneMonitorService.class);
     final MetricsService mockMetricsService = new MockMetricsService();
-    final MetricsComponent mockMetricsComponent = createMock(MetricsComponent.class);
-    final SystemMetricsAggregator mockAggregator = createMock(SystemMetricsAggregator.class);
-
 
     private static final String PREFIX = "collector";
 
@@ -104,10 +99,6 @@ public class ControlMetricsCollectorResourceTest extends ResourceTest {
         expectLastCall().times(5);
         replay(mockControlPlaneMonitorService);
 
-        mockAggregator.increment(anyObject(), anyLong());
-        expectLastCall();
-        replay(mockAggregator);
-
         basePostTest("cpu-metrics-post.json", PREFIX + "/cpu_metrics");
     }