Jian Li
Committed by Gerrit Code Review

Support to query available resources from remote node

Change-Id: I465327143b5959b9e18daac9481ffea332f889c8
...@@ -15,6 +15,8 @@ ...@@ -15,6 +15,8 @@
15 */ 15 */
16 package org.onosproject.cpman; 16 package org.onosproject.cpman;
17 17
18 +import com.google.common.collect.ImmutableSet;
19 +import org.onlab.util.Tools;
18 import org.onosproject.cluster.NodeId; 20 import org.onosproject.cluster.NodeId;
19 import org.onosproject.net.DeviceId; 21 import org.onosproject.net.DeviceId;
20 22
...@@ -30,6 +32,8 @@ import static org.onosproject.cpman.ControlResource.Type; ...@@ -30,6 +32,8 @@ import static org.onosproject.cpman.ControlResource.Type;
30 */ 32 */
31 public interface ControlPlaneMonitorService { 33 public interface ControlPlaneMonitorService {
32 34
35 + long TIMEOUT_MILLIS = 2000;
36 +
33 /** 37 /**
34 * Adds a new control metric value with a certain update interval. 38 * Adds a new control metric value with a certain update interval.
35 * 39 *
...@@ -116,8 +120,22 @@ public interface ControlPlaneMonitorService { ...@@ -116,8 +120,22 @@ public interface ControlPlaneMonitorService {
116 /** 120 /**
117 * Obtains a list of names of available resources. 121 * Obtains a list of names of available resources.
118 * 122 *
123 + * @param nodeId node identifier
124 + * @param resourceType resource type
125 + * @return completable future object of a collection of available resource names
126 + */
127 + CompletableFuture<Set<String>> availableResources(NodeId nodeId, Type resourceType);
128 +
129 + /**
130 + * Synchronous version of availableResource.
131 + * Obtains a list of names of available resources.
132 + *
133 + * @param nodeId node identifier
119 * @param resourceType resource type 134 * @param resourceType resource type
120 - * @return a collection of names of available resources 135 + * @return a collection of available resource names
121 */ 136 */
122 - Set<String> availableResources(Type resourceType); 137 + default Set<String> availableResourcesSync(NodeId nodeId, Type resourceType) {
138 + return Tools.futureGetOrElse(availableResources(nodeId, resourceType),
139 + TIMEOUT_MILLIS, TimeUnit.MILLISECONDS, ImmutableSet.of());
140 + }
123 } 141 }
...\ No newline at end of file ...\ No newline at end of file
......
1 +/*
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.cpman;
17 +
18 +import com.google.common.base.MoreObjects;
19 +
20 +import java.util.Objects;
21 +
22 +import static com.google.common.base.MoreObjects.toStringHelper;
23 +
24 +/**
25 + * A container class that is used to request available resource of remote node.
26 + */
27 +public class ControlResourceRequest {
28 + private final ControlResource.Type type;
29 +
30 + /**
31 + * Instantiates a new control resource request of the control resource type.
32 + *
33 + * @param type control resource type
34 + */
35 + public ControlResourceRequest(ControlResource.Type type) {
36 + this.type = type;
37 + }
38 +
39 + /**
40 + * Obtains control resource type.
41 + *
42 + * @return control resource type
43 + */
44 + public ControlResource.Type getType() {
45 + return type;
46 + }
47 +
48 + @Override
49 + public int hashCode() {
50 + return Objects.hash(type);
51 + }
52 +
53 + @Override
54 + public boolean equals(Object obj) {
55 + if (this == obj) {
56 + return true;
57 + }
58 + if (obj instanceof ControlResourceRequest) {
59 + final ControlResourceRequest other = (ControlResourceRequest) obj;
60 + return Objects.equals(this.type, other.type);
61 + }
62 + return false;
63 + }
64 +
65 + @Override
66 + public String toString() {
67 + MoreObjects.ToStringHelper helper;
68 + helper = toStringHelper(this)
69 + .add("type", type);
70 + return helper.toString();
71 + }
72 +}
...@@ -21,6 +21,7 @@ import org.apache.karaf.shell.console.completer.ArgumentCompleter; ...@@ -21,6 +21,7 @@ import org.apache.karaf.shell.console.completer.ArgumentCompleter;
21 import org.apache.karaf.shell.console.completer.StringsCompleter; 21 import org.apache.karaf.shell.console.completer.StringsCompleter;
22 import org.onosproject.cli.AbstractCompleter; 22 import org.onosproject.cli.AbstractCompleter;
23 import org.onosproject.cli.AbstractShellCommand; 23 import org.onosproject.cli.AbstractShellCommand;
24 +import org.onosproject.cluster.NodeId;
24 import org.onosproject.cpman.ControlPlaneMonitorService; 25 import org.onosproject.cpman.ControlPlaneMonitorService;
25 import org.onosproject.cpman.ControlResource; 26 import org.onosproject.cpman.ControlResource;
26 import org.slf4j.Logger; 27 import org.slf4j.Logger;
...@@ -40,7 +41,7 @@ public class ResourceNameCompleter extends AbstractCompleter { ...@@ -40,7 +41,7 @@ public class ResourceNameCompleter extends AbstractCompleter {
40 private static final String NETWORK = "network"; 41 private static final String NETWORK = "network";
41 private static final String DISK = "disk"; 42 private static final String DISK = "disk";
42 private static final String CONTROL_MESSAGE = "control_message"; 43 private static final String CONTROL_MESSAGE = "control_message";
43 - Set<String> resourceTypes = ImmutableSet.of(NETWORK, DISK, CONTROL_MESSAGE); 44 + private final Set<String> resourceTypes = ImmutableSet.of(NETWORK, DISK, CONTROL_MESSAGE);
44 private static final String INVALID_MSG = "Invalid type name"; 45 private static final String INVALID_MSG = "Invalid type name";
45 46
46 47
...@@ -51,7 +52,8 @@ public class ResourceNameCompleter extends AbstractCompleter { ...@@ -51,7 +52,8 @@ public class ResourceNameCompleter extends AbstractCompleter {
51 52
52 // Resource type is the second argument. 53 // Resource type is the second argument.
53 ArgumentCompleter.ArgumentList list = getArgumentList(); 54 ArgumentCompleter.ArgumentList list = getArgumentList();
54 - String type = list.getArguments()[1]; 55 + String nodeId = list.getArguments()[1];
56 + String type = list.getArguments()[2];
55 57
56 if (resourceTypes.contains(type)) { 58 if (resourceTypes.contains(type)) {
57 ControlPlaneMonitorService monitorService = 59 ControlPlaneMonitorService monitorService =
...@@ -60,13 +62,16 @@ public class ResourceNameCompleter extends AbstractCompleter { ...@@ -60,13 +62,16 @@ public class ResourceNameCompleter extends AbstractCompleter {
60 Set<String> set = Sets.newHashSet(); 62 Set<String> set = Sets.newHashSet();
61 switch (type) { 63 switch (type) {
62 case NETWORK: 64 case NETWORK:
63 - set = monitorService.availableResources(ControlResource.Type.NETWORK); 65 + set = monitorService.availableResourcesSync(NodeId.nodeId(nodeId),
66 + ControlResource.Type.NETWORK);
64 break; 67 break;
65 case DISK: 68 case DISK:
66 - set = monitorService.availableResources(ControlResource.Type.DISK); 69 + set = monitorService.availableResourcesSync(NodeId.nodeId(nodeId),
70 + ControlResource.Type.DISK);
67 break; 71 break;
68 case CONTROL_MESSAGE: 72 case CONTROL_MESSAGE:
69 - set = monitorService.availableResources(ControlResource.Type.CONTROL_MESSAGE); 73 + set = monitorService.availableResourcesSync(NodeId.nodeId(nodeId),
74 + ControlResource.Type.CONTROL_MESSAGE);
70 break; 75 break;
71 default: 76 default:
72 log.warn(INVALID_MSG); 77 log.warn(INVALID_MSG);
...@@ -76,7 +81,7 @@ public class ResourceNameCompleter extends AbstractCompleter { ...@@ -76,7 +81,7 @@ public class ResourceNameCompleter extends AbstractCompleter {
76 SortedSet<String> strings = delegate.getStrings(); 81 SortedSet<String> strings = delegate.getStrings();
77 82
78 if (set.size() != 0) { 83 if (set.size() != 0) {
79 - set.forEach(s -> strings.add(s)); 84 + set.forEach(strings::add);
80 } 85 }
81 } 86 }
82 87
......
...@@ -25,6 +25,7 @@ import org.apache.commons.lang.ArrayUtils; ...@@ -25,6 +25,7 @@ import org.apache.commons.lang.ArrayUtils;
25 import org.apache.commons.lang3.StringUtils; 25 import org.apache.commons.lang3.StringUtils;
26 import org.joda.time.LocalDateTime; 26 import org.joda.time.LocalDateTime;
27 import org.onosproject.cluster.ClusterService; 27 import org.onosproject.cluster.ClusterService;
28 +import org.onosproject.cluster.NodeId;
28 import org.onosproject.cpman.ControlLoadSnapshot; 29 import org.onosproject.cpman.ControlLoadSnapshot;
29 import org.onosproject.cpman.ControlMetricType; 30 import org.onosproject.cpman.ControlMetricType;
30 import org.onosproject.cpman.ControlPlaneMonitorService; 31 import org.onosproject.cpman.ControlPlaneMonitorService;
...@@ -96,10 +97,11 @@ public class CpmanViewMessageHandler extends UiMessageHandler { ...@@ -96,10 +97,11 @@ public class CpmanViewMessageHandler extends UiMessageHandler {
96 ControlPlaneMonitorService cpms = get(ControlPlaneMonitorService.class); 97 ControlPlaneMonitorService cpms = get(ControlPlaneMonitorService.class);
97 ClusterService cs = get(ClusterService.class); 98 ClusterService cs = get(ClusterService.class);
98 DeviceService ds = get(DeviceService.class); 99 DeviceService ds = get(DeviceService.class);
100 + NodeId localNodeId = cs.getLocalNode().id();
99 101
100 if (!Strings.isNullOrEmpty(uri)) { 102 if (!Strings.isNullOrEmpty(uri)) {
101 DeviceId deviceId = DeviceId.deviceId(uri); 103 DeviceId deviceId = DeviceId.deviceId(uri);
102 - if (cpms.availableResources(CONTROL_MESSAGE).contains(deviceId.toString())) { 104 + if (cpms.availableResourcesSync(localNodeId, CONTROL_MESSAGE).contains(deviceId.toString())) {
103 Map<ControlMetricType, Long[]> data = generateMatrix(cpms, cs, deviceId); 105 Map<ControlMetricType, Long[]> data = generateMatrix(cpms, cs, deviceId);
104 LocalDateTime ldt = new LocalDateTime(timestamp * MILLI_CONV_UNIT); 106 LocalDateTime ldt = new LocalDateTime(timestamp * MILLI_CONV_UNIT);
105 107
...@@ -110,7 +112,7 @@ public class CpmanViewMessageHandler extends UiMessageHandler { ...@@ -110,7 +112,7 @@ public class CpmanViewMessageHandler extends UiMessageHandler {
110 attachDeviceList(cm, deviceIds); 112 attachDeviceList(cm, deviceIds);
111 } 113 }
112 } else { 114 } else {
113 - Set<String> deviceIds = cpms.availableResources(CONTROL_MESSAGE); 115 + Set<String> deviceIds = cpms.availableResourcesSync(localNodeId, CONTROL_MESSAGE);
114 for (String deviceId : deviceIds) { 116 for (String deviceId : deviceIds) {
115 Map<ControlMetricType, Long> data = 117 Map<ControlMetricType, Long> data =
116 populateDeviceMetrics(cpms, cs, DeviceId.deviceId(deviceId)); 118 populateDeviceMetrics(cpms, cs, DeviceId.deviceId(deviceId));
......
...@@ -34,6 +34,8 @@ import org.onosproject.cpman.ControlMetric; ...@@ -34,6 +34,8 @@ import org.onosproject.cpman.ControlMetric;
34 import org.onosproject.cpman.ControlMetricType; 34 import org.onosproject.cpman.ControlMetricType;
35 import org.onosproject.cpman.ControlMetricsRequest; 35 import org.onosproject.cpman.ControlMetricsRequest;
36 import org.onosproject.cpman.ControlPlaneMonitorService; 36 import org.onosproject.cpman.ControlPlaneMonitorService;
37 +import org.onosproject.cpman.ControlResource;
38 +import org.onosproject.cpman.ControlResourceRequest;
37 import org.onosproject.cpman.MetricsDatabase; 39 import org.onosproject.cpman.MetricsDatabase;
38 import org.onosproject.net.DeviceId; 40 import org.onosproject.net.DeviceId;
39 import org.onosproject.store.cluster.messaging.ClusterCommunicationService; 41 import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
...@@ -86,6 +88,9 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -86,6 +88,9 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
86 private static final MessageSubject CONTROL_STATS = 88 private static final MessageSubject CONTROL_STATS =
87 new MessageSubject("control-plane-stats"); 89 new MessageSubject("control-plane-stats");
88 90
91 + private static final MessageSubject CONTROL_RESOURCE =
92 + new MessageSubject("control-plane-resources");
93 +
89 private Map<ControlMetricType, Double> cpuBuf; 94 private Map<ControlMetricType, Double> cpuBuf;
90 private Map<ControlMetricType, Double> memoryBuf; 95 private Map<ControlMetricType, Double> memoryBuf;
91 private Map<String, Map<ControlMetricType, Double>> diskBuf; 96 private Map<String, Map<ControlMetricType, Double>> diskBuf;
...@@ -96,14 +101,17 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -96,14 +101,17 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
96 private Set<DeviceId> availableDeviceIdSet; 101 private Set<DeviceId> availableDeviceIdSet;
97 102
98 private static final String METRIC_TYPE_NULL = "Control metric type cannot be null"; 103 private static final String METRIC_TYPE_NULL = "Control metric type cannot be null";
104 + private static final String RESOURCE_TYPE_NULL = "Control resource type cannot be null";
99 105
100 private static final Serializer SERIALIZER = Serializer 106 private static final Serializer SERIALIZER = Serializer
101 .using(new KryoNamespace.Builder() 107 .using(new KryoNamespace.Builder()
102 .register(KryoNamespaces.API) 108 .register(KryoNamespaces.API)
103 .register(KryoNamespaces.BASIC) 109 .register(KryoNamespaces.BASIC)
104 .register(ControlMetricsRequest.class) 110 .register(ControlMetricsRequest.class)
111 + .register(ControlResourceRequest.class)
105 .register(ControlLoadSnapshot.class) 112 .register(ControlLoadSnapshot.class)
106 .register(ControlMetricType.class) 113 .register(ControlMetricType.class)
114 + .register(ControlResource.Type.class)
107 .register(TimeUnit.class) 115 .register(TimeUnit.class)
108 .nextId(KryoNamespaces.BEGIN_USER_CUSTOM_ID).build()); 116 .nextId(KryoNamespaces.BEGIN_USER_CUSTOM_ID).build());
109 117
...@@ -125,7 +133,10 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -125,7 +133,10 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
125 availableDeviceIdSet = Sets.newConcurrentHashSet(); 133 availableDeviceIdSet = Sets.newConcurrentHashSet();
126 134
127 communicationService.<ControlMetricsRequest, ControlLoadSnapshot>addSubscriber(CONTROL_STATS, 135 communicationService.<ControlMetricsRequest, ControlLoadSnapshot>addSubscriber(CONTROL_STATS,
128 - SERIALIZER::decode, this::handleRequest, SERIALIZER::encode); 136 + SERIALIZER::decode, this::handleMetricsRequest, SERIALIZER::encode);
137 +
138 + communicationService.<ControlResourceRequest, Set<String>>addSubscriber(CONTROL_RESOURCE,
139 + SERIALIZER::decode, this::handleResourceRequest, SERIALIZER::encode);
129 140
130 log.info("Started"); 141 log.info("Started");
131 } 142 }
...@@ -141,6 +152,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -141,6 +152,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
141 ctrlMsgBuf.clear(); 152 ctrlMsgBuf.clear();
142 153
143 communicationService.removeSubscriber(CONTROL_STATS); 154 communicationService.removeSubscriber(CONTROL_STATS);
155 + communicationService.removeSubscriber(CONTROL_RESOURCE);
144 156
145 log.info("Stopped"); 157 log.info("Stopped");
146 } 158 }
...@@ -243,7 +255,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -243,7 +255,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
243 if (clusterService.getLocalNode().id().equals(nodeId)) { 255 if (clusterService.getLocalNode().id().equals(nodeId)) {
244 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, deviceId))); 256 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, deviceId)));
245 } else { 257 } else {
246 - return communicationService.sendAndReceive(createRequest(type, deviceId), 258 + return communicationService.sendAndReceive(createMetricsRequest(type, deviceId),
247 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId); 259 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
248 } 260 }
249 } 261 }
...@@ -255,7 +267,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -255,7 +267,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
255 if (clusterService.getLocalNode().id().equals(nodeId)) { 267 if (clusterService.getLocalNode().id().equals(nodeId)) {
256 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, resourceName))); 268 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, resourceName)));
257 } else { 269 } else {
258 - return communicationService.sendAndReceive(createRequest(type, resourceName), 270 + return communicationService.sendAndReceive(createMetricsRequest(type, resourceName),
259 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId); 271 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
260 } 272 }
261 } 273 }
...@@ -268,7 +280,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -268,7 +280,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
268 if (clusterService.getLocalNode().id().equals(nodeId)) { 280 if (clusterService.getLocalNode().id().equals(nodeId)) {
269 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, deviceId), duration, unit)); 281 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, deviceId), duration, unit));
270 } else { 282 } else {
271 - return communicationService.sendAndReceive(createRequest(type, duration, unit, deviceId), 283 + return communicationService.sendAndReceive(createMetricsRequest(type, duration, unit, deviceId),
272 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId); 284 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
273 } 285 }
274 } 286 }
...@@ -281,24 +293,22 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -281,24 +293,22 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
281 if (clusterService.getLocalNode().id().equals(nodeId)) { 293 if (clusterService.getLocalNode().id().equals(nodeId)) {
282 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, resourceName), duration, unit)); 294 return CompletableFuture.completedFuture(snapshot(getLocalLoad(type, resourceName), duration, unit));
283 } else { 295 } else {
284 - return communicationService.sendAndReceive(createRequest(type, duration, unit, resourceName), 296 + return communicationService.sendAndReceive(createMetricsRequest(type, duration, unit, resourceName),
285 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId); 297 CONTROL_STATS, SERIALIZER::encode, SERIALIZER::decode, nodeId);
286 } 298 }
287 } 299 }
288 300
289 @Override 301 @Override
290 - public Set<String> availableResources(Type resourceType) { 302 + public CompletableFuture<Set<String>> availableResources(NodeId nodeId,
291 - if (RESOURCE_TYPE_SET.contains(resourceType)) { 303 + Type resourceType) {
292 - if (Type.CONTROL_MESSAGE.equals(resourceType)) { 304 + if (clusterService.getLocalNode().id().equals(nodeId)) {
293 - return availableDeviceIdSet.stream().map(id -> 305 + Set<String> resources = getLocalAvailableResources(resourceType);
294 - id.toString()).collect(Collectors.toSet()); 306 + return CompletableFuture.completedFuture(resources);
295 } else { 307 } else {
296 - Set<String> res = availableResourceMap.get(resourceType); 308 + return communicationService.sendAndReceive(createResourceRequest(resourceType),
297 - return res == null ? ImmutableSet.of() : res; 309 + CONTROL_RESOURCE, SERIALIZER::encode, SERIALIZER::decode, nodeId);
298 } 310 }
299 } 311 }
300 - return ImmutableSet.of();
301 - }
302 312
303 /** 313 /**
304 * Builds and returns metric database instance with given resource name, 314 * Builds and returns metric database instance with given resource name,
...@@ -385,7 +395,8 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -385,7 +395,8 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
385 * @param request control metric request 395 * @param request control metric request
386 * @return completable future object of control load snapshot 396 * @return completable future object of control load snapshot
387 */ 397 */
388 - private CompletableFuture<ControlLoadSnapshot> handleRequest(ControlMetricsRequest request) { 398 + private CompletableFuture<ControlLoadSnapshot>
399 + handleMetricsRequest(ControlMetricsRequest request) {
389 400
390 checkArgument(request.getType() != null, METRIC_TYPE_NULL); 401 checkArgument(request.getType() != null, METRIC_TYPE_NULL);
391 402
...@@ -408,13 +419,28 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -408,13 +419,28 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
408 } 419 }
409 420
410 /** 421 /**
422 + * Handles control resource request from remote node.
423 + *
424 + * @param request control resource type
425 + * @return completable future object of control resource set
426 + */
427 + private CompletableFuture<Set<String>>
428 + handleResourceRequest(ControlResourceRequest request) {
429 +
430 + checkArgument(request.getType() != null, RESOURCE_TYPE_NULL);
431 +
432 + Set<String> resources = getLocalAvailableResources(request.getType());
433 + return CompletableFuture.completedFuture(resources);
434 + }
435 +
436 + /**
411 * Generates a control metric request. 437 * Generates a control metric request.
412 * 438 *
413 * @param type control metric type 439 * @param type control metric type
414 * @param deviceId device identifier 440 * @param deviceId device identifier
415 * @return control metric request instance 441 * @return control metric request instance
416 */ 442 */
417 - private ControlMetricsRequest createRequest(ControlMetricType type, 443 + private ControlMetricsRequest createMetricsRequest(ControlMetricType type,
418 Optional<DeviceId> deviceId) { 444 Optional<DeviceId> deviceId) {
419 return new ControlMetricsRequest(type, deviceId); 445 return new ControlMetricsRequest(type, deviceId);
420 } 446 }
...@@ -428,7 +454,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -428,7 +454,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
428 * @param deviceId device identifier 454 * @param deviceId device identifier
429 * @return control metric request instance 455 * @return control metric request instance
430 */ 456 */
431 - private ControlMetricsRequest createRequest(ControlMetricType type, 457 + private ControlMetricsRequest createMetricsRequest(ControlMetricType type,
432 int duration, TimeUnit unit, 458 int duration, TimeUnit unit,
433 Optional<DeviceId> deviceId) { 459 Optional<DeviceId> deviceId) {
434 return new ControlMetricsRequest(type, duration, unit, deviceId); 460 return new ControlMetricsRequest(type, duration, unit, deviceId);
...@@ -441,7 +467,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -441,7 +467,7 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
441 * @param resourceName resource name 467 * @param resourceName resource name
442 * @return control metric request instance 468 * @return control metric request instance
443 */ 469 */
444 - private ControlMetricsRequest createRequest(ControlMetricType type, 470 + private ControlMetricsRequest createMetricsRequest(ControlMetricType type,
445 String resourceName) { 471 String resourceName) {
446 return new ControlMetricsRequest(type, resourceName); 472 return new ControlMetricsRequest(type, resourceName);
447 } 473 }
...@@ -455,13 +481,23 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -455,13 +481,23 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
455 * @param resourceName resource name 481 * @param resourceName resource name
456 * @return control metric request instance 482 * @return control metric request instance
457 */ 483 */
458 - private ControlMetricsRequest createRequest(ControlMetricType type, 484 + private ControlMetricsRequest createMetricsRequest(ControlMetricType type,
459 int duration, TimeUnit unit, 485 int duration, TimeUnit unit,
460 String resourceName) { 486 String resourceName) {
461 return new ControlMetricsRequest(type, duration, unit, resourceName); 487 return new ControlMetricsRequest(type, duration, unit, resourceName);
462 } 488 }
463 489
464 /** 490 /**
491 + * Generates a control resource request with given resource type.
492 + *
493 + * @param type control resource type
494 + * @return control resource request instance
495 + */
496 + private ControlResourceRequest createResourceRequest(ControlResource.Type type) {
497 + return new ControlResourceRequest(type);
498 + }
499 +
500 + /**
465 * Returns a snapshot of control load. 501 * Returns a snapshot of control load.
466 * 502 *
467 * @param cl control load 503 * @param cl control load
...@@ -528,17 +564,39 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService { ...@@ -528,17 +564,39 @@ public class ControlPlaneMonitor implements ControlPlaneMonitorService {
528 * @return control load 564 * @return control load
529 */ 565 */
530 private ControlLoad getLocalLoad(ControlMetricType type, String resourceName) { 566 private ControlLoad getLocalLoad(ControlMetricType type, String resourceName) {
567 + NodeId localNodeId = clusterService.getLocalNode().id();
568 +
531 // returns disk I/O stats 569 // returns disk I/O stats
532 if (DISK_METRICS.contains(type) && 570 if (DISK_METRICS.contains(type) &&
533 - availableResources(Type.DISK).contains(resourceName)) { 571 + availableResourcesSync(localNodeId, Type.DISK).contains(resourceName)) {
534 return new DefaultControlLoad(diskMetricsMap.get(resourceName), type); 572 return new DefaultControlLoad(diskMetricsMap.get(resourceName), type);
535 } 573 }
536 574
537 // returns network I/O stats 575 // returns network I/O stats
538 if (NETWORK_METRICS.contains(type) && 576 if (NETWORK_METRICS.contains(type) &&
539 - availableResources(Type.NETWORK).contains(resourceName)) { 577 + availableResourcesSync(localNodeId, Type.NETWORK).contains(resourceName)) {
540 return new DefaultControlLoad(networkMetricsMap.get(resourceName), type); 578 return new DefaultControlLoad(networkMetricsMap.get(resourceName), type);
541 } 579 }
542 return null; 580 return null;
543 } 581 }
582 +
583 + /**
584 + * Obtains the available resource list from local node.
585 + *
586 + * @param resourceType control resource type
587 + * @return a set of available control resource
588 + */
589 + private Set<String> getLocalAvailableResources(Type resourceType) {
590 + Set<String> resources = ImmutableSet.of();
591 + if (RESOURCE_TYPE_SET.contains(resourceType)) {
592 + if (Type.CONTROL_MESSAGE.equals(resourceType)) {
593 + resources = ImmutableSet.copyOf(availableDeviceIdSet.stream()
594 + .map(DeviceId::toString).collect(Collectors.toSet()));
595 + } else {
596 + Set<String> res = availableResourceMap.get(resourceType);
597 + resources = res == null ? ImmutableSet.of() : res;
598 + }
599 + }
600 + return resources;
601 + }
544 } 602 }
......
...@@ -72,7 +72,7 @@ public class ControlMetricsWebResource extends AbstractWebResource { ...@@ -72,7 +72,7 @@ public class ControlMetricsWebResource extends AbstractWebResource {
72 public Response controlMessageMetrics() { 72 public Response controlMessageMetrics() {
73 73
74 ArrayNode deviceNodes = root.putArray("devices"); 74 ArrayNode deviceNodes = root.putArray("devices");
75 - monitorService.availableResources(CONTROL_MESSAGE).forEach(name -> { 75 + monitorService.availableResourcesSync(localNodeId, CONTROL_MESSAGE).forEach(name -> {
76 ObjectNode deviceNode = mapper().createObjectNode(); 76 ObjectNode deviceNode = mapper().createObjectNode();
77 ObjectNode valueNode = mapper().createObjectNode(); 77 ObjectNode valueNode = mapper().createObjectNode();
78 78
...@@ -147,7 +147,7 @@ public class ControlMetricsWebResource extends AbstractWebResource { ...@@ -147,7 +147,7 @@ public class ControlMetricsWebResource extends AbstractWebResource {
147 public Response diskMetrics() { 147 public Response diskMetrics() {
148 148
149 ArrayNode diskNodes = root.putArray("disks"); 149 ArrayNode diskNodes = root.putArray("disks");
150 - monitorService.availableResources(DISK).forEach(name -> { 150 + monitorService.availableResourcesSync(localNodeId, DISK).forEach(name -> {
151 ObjectNode diskNode = mapper().createObjectNode(); 151 ObjectNode diskNode = mapper().createObjectNode();
152 ObjectNode valueNode = mapper().createObjectNode(); 152 ObjectNode valueNode = mapper().createObjectNode();
153 153
...@@ -173,7 +173,7 @@ public class ControlMetricsWebResource extends AbstractWebResource { ...@@ -173,7 +173,7 @@ public class ControlMetricsWebResource extends AbstractWebResource {
173 public Response networkMetrics() { 173 public Response networkMetrics() {
174 174
175 ArrayNode networkNodes = root.putArray("networks"); 175 ArrayNode networkNodes = root.putArray("networks");
176 - monitorService.availableResources(NETWORK).forEach(name -> { 176 + monitorService.availableResourcesSync(localNodeId, NETWORK).forEach(name -> {
177 ObjectNode networkNode = mapper().createObjectNode(); 177 ObjectNode networkNode = mapper().createObjectNode();
178 ObjectNode valueNode = mapper().createObjectNode(); 178 ObjectNode valueNode = mapper().createObjectNode();
179 179
......
...@@ -240,7 +240,7 @@ public class ControlPlaneMonitorTest { ...@@ -240,7 +240,7 @@ public class ControlPlaneMonitorTest {
240 networkSet.forEach(network -> NETWORK_METRICS.forEach(cmt -> 240 networkSet.forEach(network -> NETWORK_METRICS.forEach(cmt ->
241 testUpdateMetricWithResource(cmt, mv, network))); 241 testUpdateMetricWithResource(cmt, mv, network)));
242 242
243 - assertThat(monitor.availableResources(Type.DISK), is(diskSet)); 243 + assertThat(monitor.availableResourcesSync(nodeId, Type.DISK), is(diskSet));
244 - assertThat(monitor.availableResources(Type.NETWORK), is(networkSet)); 244 + assertThat(monitor.availableResourcesSync(nodeId, Type.NETWORK), is(networkSet));
245 } 245 }
246 } 246 }
......
...@@ -75,7 +75,8 @@ public class ControlPlaneMonitorServiceAdaptor implements ControlPlaneMonitorSer ...@@ -75,7 +75,8 @@ public class ControlPlaneMonitorServiceAdaptor implements ControlPlaneMonitorSer
75 } 75 }
76 76
77 @Override 77 @Override
78 - public Set<String> availableResources(ControlResource.Type resourceType) { 78 + public CompletableFuture<Set<String>> availableResources(NodeId nodeId,
79 + ControlResource.Type resourceType) {
79 return null; 80 return null;
80 } 81 }
81 } 82 }
......
...@@ -166,7 +166,7 @@ public class ControlMetricsResourceTest extends ResourceTest { ...@@ -166,7 +166,7 @@ public class ControlMetricsResourceTest extends ResourceTest {
166 */ 166 */
167 @Test 167 @Test
168 public void testResourceEmptyArray() { 168 public void testResourceEmptyArray() {
169 - expect(mockControlPlaneMonitorService.availableResources(anyObject())) 169 + expect(mockControlPlaneMonitorService.availableResourcesSync(anyObject(), anyObject()))
170 .andReturn(ImmutableSet.of()).once(); 170 .andReturn(ImmutableSet.of()).once();
171 replay(mockControlPlaneMonitorService); 171 replay(mockControlPlaneMonitorService);
172 final WebTarget wt = target(); 172 final WebTarget wt = target();
...@@ -181,7 +181,7 @@ public class ControlMetricsResourceTest extends ResourceTest { ...@@ -181,7 +181,7 @@ public class ControlMetricsResourceTest extends ResourceTest {
181 */ 181 */
182 @Test 182 @Test
183 public void testResourcePopulatedArray() { 183 public void testResourcePopulatedArray() {
184 - expect(mockControlPlaneMonitorService.availableResources(anyObject())) 184 + expect(mockControlPlaneMonitorService.availableResourcesSync(anyObject(), anyObject()))
185 .andReturn(resourceSet).once(); 185 .andReturn(resourceSet).once();
186 expect(mockControlPlaneMonitorService.getLoad(anyObject(), anyObject(), 186 expect(mockControlPlaneMonitorService.getLoad(anyObject(), anyObject(),
187 anyString())).andReturn(null).times(4); 187 anyString())).andReturn(null).times(4);
......