홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Jian Li
Committed by Gerrit Code Review
67e1e15d 1 parent a98c2ea6

Enable to query the control metrics from remote node 

Change-Id: Ifef1c6eafd7cc79ed99be51f7faa26d97aeb2f67
Inline Side-by-side
Showing 13 changed files with 578 additions and 163 deletions
/*
* Copyright 2016-present 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;
import com.google.common.base.MoreObjects;
import java.util.Objects;
import static com.google.common.base.MoreObjects.toStringHelper;
/**
* A container class that is used to wrap the control metric response.
*/
public class ControlLoadSnapshot {
private final long latest;
private final long average;
private final long time;
/**
* Instantiates a new control metric response with given latest, average, time.
*
* @param latest latest value of control metric
* @param average average value of control metric
* @param time last logging time fo control metric
*/
public ControlLoadSnapshot(long latest, long average, long time) {
this.latest = latest;
this.average = average;
this.time = time;
}
/**
* Returns latest value of control metric.
*
* @return latest value of control metric
*/
public long latest() {
return latest;
}
/**
* Returns last logging time of control metric.
*
* @return last logging time of control metric
*/
public long time() {
return time;
}
/**
* Returns average value of control metric.
*
* @return average value of control metric
*/
public long average() {
return average;
}
@Override
public int hashCode() {
return Objects.hash(latest, average, time);
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj instanceof ControlLoadSnapshot) {
final ControlLoadSnapshot other = (ControlLoadSnapshot) obj;
return Objects.equals(this.latest, other.latest) &&
Objects.equals(this.average, other.average) &&
Objects.equals(this.time, other.time);
}
return false;
}
@Override
public String toString() {
MoreObjects.ToStringHelper helper;
helper = toStringHelper(this)
.add("latest", latest)
.add("average", average)
.add("time", time);
return helper.toString();
}
}
......@@ -13,14 +13,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.onosproject.cpman.impl;
package org.onosproject.cpman;
import com.google.common.base.MoreObjects;
import org.onosproject.cpman.ControlMetricType;
import org.onosproject.net.DeviceId;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import static com.google.common.base.MoreObjects.toStringHelper;
......@@ -31,10 +31,12 @@ public class ControlMetricsRequest {
private final ControlMetricType type;
private Optional<DeviceId> deviceId;
private String resourceName;
private int duration;
private TimeUnit unit;
/**
* Instantiates a new control metric request with the given control metric
* type and device identifier.
* Instantiates a new control metric request of the control metric type and
* device identifier.
*
* @param type control metric type
* @param deviceId device identifier
......@@ -45,8 +47,25 @@ public class ControlMetricsRequest {
}
/**
* Instantiates a new control metric request with the given control metric
* type and resource name.
* Instantiates a new control metric request of the control metric type and
* device identifier with the given projected time range.
*
* @param type control metric type
* @param duration projected time duration
* @param unit projected time unit
* @param deviceId device dientifer
*/
public ControlMetricsRequest(ControlMetricType type, int duration, TimeUnit unit,
Optional<DeviceId> deviceId) {
this.type = type;
this.deviceId = deviceId;
this.duration = duration;
this.unit = unit;
}
/**
* Instantiates a new control metric request of the control metric type and
* resource name.
*
* @param type control metric type
* @param resourceName resource name
......@@ -57,6 +76,23 @@ public class ControlMetricsRequest {
}
/**
* Instantiates a new control metric request of the control metric type and
* resource name with the given projected time range.
*
* @param type control metric type
* @param duration projected time duration
* @param unit projected time unit
* @param resourceName resource name
*/
public ControlMetricsRequest(ControlMetricType type, int duration, TimeUnit unit,
String resourceName) {
this.type = type;
this.resourceName = resourceName;
this.duration = duration;
this.unit = unit;
}
/**
* Obtains control metric type.
*
* @return control metric type
......@@ -83,9 +119,27 @@ public class ControlMetricsRequest {
return deviceId;
}
/**
* Obtains projected time duration.
*
* @return projected time duration
*/
public int getDuration() {
return duration;
}
/**
* Obtains projected time unit.
*
* @return projected time unit
*/
public TimeUnit getUnit() {
return unit;
}
@Override
public int hashCode() {
return Objects.hash(type, deviceId, resourceName);
return Objects.hash(type, deviceId, resourceName, duration, unit.toString());
}
@Override
......@@ -97,7 +151,9 @@ public class ControlMetricsRequest {
final ControlMetricsRequest other = (ControlMetricsRequest) obj;
return Objects.equals(this.type, other.type) &&
Objects.equals(this.deviceId, other.deviceId) &&
Objects.equals(this.resourceName, other.resourceName);
Objects.equals(this.resourceName, other.resourceName) &&
Objects.equals(this.duration, other.duration) &&
Objects.equals(this.unit, other.unit);
}
return false;
}
......@@ -107,7 +163,9 @@ public class ControlMetricsRequest {
MoreObjects.ToStringHelper helper;
helper = toStringHelper(this)
.add("type", type)
.add("resourceName", resourceName);
.add("resourceName", resourceName)
.add("duration", duration)
.add("timeUnit", unit);
if (deviceId != null) {
helper.add("deviceId", deviceId.get());
}
......
......@@ -21,8 +21,9 @@ import org.onosproject.net.DeviceId;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import static org.onosproject.cpman.ControlResource.*;
import static org.onosproject.cpman.ControlResource.Type;
/**
* Control Plane Statistics Service Interface.
......@@ -52,49 +53,64 @@ public interface ControlPlaneMonitorService {
String resourceName);
/**
* Obtains local control plane load of a specific device.
* Obtains snapshot of control plane load of a specific device.
* The metrics range from control messages and system metrics
* (e.g., CPU and memory info)
* (e.g., CPU and memory info).
* If the device id is not specified, it returns system metrics, otherwise,
* it returns control message stats of the given device.
*
* @param nodeId node identifier
* @param type control metric type
* @param deviceId device identifier
* @return control plane load
* @return completable future object of control load snapshot
*/
ControlLoad getLocalLoad(ControlMetricType type, Optional<DeviceId> deviceId);
CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
Optional<DeviceId> deviceId);
/**
* Obtains local control plane load of a specific resource.
* The metrics range from I/O device metrics
* (e.g., disk and network interface)
* Obtains snapshot of control plane load of a specific resource.
* The metrics include I/O device metrics (e.g., disk and network metrics).
*
* @param nodeId node identifier
* @param type control metric type
* @param resourceName resource name
* @return control plane load
* @return completable future object of control load snapshot
*/
ControlLoad getLocalLoad(ControlMetricType type, String resourceName);
CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
String resourceName);
/**
* Obtains remote control plane load of a specific device.
* Obtains snapshot of control plane load of a specific device with the
* projected range.
*
* @param nodeId node identifier
* @param type control metric type
* @param duration projected duration
* @param unit projected time unit
* @param deviceId device identifier
* @return completable future object of control load
* @return completable future object of control load snapshot
*/
CompletableFuture<ControlLoad> getRemoteLoad(NodeId nodeId,
CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
int duration, TimeUnit unit,
Optional<DeviceId> deviceId);
/**
* Obtains remote control plane load of a specific resource.
* Obtains snapshot of control plane load of a specific resource with the
* projected range.
*
* @param nodeId node identifier
* @param type control metric type
* @param duration projected duration
* @param unit projected time unit
* @param resourceName resource name
* @return completable future object of control load
* @return completable future object of control load snapshot
*/
CompletableFuture<ControlLoad> getRemoteLoad(NodeId nodeId,
CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
int duration, TimeUnit unit,
String resourceName);
/**
......
......@@ -17,16 +17,18 @@ package org.onosproject.cpman.cli;
import org.apache.karaf.shell.commands.Argument;
import org.apache.karaf.shell.commands.Command;
import org.onlab.util.Tools;
import org.onosproject.cli.AbstractShellCommand;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.NodeId;
import org.onosproject.cpman.ControlLoad;
import org.onosproject.cpman.ControlLoadSnapshot;
import org.onosproject.cpman.ControlMetricType;
import org.onosproject.cpman.ControlPlaneMonitorService;
import org.onosproject.net.DeviceId;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import static org.onosproject.cpman.ControlResource.CONTROL_MESSAGE_METRICS;
import static org.onosproject.cpman.ControlResource.CPU_METRICS;
......@@ -45,20 +47,25 @@ public class ControlMetricsStatsListCommand extends AbstractShellCommand {
"averageValue=%d, latestTime=%s";
private static final String INVALID_TYPE = "Invalid control resource type.";
@Argument(index = 0, name = "type",
private static final long TIMEOUT_MILLIS = 3000;
@Argument(index = 0, name = "node", description = "ONOS node identifier",
required = true, multiValued = false)
String node = null;
@Argument(index = 1, name = "type",
description = "Resource type (cpu|memory|disk|network|control_message)",
required = true, multiValued = false)
String type = null;
@Argument(index = 1, name = "name", description = "Resource name (or Device Id)",
@Argument(index = 2, name = "name", description = "Resource name (or Device Id)",
required = false, multiValued = false)
String name = null;
@Override
protected void execute() {
ControlPlaneMonitorService service = get(ControlPlaneMonitorService.class);
ClusterService clusterService = get(ClusterService.class);
NodeId nodeId = clusterService.getLocalNode().id();
NodeId nodeId = NodeId.nodeId(node);
switch (type) {
case "cpu":
printMetricsStats(service, nodeId, CPU_METRICS);
......@@ -74,7 +81,8 @@ public class ControlMetricsStatsListCommand extends AbstractShellCommand {
break;
case "control_message":
if (name != null) {
printMetricsStats(service, nodeId, CONTROL_MESSAGE_METRICS, DeviceId.deviceId(name));
printMetricsStats(service, nodeId, CONTROL_MESSAGE_METRICS,
DeviceId.deviceId(name));
}
break;
default:
......@@ -89,8 +97,8 @@ public class ControlMetricsStatsListCommand extends AbstractShellCommand {
}
private void printMetricsStats(ControlPlaneMonitorService service, NodeId nodeId,
Set<ControlMetricType> typeSet, String name) {
printMetricsStats(service, nodeId, typeSet, name, null);
Set<ControlMetricType> typeSet, String resName) {
printMetricsStats(service, nodeId, typeSet, resName, null);
}
private void printMetricsStats(ControlPlaneMonitorService service, NodeId nodeId,
......@@ -99,19 +107,39 @@ public class ControlMetricsStatsListCommand extends AbstractShellCommand {
}
private void printMetricsStats(ControlPlaneMonitorService service, NodeId nodeId,
Set<ControlMetricType> typeSet, String name, DeviceId did) {
if (name == null && did == null) {
typeSet.forEach(s -> print(s, service.getLocalLoad(s, Optional.ofNullable(null))));
} else if (name == null && did != null) {
typeSet.forEach(s -> print(s, service.getLocalLoad(s, Optional.of(did))));
} else if (name != null && did == null) {
typeSet.forEach(s -> print(s, service.getLocalLoad(s, name)));
Set<ControlMetricType> typeSet, String resName, DeviceId did) {
if (resName == null && did == null) {
typeSet.forEach(s -> {
CompletableFuture<ControlLoadSnapshot> cf =
service.getLoad(nodeId, s, Optional.empty());
ControlLoadSnapshot cmr =
Tools.futureGetOrElse(cf, TIMEOUT_MILLIS, TimeUnit.MILLISECONDS, null);
printRemote(s, cmr);
});
} else if (resName == null && did != null) {
typeSet.forEach(s -> {
CompletableFuture<ControlLoadSnapshot> cf =
service.getLoad(nodeId, s, Optional.of(did));
ControlLoadSnapshot cmr =
Tools.futureGetOrElse(cf, TIMEOUT_MILLIS, TimeUnit.MILLISECONDS, null);
printRemote(s, cmr);
});
} else if (resName != null && did == null) {
typeSet.forEach(s -> {
CompletableFuture<ControlLoadSnapshot> cf =
service.getLoad(nodeId, s, resName);
ControlLoadSnapshot cmr =
Tools.futureGetOrElse(cf, TIMEOUT_MILLIS, TimeUnit.MILLISECONDS, null);
printRemote(s, cmr);
});
}
}
private void print(ControlMetricType type, ControlLoad cl) {
if (cl != null) {
print(FMT, type.toString(), cl.latest(), cl.average(), cl.time());
private void printRemote(ControlMetricType cmType, ControlLoadSnapshot cls) {
if (cls != null) {
print(FMT, cmType.toString(), cls.latest(), cls.average(), cls.time());
} else {
print("Failed to retrieve metric value for type {}", cmType.toString());
}
}
}
......
......@@ -18,22 +18,22 @@ package org.onosproject.cpman.codec;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.onosproject.codec.CodecContext;
import org.onosproject.codec.JsonCodec;
import org.onosproject.cpman.ControlLoad;
import org.onosproject.cpman.ControlLoadSnapshot;
/**
* Control load codec.
* Control load snapshot codec.
*/
public final class ControlLoadCodec extends JsonCodec<ControlLoad> {
public final class ControlLoadSnapshotCodec extends JsonCodec<ControlLoadSnapshot> {
private static final String TIME = "time";
private static final String LATEST = "latest";
private static final String AVERAGE = "average";
@Override
public ObjectNode encode(ControlLoad controlLoad, CodecContext context) {
public ObjectNode encode(ControlLoadSnapshot controlLoadSnapshot, CodecContext context) {
return context.mapper().createObjectNode()
.put(TIME, controlLoad.time())
.put(LATEST, controlLoad.latest())
.put(AVERAGE, controlLoad.average());
.put(TIME, controlLoadSnapshot.time())
.put(LATEST, controlLoadSnapshot.latest())
.put(AVERAGE, controlLoadSnapshot.average());
}
}
......
......@@ -29,8 +29,10 @@ import org.onlab.util.KryoNamespace;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.NodeId;
import org.onosproject.cpman.ControlLoad;
import org.onosproject.cpman.ControlLoadSnapshot;
import org.onosproject.cpman.ControlMetric;
import org.onosproject.cpman.ControlMetricType;
import org.onosproject.cpman.ControlMetricsRequest;
import org.onosproject.cpman.ControlPlaneMonitorService;
import org.onosproject.cpman.MetricsDatabase;
import org.onosproject.net.DeviceId;
......@@ -45,12 +47,10 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static com.google.common.base.Preconditions.checkArgument;
import static org.onlab.util.Tools.groupedThreads;
import static org.onosproject.cpman.ControlResource.CONTROL_MESSAGE_METRICS;
import static org.onosproject.cpman.ControlResource.CPU_METRICS;
import static org.onosproject.cpman.ControlResource.DISK_METRICS;
......@@ -95,19 +95,16 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
private Map<Type, Set<String>> availableResourceMap;
private Set<DeviceId> availableDeviceIdSet;
private ExecutorService messageHandlingExecutor;
private static final String METRIC_TYPE_NULL = "Control metric type cannot be null";
Set<Map<ControlMetricType, Double>> debugSets = Sets.newHashSet();
private static final Serializer SERIALIZER = Serializer
.using(new KryoNamespace.Builder()
.register(KryoNamespaces.API)
.register(KryoNamespaces.BASIC)
.register(ControlMetricsRequest.class)
.register(DefaultControlLoad.class)
.register(DefaultMetricsDatabase.class)
.register(ControlLoadSnapshot.class)
.register(ControlMetricType.class)
.register(TimeUnit.class)
.nextId(KryoNamespaces.BEGIN_USER_CUSTOM_ID).build());
@Activate
......@@ -127,11 +124,8 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
availableResourceMap = Maps.newConcurrentMap();
availableDeviceIdSet = Sets.newConcurrentHashSet();
messageHandlingExecutor = Executors.newSingleThreadScheduledExecutor(
groupedThreads("onos/app/cpman", "message-handlers", log));
communicationService.addSubscriber(CONTROL_STATS,
SERIALIZER::decode, this::handleRequest, messageHandlingExecutor);
communicationService.<ControlMetricsRequest, ControlLoadSnapshot>addSubscriber(CONTROL_STATS,
SERIALIZER::decode, this::handleRequest, SERIALIZER::encode);
log.info("Started");
}
......@@ -243,57 +237,54 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
}
@Override
public ControlLoad getLocalLoad(ControlMetricType type,
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
Optional<DeviceId> deviceId) {
if (deviceId.isPresent()) {
if (CONTROL_MESSAGE_METRICS.contains(type) &&
availableDeviceIdSet.contains(deviceId.get())) {
return new DefaultControlLoad(controlMessageMap.get(deviceId.get()), type);
}
if (clusterService.getLocalNode().id().equals(nodeId)) {
return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, deviceId)));
} 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);
}
return communicationService.sendAndReceive(createRequest(type, deviceId),
CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
}
return null;
}
@Override
public ControlLoad getLocalLoad(ControlMetricType type, String resourceName) {
if (DISK_METRICS.contains(type) &&
availableResources(Type.DISK).contains(resourceName)) {
return new DefaultControlLoad(diskMetricsMap.get(resourceName), type);
}
if (NETWORK_METRICS.contains(type) &&
availableResources(Type.NETWORK).contains(resourceName)) {
return new DefaultControlLoad(networkMetricsMap.get(resourceName), type);
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
String resourceName) {
if (clusterService.getLocalNode().id().equals(nodeId)) {
return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, resourceName)));
} else {
return communicationService.sendAndReceive(createRequest(type, resourceName),
CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
}
return null;
}
@Override
public CompletableFuture<ControlLoad> getRemoteLoad(NodeId nodeId,
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
int duration, TimeUnit unit,
Optional<DeviceId> deviceId) {
return communicationService.sendAndReceive(createRequest(type, deviceId),
if (clusterService.getLocalNode().id().equals(nodeId)) {
return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, deviceId), duration, unit));
} else {
return communicationService.sendAndReceive(createRequest(type, duration, unit, deviceId),
CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
}
}
@Override
public CompletableFuture<ControlLoad> getRemoteLoad(NodeId nodeId,
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
int duration, TimeUnit unit,
String resourceName) {
return communicationService.sendAndReceive(createRequest(type, resourceName),
if (clusterService.getLocalNode().id().equals(nodeId)) {
return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, resourceName), duration, unit));
} else {
return communicationService.sendAndReceive(createRequest(type, duration, unit, resourceName),
CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
}
}
@Override
public Set<String> availableResources(Type resourceType) {
......@@ -309,6 +300,15 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
return ImmutableSet.of();
}
/**
* Builds and returns metric database instance with given resource name,
* resource type and metric type.
*
* @param resourceName resource name
* @param resourceType resource type
* @param metricTypes metric type
* @return metric database instance
*/
private MetricsDatabase genMDbBuilder(String resourceName,
Type resourceType,
Set<ControlMetricType> metricTypes) {
......@@ -319,56 +319,218 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
return builder.build();
}
/**
* Updates network metrics with given metric map and resource name.
*
* @param metricMap a metric map which is comprised of metric type and value
* @param resourceName resource name
*/
private void updateNetworkMetrics(Map<ControlMetricType, Double> metricMap,
String resName) {
networkMetricsMap.putIfAbsent(resName, genMDbBuilder(resName,
String resourceName) {
networkMetricsMap.putIfAbsent(resourceName, genMDbBuilder(resourceName,
Type.NETWORK, NETWORK_METRICS));
networkMetricsMap.get(resName).updateMetrics(convertMap(metricMap));
networkMetricsMap.get(resourceName).updateMetrics(convertMap(metricMap));
}
/**
* Updates disk metrics with given metric map and resource name.
*
* @param metricMap a metric map which is comprised of metric type and value
* @param resourceName resource name
*/
private void updateDiskMetrics(Map<ControlMetricType, Double> metricMap,
String resName) {
diskMetricsMap.putIfAbsent(resName, genMDbBuilder(resName,
String resourceName) {
diskMetricsMap.putIfAbsent(resourceName, genMDbBuilder(resourceName,
Type.DISK, DISK_METRICS));
diskMetricsMap.get(resName).updateMetrics(convertMap(metricMap));
diskMetricsMap.get(resourceName).updateMetrics(convertMap(metricMap));
}
/**
* Updates control message metrics with given metric map and device identifier.
*
* @param metricMap a metric map which is comprised of metric type and value
* @param deviceId device identifier
*/
private void updateControlMessages(Map<ControlMetricType, Double> metricMap,
DeviceId devId) {
controlMessageMap.putIfAbsent(devId, genMDbBuilder(devId.toString(),
DeviceId deviceId) {
controlMessageMap.putIfAbsent(deviceId, genMDbBuilder(deviceId.toString(),
Type.CONTROL_MESSAGE, CONTROL_MESSAGE_METRICS));
controlMessageMap.get(devId).updateMetrics(convertMap(metricMap));
controlMessageMap.get(deviceId).updateMetrics(convertMap(metricMap));
}
private Map convertMap(Map<ControlMetricType, Double> map) {
if (map == null) {
/**
* Converts metric map into a new map which contains string formatted metric type as key.
*
* @param metricMap metric map in which ControlMetricType is key
* @return a new map in which string formatted metric type is key
*/
private Map<String, Double> convertMap(Map<ControlMetricType, Double> metricMap) {
if (metricMap == null) {
return ImmutableMap.of();
}
Map newMap = Maps.newConcurrentMap();
map.forEach((k, v) -> newMap.putIfAbsent(k.toString(), v));
metricMap.forEach((k, v) -> newMap.putIfAbsent(k.toString(), v));
return newMap;
}
private CompletableFuture<ControlLoad> handleRequest(ControlMetricsRequest request) {
/**
* Handles control metric request from remote node.
*
* @param request control metric request
* @return completable future object of control load snapshot
*/
private CompletableFuture<ControlLoadSnapshot> handleRequest(ControlMetricsRequest request) {
checkArgument(request.getType() != null, METRIC_TYPE_NULL);
ControlLoad load;
if (request.getResourceName() != null) {
if (request.getResourceName() != null && request.getUnit() != null) {
load = getLocalLoad(request.getType(), request.getResourceName());
} else {
load = getLocalLoad(request.getType(), request.getDeviceId());
}
return CompletableFuture.completedFuture(load);
long average;
if (request.getUnit() != null) {
average = load.average(request.getDuration(), request.getUnit());
} else {
average = load.average();
}
ControlLoadSnapshot resp =
new ControlLoadSnapshot(load.latest(), average, load.time());
return CompletableFuture.completedFuture(resp);
}
/**
* Generates a control metric request.
*
* @param type control metric type
* @param deviceId device identifier
* @return control metric request instance
*/
private ControlMetricsRequest createRequest(ControlMetricType type,
Optional<DeviceId> deviceId) {
return new ControlMetricsRequest(type, deviceId);
}
/**
* Generates a control metric request with given projected time range.
*
* @param type control metric type
* @param duration projected time duration
* @param unit projected time unit
* @param deviceId device identifier
* @return control metric request instance
*/
private ControlMetricsRequest createRequest(ControlMetricType type,
int duration, TimeUnit unit,
Optional<DeviceId> deviceId) {
return new ControlMetricsRequest(type, duration, unit, deviceId);
}
/**
* Generates a control metric request.
*
* @param type control metric type
* @param resourceName resource name
* @return control metric request instance
*/
private ControlMetricsRequest createRequest(ControlMetricType type,
String resourceName) {
return new ControlMetricsRequest(type, resourceName);
}
/**
* Generates a control metric request with given projected time range.
*
* @param type control metric type
* @param duration projected time duration
* @param unit projected time unit
* @param resourceName resource name
* @return control metric request instance
*/
private ControlMetricsRequest createRequest(ControlMetricType type,
int duration, TimeUnit unit,
String resourceName) {
return new ControlMetricsRequest(type, duration, unit, resourceName);
}
/**
* Returns a snapshot of control load.
*
* @param cl control load
* @return a snapshot of control load
*/
private ControlLoadSnapshot snapshot(ControlLoad cl) {
if (cl != null) {
return new ControlLoadSnapshot(cl.latest(), cl.average(), cl.time());
}
return null;
}
/**
* Returns a snapshot of control load with given projected time range.
*
* @param cl control load
* @param duration projected time duration
* @param unit projected time unit
* @return a snapshot of control load
*/
private ControlLoadSnapshot snapshot(ControlLoad cl, int duration, TimeUnit unit) {
if (cl != null) {
return new ControlLoadSnapshot(cl.latest(), cl.average(duration, unit), cl.time());
}
return null;
}
/**
* Returns local control load.
*
* @param type metric type
* @param deviceId device identifier
* @return control load
*/
private ControlLoad getLocalLoad(ControlMetricType type,
Optional<DeviceId> deviceId) {
if (deviceId.isPresent()) {
// returns control message stats
if (CONTROL_MESSAGE_METRICS.contains(type) &&
availableDeviceIdSet.contains(deviceId.get())) {
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);
}
}
return null;
}
/**
* Returns local control load.
*
* @param type metric type
* @param resourceName resource name
* @return control load
*/
private ControlLoad getLocalLoad(ControlMetricType type, String resourceName) {
// returns disk I/O stats
if (DISK_METRICS.contains(type) &&
availableResources(Type.DISK).contains(resourceName)) {
return new DefaultControlLoad(diskMetricsMap.get(resourceName), type);
}
// returns network I/O stats
if (NETWORK_METRICS.contains(type) &&
availableResources(Type.NETWORK).contains(resourceName)) {
return new DefaultControlLoad(networkMetricsMap.get(resourceName), type);
}
return null;
}
}
......
......@@ -21,8 +21,8 @@ import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.onosproject.codec.CodecService;
import org.onosproject.cpman.ControlLoad;
import org.onosproject.cpman.codec.ControlLoadCodec;
import org.onosproject.cpman.ControlLoadSnapshot;
import org.onosproject.cpman.codec.ControlLoadSnapshotCodec;
import org.slf4j.Logger;
import static org.slf4j.LoggerFactory.getLogger;
......@@ -40,14 +40,14 @@ public class CPManCodecRegistrator {
@Activate
public void activate() {
codecService.registerCodec(ControlLoad.class, new ControlLoadCodec());
codecService.registerCodec(ControlLoadSnapshot.class, new ControlLoadSnapshotCodec());
log.info("Started");
}
@Deactivate
public void deactivate() {
codecService.unregisterCodec(ControlLoad.class);
codecService.unregisterCodec(ControlLoadSnapshot.class);
log.info("Stopped");
}
......
......@@ -18,9 +18,10 @@ package org.onosproject.cpman.rest;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.commons.lang3.StringUtils;
import org.onlab.util.Tools;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.NodeId;
import org.onosproject.cpman.ControlLoad;
import org.onosproject.cpman.ControlLoadSnapshot;
import org.onosproject.cpman.ControlMetricType;
import org.onosproject.cpman.ControlPlaneMonitorService;
import org.onosproject.net.DeviceId;
......@@ -34,6 +35,8 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import static org.onosproject.cpman.ControlResource.CONTROL_MESSAGE_METRICS;
import static org.onosproject.cpman.ControlResource.CPU_METRICS;
......@@ -55,6 +58,7 @@ public class ControlMetricsWebResource extends AbstractWebResource {
private final ClusterService clusterService = get(ClusterService.class);
private final NodeId localNodeId = clusterService.getLocalNode().id();
private final ObjectNode root = mapper().createObjectNode();
private static final long TIMEOUT_MILLIS = 1000;
/**
* Returns control message metrics of all devices.
......@@ -248,32 +252,52 @@ public class ControlMetricsWebResource extends AbstractWebResource {
if (name == null && did == null) {
typeSet.forEach(type -> {
ObjectNode metricNode = mapper().createObjectNode();
ControlLoad load = service.getLocalLoad(type, Optional.ofNullable(null));
if (load != null) {
metricNode.set(toCamelCase(type.toString(), true), codec(ControlLoad.class)
.encode(service.getLocalLoad(type, Optional.ofNullable(null)), this));
CompletableFuture<ControlLoadSnapshot> cf =
service.getLoad(nodeId, type, Optional.empty());
if (cf != null) {
ControlLoadSnapshot cmr =
Tools.futureGetOrElse(cf, TIMEOUT_MILLIS, TimeUnit.MILLISECONDS, null);
if (cmr != null) {
metricNode.set(toCamelCase(type.toString(), true),
codec(ControlLoadSnapshot.class).encode(cmr, this));
metricsNode.add(metricNode);
}
}
});
} else if (name == null) {
typeSet.forEach(type -> {
ObjectNode metricNode = mapper().createObjectNode();
ControlLoad load = service.getLocalLoad(type, Optional.of(did));
if (load != null) {
CompletableFuture<ControlLoadSnapshot> cf =
service.getLoad(nodeId, type, Optional.of(did));
if (cf != null) {
ControlLoadSnapshot cmr =
Tools.futureGetOrElse(cf, TIMEOUT_MILLIS, TimeUnit.MILLISECONDS, null);
if (cmr != null) {
metricNode.set(toCamelCase(type.toString(), true),
codec(ControlLoad.class).encode(load, this));
codec(ControlLoadSnapshot.class).encode(cmr, this));
metricsNode.add(metricNode);
}
}
});
} else if (did == null) {
typeSet.forEach(type -> {
ObjectNode metricNode = mapper().createObjectNode();
ControlLoad load = service.getLocalLoad(type, name);
if (load != null) {
CompletableFuture<ControlLoadSnapshot> cf =
service.getLoad(nodeId, type, name);
if (cf != null) {
ControlLoadSnapshot cmr =
Tools.futureGetOrElse(cf, TIMEOUT_MILLIS, TimeUnit.MILLISECONDS, null);
if (cmr != null) {
metricNode.set(toCamelCase(type.toString(), true),
codec(ControlLoad.class).encode(load, this));
codec(ControlLoadSnapshot.class).encode(cmr, this));
metricsNode.add(metricNode);
}
}
});
}
......
......@@ -106,31 +106,31 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
long cpuIdleTime = nullIsIllegal(jsonTree.get("cpuIdleTime").asLong(), INVALID_REQUEST);
aggregator.setMetricsService(metricsService);
aggregator.addMetrics(Optional.ofNullable(null), SYSTEM_TYPE);
aggregator.addMetrics(Optional.empty(), SYSTEM_TYPE);
cm = new ControlMetric(ControlMetricType.CPU_LOAD,
new MetricValue.Builder().load(cpuLoad).add());
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.ofNullable(null));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
aggregator.increment(ControlMetricType.CPU_IDLE_TIME, cpuIdleTime);
} catch (IOException e) {
......@@ -167,26 +167,26 @@ public class SystemMetricsCollectorWebResource extends AbstractWebResource {
long memFreeRatio = memTotal == 0L ? 0L : (memFree * PERCENT_CONSTANT) / memTotal;
aggregator.setMetricsService(metricsService);
aggregator.addMetrics(Optional.ofNullable(null), SYSTEM_TYPE);
aggregator.addMetrics(Optional.empty(), 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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
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));
monitorService.updateMetric(cm, UPDATE_INTERVAL_IN_MINUTE, Optional.empty());
aggregator.increment(ControlMetricType.MEMORY_FREE, memFree);
} catch (IOException e) {
......
......@@ -19,12 +19,14 @@
<command>
<action class="org.onosproject.cpman.cli.ControlMetricsStatsListCommand"/>
<completers>
<ref component-id="nodeIdCompleter"/>
<ref component-id="controlResourceTypeCompleter"/>
<ref component-id="resourceNameCompleter"/>
</completers>
</command>
</command-bundle>
<bean id="nodeIdCompleter" class="org.onosproject.cli.NodeIdCompleter"/>
<bean id="controlResourceTypeCompleter" class="org.onosproject.cpman.cli.ControlResourceTypeCompleter"/>
<bean id="resourceNameCompleter" class="org.onosproject.cpman.cli.ResourceNameCompleter"/>
</blueprint>
\ No newline at end of file
......
......@@ -31,6 +31,7 @@ import org.onosproject.store.cluster.messaging.ClusterCommunicationServiceAdapte
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.createMock;
......@@ -109,11 +110,17 @@ public class ControlPlaneMonitorTest {
private void testUpdateMetricWithoutId(ControlMetricType cmt, MetricValue mv) {
ControlMetric cm = new ControlMetric(cmt, mv);
monitor.updateMetric(cm, UPDATE_INTERVAL, Optional.ofNullable(null));
monitor.updateMetric(cm, UPDATE_INTERVAL, Optional.empty());
}
private void testLoadMetricWithoutId(ControlMetricType cmt, MetricValue mv) {
assertThat(monitor.getLocalLoad(cmt, Optional.ofNullable(null)).latest(), is(mv.getLoad()));
private void testLoadMetric(NodeId nodeId, ControlMetricType cmt, MetricValue mv) {
try {
assertThat(monitor.getLoad(nodeId, cmt, Optional.empty()).get().latest(), is(mv.getLoad()));
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
private void testUpdateMetricWithResource(ControlMetricType cmt, MetricValue mv, String resourceName) {
......@@ -121,8 +128,14 @@ public class ControlPlaneMonitorTest {
monitor.updateMetric(cm, UPDATE_INTERVAL, resourceName);
}
private void testLoadMetricWithResource(ControlMetricType cmt, MetricValue mv, String resourceName) {
assertThat(monitor.getLocalLoad(cmt, resourceName).latest(), is(mv.getLoad()));
private void testLoadMetricWithResource(NodeId nodeId, ControlMetricType cmt, MetricValue mv, String resourceName) {
try {
assertThat(monitor.getLoad(nodeId, cmt, resourceName).get().latest(), is(mv.getLoad()));
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
private void testUpdateMetricWithId(ControlMetricType cmt, MetricValue mv, DeviceId did) {
......@@ -130,8 +143,14 @@ public class ControlPlaneMonitorTest {
monitor.updateMetric(cm, UPDATE_INTERVAL, Optional.of(did));
}
private void testLoadMetricWithId(ControlMetricType cmt, MetricValue mv, DeviceId did) {
assertThat(monitor.getLocalLoad(cmt, Optional.of(did)).latest(), is(mv.getLoad()));
private void testLoadMetricWithId(NodeId nodeId, ControlMetricType cmt, MetricValue mv, DeviceId did) {
try {
assertThat(monitor.getLoad(nodeId, cmt, Optional.of(did)).get().latest(), is(mv.getLoad()));
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
/**
......@@ -142,7 +161,7 @@ public class ControlPlaneMonitorTest {
MetricValue mv = new MetricValue.Builder().load(30).add();
CPU_METRICS.forEach(cmt -> testUpdateMetricWithoutId(cmt, mv));
CPU_METRICS.forEach(cmt -> testLoadMetricWithoutId(cmt, mv));
CPU_METRICS.forEach(cmt -> testLoadMetric(nodeId, cmt, mv));
}
/**
......@@ -153,7 +172,7 @@ public class ControlPlaneMonitorTest {
MetricValue mv = new MetricValue.Builder().load(40).add();
MEMORY_METRICS.forEach(cmt -> testUpdateMetricWithoutId(cmt, mv));
MEMORY_METRICS.forEach(cmt -> testLoadMetricWithoutId(cmt, mv));
MEMORY_METRICS.forEach(cmt -> testLoadMetric(nodeId, cmt, mv));
}
/**
......@@ -169,7 +188,7 @@ public class ControlPlaneMonitorTest {
testUpdateMetricWithResource(cmt, mv, disk)));
set.forEach(disk -> DISK_METRICS.forEach(cmt ->
testLoadMetricWithResource(cmt, mv, disk)));
testLoadMetricWithResource(nodeId, cmt, mv, disk)));
}
/**
......@@ -185,7 +204,7 @@ public class ControlPlaneMonitorTest {
testUpdateMetricWithResource(cmt, mv, network)));
set.forEach(network -> NETWORK_METRICS.forEach(cmt ->
testLoadMetricWithResource(cmt, mv, network)));
testLoadMetricWithResource(nodeId, cmt, mv, network)));
}
/**
......@@ -201,7 +220,7 @@ public class ControlPlaneMonitorTest {
testUpdateMetricWithId(cmt, mv, devId)));
set.forEach(devId -> CONTROL_MESSAGE_METRICS.forEach(cmt ->
testLoadMetricWithId(cmt, mv, devId)));
testLoadMetricWithId(nodeId, cmt, mv, devId)));
}
/**
......
......@@ -16,7 +16,7 @@
package org.onosproject.cpman.impl.message;
import org.onosproject.cluster.NodeId;
import org.onosproject.cpman.ControlLoad;
import org.onosproject.cpman.ControlLoadSnapshot;
import org.onosproject.cpman.ControlMetric;
import org.onosproject.cpman.ControlMetricType;
import org.onosproject.cpman.ControlPlaneMonitorService;
......@@ -26,6 +26,7 @@ import org.onosproject.net.DeviceId;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
/**
* Test adapter control plane monitoring service.
......@@ -44,26 +45,31 @@ public class ControlPlaneMonitorServiceAdaptor implements ControlPlaneMonitorSer
}
@Override
public ControlLoad getLocalLoad(ControlMetricType type,
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
Optional<DeviceId> deviceId) {
return null;
}
@Override
public ControlLoad getLocalLoad(ControlMetricType type, String resourceName) {
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
String resourceName) {
return null;
}
@Override
public CompletableFuture<ControlLoad> getRemoteLoad(NodeId nodeId,
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
int duration, TimeUnit unit,
Optional<DeviceId> deviceId) {
return null;
}
@Override
public CompletableFuture<ControlLoad> getRemoteLoad(NodeId nodeId,
public CompletableFuture<ControlLoadSnapshot> getLoad(NodeId nodeId,
ControlMetricType type,
int duration, TimeUnit unit,
String resourceName) {
return null;
}
......
......@@ -15,11 +15,7 @@
*/
package org.onosproject.cpman.rest;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.client.WebTarget;
import com.google.common.collect.ImmutableSet;
import org.glassfish.jersey.server.ResourceConfig;
import org.junit.Before;
import org.junit.Test;
......@@ -33,11 +29,14 @@ import org.onosproject.cluster.NodeId;
import org.onosproject.codec.CodecService;
import org.onosproject.codec.impl.CodecManager;
import org.onosproject.cpman.ControlLoad;
import org.onosproject.cpman.ControlLoadSnapshot;
import org.onosproject.cpman.ControlPlaneMonitorService;
import org.onosproject.cpman.codec.ControlLoadCodec;
import org.onosproject.cpman.codec.ControlLoadSnapshotCodec;
import org.onosproject.rest.resources.ResourceTest;
import com.google.common.collect.ImmutableSet;
import javax.ws.rs.client.WebTarget;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.anyString;
......@@ -145,7 +144,7 @@ public class ControlMetricsResourceTest extends ResourceTest {
public void setUpTest() {
final CodecManager codecService = new CodecManager();
codecService.activate();
codecService.registerCodec(ControlLoad.class, new ControlLoadCodec());
codecService.registerCodec(ControlLoadSnapshot.class, new ControlLoadSnapshotCodec());
ServiceDirectory testDirectory =
new TestServiceDirectory()
.add(ControlPlaneMonitorService.class,
......@@ -184,7 +183,7 @@ public class ControlMetricsResourceTest extends ResourceTest {
public void testResourcePopulatedArray() {
expect(mockControlPlaneMonitorService.availableResources(anyObject()))
.andReturn(resourceSet).once();
expect(mockControlPlaneMonitorService.getLocalLoad(anyObject(),
expect(mockControlPlaneMonitorService.getLoad(anyObject(), anyObject(),
anyString())).andReturn(null).times(4);
replay(mockControlPlaneMonitorService);
......