Implement updateMetric and getLoad methods of ControlPlaneMonitor

- Add rrd4j jar, and wrap it as a bundle for karaf
- Implement updateMetric and getLoad methods
- Add unit test for two methods
- Revise the DefaultMetricDatabase to make it generate unique in
  memory storage space
- Revise the ControlPlaneMonitor interface
- Rename percentage to ratio, due to long string unsupport
  issue in RRD

Change-Id: Ia9d56f8e4f4bcd7ef7a29732668caa9c6a885ecf

@@ -28,11 +28,11 @@ public class ControlMetric {
         this.metricValue = metricValue;
     }
 
-    ControlMetricType metricType() {
+    public ControlMetricType metricType() {
         return metricType;
     }
 
-    MetricValue metricValue() {
+    public MetricValue metricValue() {
         return metricValue;
     }
 }

@@ -62,11 +62,11 @@ public enum ControlMetricType {
     /* CPU Idle Time. **/
     CPU_IDLE_TIME,
 
-    /* Percentage of Used Memory Amount. */
-    MEMORY_USED_PERCENTAGE,
+    /* Ratio of Used Memory Amount. */
+    MEMORY_USED_RATIO,
 
-    /* Percentage of Free Memory Amount. **/
-    MEMORY_FREE_PERCENTAGE,
+    /* Ratio of Free Memory Amount. **/
+    MEMORY_FREE_RATIO,
 
     /* Used Memory Amount. **/
     MEMORY_USED,

@@ -19,7 +19,6 @@ import org.onosproject.cluster.NodeId;
 import org.onosproject.net.DeviceId;
 
 import java.util.Optional;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Control Plane Statistics Service Interface.
@@ -46,6 +45,8 @@ public interface ControlPlaneMonitorService {
 
     /**
      * Obtains the control plane load of a specific device.
+     * The metrics range from control messages and system metrics
+     * (e.g., CPU and memory info)
      *
      * @param nodeId   node id {@link org.onosproject.cluster.NodeId}
      * @param type     control metric type
@@ -55,15 +56,13 @@ public interface ControlPlaneMonitorService {
     ControlLoad getLoad(NodeId nodeId, ControlMetricType type, Optional<DeviceId> deviceId);
 
     /**
-     * Obtains the control plane load of a specific device with a specific time duration.
+     * Obtains the control plane load of a specific device.
+     * The metrics range from I/O device metrics (e.g., disk and network interface)
      *
      * @param nodeId        node id {@link org.onosproject.cluster.NodeId}
      * @param type          control metric type
-     * @param duration time duration
-     * @param unit     time unit
-     * @param deviceId device id {@link org.onosproject.net.Device}
+     * @param resourceName  resource name
      * @return control plane load
      */
-    ControlLoad getLoad(NodeId nodeId, ControlMetricType type, Optional<DeviceId> deviceId,
-                        int duration, TimeUnit unit);
+    ControlLoad getLoad(NodeId nodeId, ControlMetricType type, String resourceName);
 }
\ No newline at end of file

@@ -20,5 +20,6 @@
         <feature>onos-api</feature>
         <bundle>mvn:${project.groupId}/onos-app-cpman-api/${project.version}</bundle>
         <bundle>mvn:${project.groupId}/onos-app-cpman/${project.version}</bundle>
+        <bundle>wrap:mvn:org.rrd4j/rrd4j/2.2$Bundle-SymbolicName=rrd4j&amp;Bundle-Version=2.2</bundle>
     </feature>
 </features>

@@ -108,6 +108,7 @@
             <groupId>org.rrd4j</groupId>
             <artifactId>rrd4j</artifactId>
             <version>2.2</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>com.sun.jersey</groupId>
@@ -184,7 +185,8 @@
                             org.onlab.rest.*,
                             org.onosproject.*,
                             org.onlab.util.*,
-                            org.jboss.netty.util.*
+                            org.jboss.netty.util.*,
+                            org.rrd4j.*
                         </Import-Package>
                         <Web-ContextPath>${web.context}</Web-ContextPath>
                     </instructions>

@@ -48,8 +48,8 @@ public final class ControlMetricsFactory {
     private MetricsAggregator cpuIdleTime;
     private MetricsAggregator memoryUsed;
     private MetricsAggregator memoryFree;
-    private MetricsAggregator memoryUsedPercentage;
-    private MetricsAggregator memoryFreePercentage;
+    private MetricsAggregator memoryUsedRatio;
+    private MetricsAggregator memoryFreeRatio;
     private Map<String, MetricsAggregator> diskReadBytes;
     private Map<String, MetricsAggregator> diskWriteBytes;
     private Map<String, MetricsAggregator> nwIncomingBytes;
@@ -283,10 +283,10 @@ public final class ControlMetricsFactory {
         /* Memory */
         memoryFree = new MetricsAggregator(metricsService, ControlMetricType.MEMORY_FREE);
         memoryUsed = new MetricsAggregator(metricsService, ControlMetricType.MEMORY_USED);
-        memoryFreePercentage = new MetricsAggregator(metricsService,
-                                        ControlMetricType.MEMORY_FREE_PERCENTAGE);
-        memoryUsedPercentage = new MetricsAggregator(metricsService,
-                                        ControlMetricType.MEMORY_USED_PERCENTAGE);
+        memoryFreeRatio = new MetricsAggregator(metricsService,
+                                        ControlMetricType.MEMORY_FREE_RATIO);
+        memoryUsedRatio = new MetricsAggregator(metricsService,
+                                        ControlMetricType.MEMORY_USED_RATIO);
 
         /* Disk I/O */
         diskReadBytes = new ConcurrentHashMap<>();
@@ -350,12 +350,12 @@ public final class ControlMetricsFactory {
         return cpuIdleTime;
     }
 
-    public MetricsAggregator memoryFreePercentage() {
-        return memoryFreePercentage;
+    public MetricsAggregator memoryFreeRatio() {
+        return memoryFreeRatio;
     }
 
-    public MetricsAggregator memoryUsedPercentage() {
-        return memoryUsedPercentage;
+    public MetricsAggregator memoryUsedRatio() {
+        return memoryUsedRatio;
     }
 
     public MetricsAggregator diskReadBytes(String partitionName) {

@@ -53,5 +53,4 @@ public class ControlPlaneManager {
     protected void deactivate() {
         log.info("Stopped");
     }
-
 }
\ No newline at end of file

@@ -15,22 +15,50 @@
  */
 package org.onosproject.cpman.impl;
 
+import com.google.common.collect.ImmutableSet;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.ControllerNode;
 import org.onosproject.cluster.NodeId;
 import org.onosproject.cpman.ControlLoad;
 import org.onosproject.cpman.ControlMetric;
 import org.onosproject.cpman.ControlMetricType;
 import org.onosproject.cpman.ControlPlaneMonitorService;
+import org.onosproject.cpman.MetricsDatabase;
 import org.onosproject.net.DeviceId;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ConcurrentHashMap;
 
-import static org.slf4j.LoggerFactory.getLogger;
+import static org.onosproject.cpman.ControlMetricType.CPU_IDLE_TIME;
+import static org.onosproject.cpman.ControlMetricType.CPU_LOAD;
+import static org.onosproject.cpman.ControlMetricType.DISK_READ_BYTES;
+import static org.onosproject.cpman.ControlMetricType.DISK_WRITE_BYTES;
+import static org.onosproject.cpman.ControlMetricType.FLOW_MOD_PACKET;
+import static org.onosproject.cpman.ControlMetricType.FLOW_REMOVED_PACKET;
+import static org.onosproject.cpman.ControlMetricType.INBOUND_PACKET;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_FREE;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_FREE_RATIO;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_USED;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_USED_RATIO;
+import static org.onosproject.cpman.ControlMetricType.NW_INCOMING_BYTES;
+import static org.onosproject.cpman.ControlMetricType.NW_INCOMING_PACKETS;
+import static org.onosproject.cpman.ControlMetricType.NW_OUTGOING_BYTES;
+import static org.onosproject.cpman.ControlMetricType.NW_OUTGOING_PACKETS;
+import static org.onosproject.cpman.ControlMetricType.OUTBOUND_PACKET;
+import static org.onosproject.cpman.ControlMetricType.REPLY_PACKET;
+import static org.onosproject.cpman.ControlMetricType.REQUEST_PACKET;
+import static org.onosproject.cpman.ControlMetricType.SYS_CPU_TIME;
+import static org.onosproject.cpman.ControlMetricType.TOTAL_CPU_TIME;
+import static org.onosproject.cpman.ControlMetricType.USER_CPU_TIME;
 
 /**
  * Control plane monitoring service class.
@@ -39,37 +67,219 @@ import static org.slf4j.LoggerFactory.getLogger;
 @Service
 public class ControlPlaneMonitor implements ControlPlaneMonitorService {
 
-    private final Logger log = getLogger(getClass());
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private MetricsDatabase cpuMetrics;
+    private MetricsDatabase memoryMetrics;
+    private Map<DeviceId, MetricsDatabase> controlMessageMap;
+    private Map<String, MetricsDatabase> diskMetricsMap;
+    private Map<String, MetricsDatabase> networkMetricsMap;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected ClusterService clusterService;
+
+    private static final String CPU = "Cpu";
+    private static final String MEMORY = "Memory";
+    private static final String CTRL_MSG = "ControlMessage";
+    private static final String DISK = "Disk";
+    private static final String NETWORK = "Network";
+
+    private static final ImmutableSet<ControlMetricType> CPU_METRICS =
+            ImmutableSet.of(CPU_IDLE_TIME, CPU_LOAD, SYS_CPU_TIME,
+                    USER_CPU_TIME, TOTAL_CPU_TIME);
+    private static final ImmutableSet<ControlMetricType> MEMORY_METRICS =
+            ImmutableSet.of(MEMORY_FREE, MEMORY_FREE_RATIO, MEMORY_USED,
+                    MEMORY_USED_RATIO);
+    private static final ImmutableSet<ControlMetricType> DISK_METRICS =
+            ImmutableSet.of(DISK_READ_BYTES, DISK_WRITE_BYTES);
+    private static final ImmutableSet<ControlMetricType> NETWORK_METRICS =
+            ImmutableSet.of(NW_INCOMING_BYTES, NW_OUTGOING_BYTES,
+                    NW_INCOMING_PACKETS, NW_OUTGOING_PACKETS);
+    private static final ImmutableSet<ControlMetricType> CTRL_MSGS =
+            ImmutableSet.of(INBOUND_PACKET, OUTBOUND_PACKET, FLOW_MOD_PACKET,
+                    FLOW_REMOVED_PACKET, REQUEST_PACKET, REPLY_PACKET);
+    private Map<ControlMetricType, Double> cpuBuf;
+    private Map<ControlMetricType, Double> memoryBuf;
+    private Map<String, Map<ControlMetricType, Double>> diskBuf;
+    private Map<String, Map<ControlMetricType, Double>> networkBuf;
+    private Map<DeviceId, Map<ControlMetricType, Double>> ctrlMsgBuf;
 
     @Activate
     public void activate() {
+        cpuMetrics = genMDbBuilder(CPU, CPU_METRICS);
+        memoryMetrics = genMDbBuilder(MEMORY, MEMORY_METRICS);
+        controlMessageMap = new ConcurrentHashMap<>();
+        diskMetricsMap = new ConcurrentHashMap<>();
+        networkMetricsMap = new ConcurrentHashMap<>();
+
+        cpuBuf = new ConcurrentHashMap<>();
+        memoryBuf = new ConcurrentHashMap<>();
+        diskBuf = new ConcurrentHashMap<>();
+        networkBuf = new ConcurrentHashMap<>();
+        ctrlMsgBuf = new ConcurrentHashMap<>();
+
+        log.info("Started");
     }
 
     @Deactivate
     public void deactivate() {
+
+        // TODO: need to handle the mdb close.
+        cpuBuf.clear();
+        memoryBuf.clear();
+        diskBuf.clear();
+        networkBuf.clear();
+        ctrlMsgBuf.clear();
+
+        log.info("Stopped");
     }
 
     @Override
-    public void updateMetric(ControlMetric cpm, Integer updateInterval,
+    public void updateMetric(ControlMetric cm, Integer updateInterval,
                              Optional<DeviceId> deviceId) {
+        if (deviceId.isPresent()) {
+
+            // insert a new device entry if we cannot find any
+            ctrlMsgBuf.putIfAbsent(deviceId.get(), new ConcurrentHashMap<>());
+
+            // update control message metrics
+            if (CTRL_MSGS.contains(cm.metricType())) {
+
+                // we will accumulate the metric value into buffer first
+                ctrlMsgBuf.get(deviceId.get()).putIfAbsent(cm.metricType(),
+                        (double) cm.metricValue().getLoad());
+
+                // if buffer contains all control message metrics,
+                // we simply set and update the values into MetricsDatabase.
+                if (ctrlMsgBuf.get(deviceId.get()).keySet().containsAll(CTRL_MSGS)) {
+                    updateControlMessages(ctrlMsgBuf.get(deviceId.get()), deviceId.get());
+                    ctrlMsgBuf.get(deviceId.get()).clear();
+                }
+            }
+        } else {
+
+            // update cpu metrics
+            if (CPU_METRICS.contains(cm.metricType())) {
+                cpuBuf.putIfAbsent(cm.metricType(),
+                        (double) cm.metricValue().getLoad());
+                if (cpuBuf.keySet().containsAll(CPU_METRICS)) {
+                    cpuMetrics.updateMetrics(convertMap(cpuBuf));
+                    cpuBuf.clear();
+                }
+            }
 
+            // update memory metrics
+            if (MEMORY_METRICS.contains(cm.metricType())) {
+                memoryBuf.putIfAbsent(cm.metricType(),
+                        (double) cm.metricValue().getLoad());
+                if (memoryBuf.keySet().containsAll(MEMORY_METRICS)) {
+                    memoryMetrics.updateMetrics(convertMap(memoryBuf));
+                    memoryBuf.clear();
+                }
+            }
+        }
     }
 
     @Override
-    public void updateMetric(ControlMetric controlMetric, Integer updateInterval,
+    public void updateMetric(ControlMetric cm, Integer updateInterval,
                              String resourceName) {
+        // update disk metrics
+        if (DISK_METRICS.contains(cm.metricType())) {
+            diskBuf.putIfAbsent(resourceName, new ConcurrentHashMap<>());
+            diskBuf.get(resourceName).putIfAbsent(cm.metricType(),
+                    (double) cm.metricValue().getLoad());
+            if (diskBuf.get(resourceName).keySet().containsAll(DISK_METRICS)) {
+                updateDiskMetrics(diskBuf.get(resourceName), resourceName);
+                diskBuf.clear();
+            }
+        }
 
+        // update network metrics
+        if (NETWORK_METRICS.contains(cm.metricType())) {
+            networkBuf.putIfAbsent(resourceName, new ConcurrentHashMap<>());
+            networkBuf.get(resourceName).putIfAbsent(cm.metricType(),
+                    (double) cm.metricValue().getLoad());
+            if (networkBuf.get(resourceName).keySet().containsAll(NETWORK_METRICS)) {
+                updateNetworkMetrics(networkBuf.get(resourceName), resourceName);
+                networkBuf.clear();
+            }
+        }
     }
 
     @Override
     public ControlLoad getLoad(NodeId nodeId, ControlMetricType type,
                                Optional<DeviceId> deviceId) {
+        ControllerNode node = clusterService.getNode(nodeId);
+        if (clusterService.getLocalNode().equals(node)) {
+
+            if (deviceId.isPresent()) {
+                if (CTRL_MSGS.contains(type)) {
+                    return new DefaultControlLoad(controlMessageMap.get(deviceId.get()), type);
+                }
+            } else {
+                // returns controlLoad of CPU metrics
+                if (CPU_METRICS.contains(type)) {
+                    return new DefaultControlLoad(cpuMetrics, type);
+                }
+
+                // returns memoryLoad of memory metrics
+                if (MEMORY_METRICS.contains(type)) {
+                    return new DefaultControlLoad(memoryMetrics, type);
+                }
+            }
+        } else {
+            // TODO: currently only query the metrics of local node
+            return null;
+        }
         return null;
     }
 
     @Override
     public ControlLoad getLoad(NodeId nodeId, ControlMetricType type,
-                               Optional<DeviceId> deviceId, int duration, TimeUnit unit) {
+                               String resourceName) {
+        if (clusterService.getLocalNode().id().equals(nodeId)) {
+            if (DISK_METRICS.contains(type)) {
+                return new DefaultControlLoad(diskMetricsMap.get(resourceName), type);
+            }
+
+            if (NETWORK_METRICS.contains(type)) {
+                return new DefaultControlLoad(networkMetricsMap.get(resourceName), type);
+            }
+        } else {
+            // TODO: currently only query the metrics of local node
             return null;
         }
+        return null;
+    }
+
+    private MetricsDatabase genMDbBuilder(String metricName,
+                                          ImmutableSet<ControlMetricType> metricTypes) {
+        MetricsDatabase.Builder builder = new DefaultMetricsDatabase.Builder();
+        builder.withMetricName(metricName);
+        metricTypes.forEach(type -> builder.addMetricType(type.toString()));
+        return builder.build();
+    }
+
+    private void updateNetworkMetrics(Map<ControlMetricType, Double> metricMap,
+                                      String resName) {
+        networkMetricsMap.putIfAbsent(resName, genMDbBuilder(NETWORK, NETWORK_METRICS));
+        networkMetricsMap.get(resName).updateMetrics(convertMap(metricMap));
+    }
+
+    private void updateDiskMetrics(Map<ControlMetricType, Double> metricMap,
+                                   String resName) {
+        diskMetricsMap.putIfAbsent(resName, genMDbBuilder(DISK, DISK_METRICS));
+        diskMetricsMap.get(resName).updateMetrics(convertMap(metricMap));
+    }
+
+    private void updateControlMessages(Map<ControlMetricType, Double> metricMap,
+                                       DeviceId devId) {
+        controlMessageMap.putIfAbsent(devId, genMDbBuilder(CTRL_MSG, CTRL_MSGS));
+        controlMessageMap.get(devId).updateMetrics(convertMap(metricMap));
+    }
+
+    private Map convertMap(Map<ControlMetricType, Double> map) {
+        Map newMap = new ConcurrentHashMap<>();
+        map.forEach((k, v) -> newMap.putIfAbsent(k.toString(), v));
+        return newMap;
+    }
 }
\ No newline at end of file

@@ -243,10 +243,6 @@ public final class DefaultMetricsDatabase implements MetricsDatabase {
         private String metricName;
 
         public Builder() {
-
-            // define the resolution of monitored metrics
-            rrdDef = new RrdDef(DB_PATH, RESOLUTION);
-
             // initialize data source definition list
             dsDefs = new ArrayList<>();
         }
@@ -254,6 +250,9 @@ public final class DefaultMetricsDatabase implements MetricsDatabase {
         @Override
         public Builder withMetricName(String metricName) {
             this.metricName = metricName;
+
+            // define the resolution of monitored metrics
+            rrdDef = new RrdDef(DB_PATH + "_" + metricName, RESOLUTION);
             return this;
         }
 

@@ -122,20 +122,20 @@ public class ControlMetricsCollectorWebResource extends AbstractWebResource {
         ControlMetric cm;
         try {
             ObjectNode jsonTree = (ObjectNode) mapper().readTree(stream);
-            JsonNode memUsedPerc = jsonTree.get("memoryUsedPercentage");
-            JsonNode memFreePerc = jsonTree.get("memoryFreePercentage");
+            JsonNode memUsedRatio = jsonTree.get("memoryUsedRatio");
+            JsonNode memFreeRatio = jsonTree.get("memoryFreeRatio");
             JsonNode memUsed = jsonTree.get("memoryUsed");
             JsonNode memFree = jsonTree.get("memoryFree");
 
-            if (memUsedPerc != null) {
-                cm = new ControlMetric(ControlMetricType.MEMORY_USED_PERCENTAGE,
-                        new MetricValue.Builder().load(memUsedPerc.asLong()).add());
+            if (memUsedRatio != null) {
+                cm = new ControlMetric(ControlMetricType.MEMORY_USED_RATIO,
+                        new MetricValue.Builder().load(memUsedRatio.asLong()).add());
                 service.updateMetric(cm, UPDATE_INTERVAL, Optional.ofNullable(null));
             }
 
-            if (memFreePerc != null) {
-                cm = new ControlMetric(ControlMetricType.MEMORY_FREE_PERCENTAGE,
-                        new MetricValue.Builder().load(memFreePerc.asLong()).add());
+            if (memFreeRatio != null) {
+                cm = new ControlMetric(ControlMetricType.MEMORY_FREE_RATIO,
+                        new MetricValue.Builder().load(memFreeRatio.asLong()).add());
                 service.updateMetric(cm, UPDATE_INTERVAL, Optional.ofNullable(null));
             }
 

 {
   "type": "object",
   "required": [
-    "memoryUsedPercentage",
-    "memoryFreePercentage",
+    "memoryUsedRatio",
+    "memoryFreeRatio",
     "memoryUsed",
     "memoryFree"
   ],
   "properties": {
-    "memoryUsedPercentage": {
+    "memoryUsedRatio": {
       "type": "integer",
       "format": "int64",
       "example": "30"
     },
-    "memoryFreePercentage": {
+    "memoryFreeRatio": {
       "type": "integer",
       "format": "int64",
       "example": "70"

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * 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.
+ */
+package org.onosproject.cpman.impl;
+
+import com.google.common.collect.ImmutableSet;
+import org.junit.Before;
+import org.junit.Test;
+import org.onlab.packet.IpAddress;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.ControllerNode;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.cpman.ControlMetric;
+import org.onosproject.cpman.ControlMetricType;
+import org.onosproject.cpman.MetricValue;
+import org.onosproject.net.DeviceId;
+
+import java.util.Optional;
+
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.onosproject.cpman.ControlMetricType.CPU_IDLE_TIME;
+import static org.onosproject.cpman.ControlMetricType.CPU_LOAD;
+import static org.onosproject.cpman.ControlMetricType.DISK_READ_BYTES;
+import static org.onosproject.cpman.ControlMetricType.DISK_WRITE_BYTES;
+import static org.onosproject.cpman.ControlMetricType.FLOW_MOD_PACKET;
+import static org.onosproject.cpman.ControlMetricType.FLOW_REMOVED_PACKET;
+import static org.onosproject.cpman.ControlMetricType.INBOUND_PACKET;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_FREE;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_FREE_RATIO;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_USED;
+import static org.onosproject.cpman.ControlMetricType.MEMORY_USED_RATIO;
+import static org.onosproject.cpman.ControlMetricType.NW_INCOMING_BYTES;
+import static org.onosproject.cpman.ControlMetricType.NW_INCOMING_PACKETS;
+import static org.onosproject.cpman.ControlMetricType.NW_OUTGOING_BYTES;
+import static org.onosproject.cpman.ControlMetricType.NW_OUTGOING_PACKETS;
+import static org.onosproject.cpman.ControlMetricType.OUTBOUND_PACKET;
+import static org.onosproject.cpman.ControlMetricType.REPLY_PACKET;
+import static org.onosproject.cpman.ControlMetricType.REQUEST_PACKET;
+import static org.onosproject.cpman.ControlMetricType.SYS_CPU_TIME;
+import static org.onosproject.cpman.ControlMetricType.TOTAL_CPU_TIME;
+import static org.onosproject.cpman.ControlMetricType.USER_CPU_TIME;
+
+/**
+ * Unit test of control plane monitoring service.
+ */
+public class ControlPlaneMonitorTest {
+
+    private ControlPlaneMonitor monitor;
+    private static final Integer UPDATE_INTERVAL = 1;
+    private ClusterService mockClusterService;
+    private ControllerNode mockControllerNode;
+    private NodeId nodeId;
+    private static final ImmutableSet<ControlMetricType> CPU_METRICS =
+            ImmutableSet.of(CPU_IDLE_TIME, CPU_LOAD, SYS_CPU_TIME,
+                    USER_CPU_TIME, TOTAL_CPU_TIME);
+    private static final ImmutableSet<ControlMetricType> MEMORY_METRICS =
+            ImmutableSet.of(MEMORY_FREE, MEMORY_FREE_RATIO, MEMORY_USED,
+                    MEMORY_USED_RATIO);
+    private static final ImmutableSet<ControlMetricType> DISK_METRICS =
+            ImmutableSet.of(DISK_READ_BYTES, DISK_WRITE_BYTES);
+    private static final ImmutableSet<ControlMetricType> NETWORK_METRICS =
+            ImmutableSet.of(NW_INCOMING_BYTES, NW_OUTGOING_BYTES,
+                    NW_INCOMING_PACKETS, NW_OUTGOING_PACKETS);
+    private static final ImmutableSet<ControlMetricType> CTRL_MSGS =
+            ImmutableSet.of(INBOUND_PACKET, OUTBOUND_PACKET, FLOW_MOD_PACKET,
+                    FLOW_REMOVED_PACKET, REQUEST_PACKET, REPLY_PACKET);
+
+    @Before
+    public void setup() throws Exception {
+        monitor = new ControlPlaneMonitor();
+        monitor.activate();
+
+        nodeId = new NodeId("1");
+        mockControllerNode = new MockControllerNode(nodeId);
+        mockClusterService = createMock(ClusterService.class);
+        monitor.clusterService = mockClusterService;
+
+        expect(mockClusterService.getNode(anyObject()))
+                .andReturn(mockControllerNode).anyTimes();
+        expect(mockClusterService.getLocalNode())
+                .andReturn(mockControllerNode).anyTimes();
+        replay(mockClusterService);
+    }
+
+    /**
+     * Mock class for a controller node.
+     */
+    private static class MockControllerNode implements ControllerNode {
+        final NodeId id;
+
+        public MockControllerNode(NodeId id) {
+            this.id = id;
+        }
+
+        @Override
+        public NodeId id() {
+            return this.id;
+        }
+
+        @Override
+        public IpAddress ip() {
+            return null;
+        }
+
+        @Override
+        public int tcpPort() {
+            return 0;
+        }
+    }
+
+    private void testUpdateMetricWithoutId(ControlMetricType cmt, MetricValue mv) {
+        ControlMetric cm = new ControlMetric(cmt, mv);
+        monitor.updateMetric(cm, UPDATE_INTERVAL, Optional.ofNullable(null));
+    }
+
+    private void testLoadMetricWithoutId(ControlMetricType cmt, MetricValue mv) {
+        assertThat(monitor.getLoad(nodeId, cmt, Optional.ofNullable(null)).latest(), is(mv.getLoad()));
+    }
+
+    private void testUpdateMetricWithResource(ControlMetricType cmt, MetricValue mv, String resoureName) {
+        ControlMetric cm = new ControlMetric(cmt, mv);
+        monitor.updateMetric(cm, UPDATE_INTERVAL, resoureName);
+    }
+
+    private void testLoadMetricWithResource(ControlMetricType cmt, MetricValue mv, String resoureName) {
+        assertThat(monitor.getLoad(nodeId, cmt, resoureName).latest(), is(mv.getLoad()));
+    }
+
+    private void testUpdateMetricWithId(ControlMetricType cmt, MetricValue mv, DeviceId did) {
+        ControlMetric cm = new ControlMetric(cmt, mv);
+        monitor.updateMetric(cm, UPDATE_INTERVAL, Optional.of(did));
+    }
+
+    private void testLoadMetricWithId(ControlMetricType cmt, MetricValue mv, DeviceId did) {
+        assertThat(monitor.getLoad(nodeId, cmt, Optional.of(did)).latest(), is(mv.getLoad()));
+    }
+
+    @Test
+    public void testCpuMetric() {
+        MetricValue mv = new MetricValue.Builder().load(30).add();
+
+        CPU_METRICS.forEach(cmt -> testUpdateMetricWithoutId(cmt, mv));
+        CPU_METRICS.forEach(cmt -> testLoadMetricWithoutId(cmt, mv));
+    }
+
+    @Test
+    public void testMemoryMetric() {
+        MetricValue mv = new MetricValue.Builder().load(40).add();
+
+        MEMORY_METRICS.forEach(cmt -> testUpdateMetricWithoutId(cmt, mv));
+        MEMORY_METRICS.forEach(cmt -> testLoadMetricWithoutId(cmt, mv));
+    }
+
+    @Test
+    public void testDiskMetric() {
+        MetricValue mv = new MetricValue.Builder().load(50).add();
+
+        ImmutableSet<String> set = ImmutableSet.of("disk1", "disk2");
+
+        set.forEach(disk -> DISK_METRICS.forEach(cmt ->
+                testUpdateMetricWithResource(cmt, mv, disk)));
+
+        set.forEach(disk -> DISK_METRICS.forEach(cmt ->
+                testLoadMetricWithResource(cmt, mv, disk)));
+    }
+
+    @Test
+    public void testNetworkMetric() {
+        MetricValue mv = new MetricValue.Builder().load(10).add();
+
+        ImmutableSet<String> set = ImmutableSet.of("eth0", "eth1");
+
+        set.forEach(network -> NETWORK_METRICS.forEach(cmt ->
+                testUpdateMetricWithResource(cmt, mv, network)));
+
+        set.forEach(network -> NETWORK_METRICS.forEach(cmt ->
+                testLoadMetricWithResource(cmt, mv, network)));
+    }
+
+    @Test
+    public void testUpdateControlMessage() {
+        MetricValue mv = new MetricValue.Builder().load(10).add();
+
+        ImmutableSet<String> set = ImmutableSet.of("of:0000000000000001",
+                                                   "of:0000000000000002");
+
+        set.forEach(ctrlMsg -> CTRL_MSGS.forEach(cmt ->
+                testUpdateMetricWithId(cmt, mv, DeviceId.deviceId(ctrlMsg))));
+
+        set.forEach(ctrlMsg -> CTRL_MSGS.forEach(cmt ->
+                testLoadMetricWithId(cmt, mv, DeviceId.deviceId(ctrlMsg))));
+    }
+}

 {
-  "memoryUsedPercentage": 30,
-  "memoryFreePercentage": 70,
+  "memoryUsedRatio": 30,
+  "memoryFreeRatio": 70,
   "memoryUsed": 1024,
   "memoryFree": 2048
 }
\ No newline at end of file