You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by wi...@apache.org on 2021/05/03 15:39:35 UTC
[incubator-streampipes] 02/02: add skeleton for node resource
management
This is an automated email from the ASF dual-hosted git repository.
wiener pushed a commit to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 3b5901be5c8e27f29dca8d04bbc8f768e7d24cd8
Author: Patrick Wiener <wi...@fzi.de>
AuthorDate: Mon May 3 17:22:19 2021 +0200
add skeleton for node resource management
---
.../backend/StreamPipesBackendApplication.java | 5 +-
.../model/node/monitor}/ResourceMetrics.java | 2 +-
.../api/NodeInfoDescriptionResource.java | 1 -
.../management/resource/ResourceManager.java | 2 +-
.../node/management/NodeManagement.java | 43 ++++++++---
.../{health/HealthCheck.java => NodeMonitor.java} | 10 +--
.../monitor/health/ClusterHealthCheckMonitor.java | 2 +-
.../operation/monitor/health/NodeHealthCheck.java | 29 +++----
.../monitor/resource/ClusterResourceMonitor.java | 65 ++++++++++++++++
.../NodeResourceCollector.java} | 49 ++++++------
.../monitor/resources/ClusterResourceMonitor.java | 90 ----------------------
.../migration/MigrationPipelineGenerator.java | 2 +-
.../rest/impl/NodeManagementResource.java | 14 ++--
13 files changed, 157 insertions(+), 157 deletions(-)
diff --git a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
index c5ca79b..95263dcb 100644
--- a/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
+++ b/streampipes-backend/src/main/java/org/apache/streampipes/backend/StreamPipesBackendApplication.java
@@ -23,6 +23,7 @@ import org.apache.shiro.web.servlet.ShiroFilter;
import org.apache.streampipes.manager.operations.Operations;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.node.management.NodeManagement;
import org.apache.streampipes.node.management.operation.monitor.health.ClusterHealthCheckMonitor;
import org.apache.streampipes.rest.notifications.NotificationListener;
import org.apache.streampipes.storage.management.StorageDispatcher;
@@ -61,8 +62,8 @@ public class StreamPipesBackendApplication {
ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
executorService.schedule(this::startAllPreviouslyStoppedPipelines, 5, TimeUnit.SECONDS);
- LOG.info("Starting StreamPipes cluster monitor...");
- ClusterHealthCheckMonitor.getInstance().run();
+ LOG.info("Starting StreamPipes node management ...");
+ NodeManagement.getInstance().init();
}
@PreDestroy
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/model/ResourceMetrics.java b/streampipes-model/src/main/java/org/apache/streampipes/model/node/monitor/ResourceMetrics.java
similarity index 97%
rename from streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/model/ResourceMetrics.java
rename to streampipes-model/src/main/java/org/apache/streampipes/model/node/monitor/ResourceMetrics.java
index ddaaff6..ac33845 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/model/ResourceMetrics.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/node/monitor/ResourceMetrics.java
@@ -15,7 +15,7 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.node.controller.management.resource.model;
+package org.apache.streampipes.model.node.monitor;
public class ResourceMetrics {
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeInfoDescriptionResource.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeInfoDescriptionResource.java
index 4e4aa76..d37e246 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeInfoDescriptionResource.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/api/NodeInfoDescriptionResource.java
@@ -17,7 +17,6 @@
*/
package org.apache.streampipes.node.controller.api;
-import org.apache.streampipes.model.node.NodeCondition;
import org.apache.streampipes.model.node.NodeInfoDescription;
import org.apache.streampipes.node.controller.management.node.NodeManager;
import org.apache.streampipes.node.controller.management.relay.DataStreamRelayManager;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java
index e0d185b..737f521 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/resource/ResourceManager.java
@@ -24,7 +24,7 @@ import org.apache.streampipes.node.controller.management.offloading.AutoOffloadi
import org.apache.streampipes.node.controller.management.offloading.policies.Comparator;
import org.apache.streampipes.node.controller.management.offloading.policies.MultiOccurrenceThresholdViolationPolicy;
import org.apache.streampipes.node.controller.management.offloading.policies.OffloadingPolicy;
-import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
import org.apache.streampipes.node.controller.management.resource.utils.DiskSpace;
import org.apache.streampipes.node.controller.management.resource.utils.ResourceUtils;
import org.apache.streampipes.serializers.json.JacksonSerializer;
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/NodeManagement.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/NodeManagement.java
index c4add7d..db5dbcb 100644
--- a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/NodeManagement.java
+++ b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/NodeManagement.java
@@ -25,6 +25,7 @@ import org.apache.streampipes.model.message.Notifications;
import org.apache.streampipes.model.node.NodeCondition;
import org.apache.streampipes.model.node.NodeInfoDescription;
import org.apache.streampipes.node.management.operation.monitor.health.ClusterHealthCheckMonitor;
+import org.apache.streampipes.node.management.operation.monitor.resource.ClusterResourceMonitor;
import org.apache.streampipes.node.management.operation.relay.RelayHandler;
import org.apache.streampipes.node.management.operation.sync.SynchronizationFactory;
import org.apache.streampipes.node.management.operation.sync.SynchronizationType;
@@ -36,17 +37,41 @@ import java.util.List;
import java.util.Optional;
public class NodeManagement {
+
private static final Logger LOG = LoggerFactory.getLogger(NodeManagement.class.getCanonicalName());
+ private static NodeManagement instance = null;
+
+ private NodeManagement() {
+ }
+
+ public static NodeManagement getInstance() {
+ if (instance == null) {
+ synchronized (NodeManagement.class) {
+ if (instance == null)
+ instance = new NodeManagement();
+ }
+ }
+ return instance;
+ }
+
+
+ public void init() {
+ LOG.info("Starting StreamPipes healthcheck monitor...");
+ ClusterHealthCheckMonitor.getInstance().run();
+
+ LOG.info("Starting StreamPipes resource monitor...");
+ ClusterResourceMonitor.getInstance().run();
+ }
- public static List<NodeInfoDescription> getOnlineNodes() {
+ public List<NodeInfoDescription> getOnlineNodes() {
return ClusterHealthCheckMonitor.getInstance().getAllHealthyNodes();
}
- public static List<NodeInfoDescription> getAllNodes() {
+ public List<NodeInfoDescription> getAllNodes() {
return StorageUtils.getAllNodes();
}
- public static boolean updateNodeCondition(String nodeControllerId, NodeCondition condition) {
+ public boolean updateNodeCondition(String nodeControllerId, NodeCondition condition) {
Optional<NodeInfoDescription> storedNode = StorageUtils.getNode(nodeControllerId);
boolean status = false;
@@ -69,7 +94,7 @@ public class NodeManagement {
return status;
}
- public static Message updateNode(NodeInfoDescription desc) {
+ public Message updateNode(NodeInfoDescription desc) {
boolean successfullyUpdated = SynchronizationFactory.synchronize(desc, SynchronizationType.UPDATE_NODE);
if (successfullyUpdated) {
StorageUtils.updateNode(desc);
@@ -78,16 +103,16 @@ public class NodeManagement {
return Notifications.error("Could not update node");
}
- public static void deleteNode(String nodeControllerId) {
+ public void deleteNode(String nodeControllerId) {
StorageUtils.deleteNode(nodeControllerId);
}
- public static Message syncRemoteNodeUpdateRequest(NodeInfoDescription desc) {
+ public Message syncRemoteNodeUpdateRequest(NodeInfoDescription desc) {
StorageUtils.updateNode(desc);
return Notifications.success("Node updated");
}
- public static Message addOrRejoin(NodeInfoDescription desc) {
+ public Message addOrRejoin(NodeInfoDescription desc) {
Optional<NodeInfoDescription> latestDesc = StorageUtils.getLatestNodeOrElseEmpty(desc.getNodeControllerId());
boolean alreadyRegistered = false;
@@ -104,7 +129,7 @@ public class NodeManagement {
}
}
- private static Message addNewNode(NodeInfoDescription desc) throws RuntimeException {
+ private Message addNewNode(NodeInfoDescription desc) throws RuntimeException {
try {
StorageUtils.storeNode(desc);
LOG.info("New cluster node successfully joined http://{}:{}", desc.getHostname(), desc.getPort());
@@ -114,7 +139,7 @@ public class NodeManagement {
}
}
- private static Message rejoinAndSyncNode(NodeInfoDescription desc) {
+ private Message rejoinAndSyncNode(NodeInfoDescription desc) {
LOG.info("Sync latest node description to http://{}:{}", desc.getHostname(), desc.getPort());
boolean success = SynchronizationFactory.synchronize(desc, SynchronizationType.UPDATE_NODE);
if (success) {
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/HealthCheck.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/NodeMonitor.java
similarity index 83%
rename from streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/HealthCheck.java
rename to streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/NodeMonitor.java
index 784cb9c..f54ce73 100644
--- a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/HealthCheck.java
+++ b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/NodeMonitor.java
@@ -15,13 +15,11 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.node.management.operation.monitor.health;
-
-import org.apache.streampipes.model.NodeHealthStatus;
+package org.apache.streampipes.node.management.operation.monitor;
import java.util.concurrent.Callable;
-public interface HealthCheck {
- Callable<NodeHealthStatus> nodeHealthStatusCallable();
- NodeHealthStatus execute();
+public interface NodeMonitor<T> {
+ Callable<T> monitoringCallable();
+ T execute();
}
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/ClusterHealthCheckMonitor.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/ClusterHealthCheckMonitor.java
index 70c8fba..6fa823d 100644
--- a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/ClusterHealthCheckMonitor.java
+++ b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/ClusterHealthCheckMonitor.java
@@ -162,7 +162,7 @@ public class ClusterHealthCheckMonitor {
private void persistNodeUpdate(NodeInfoDescription node, boolean syncWithNodeController) {
if (syncWithNodeController) {
- NodeManagement.updateNode(node);
+ NodeManagement.getInstance().updateNode(node);
} else {
StorageUtils.updateNode(node);
}
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/NodeHealthCheck.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/NodeHealthCheck.java
index a0a0dc3..531bedd 100644
--- a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/NodeHealthCheck.java
+++ b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/NodeHealthCheck.java
@@ -18,13 +18,14 @@
package org.apache.streampipes.node.management.operation.monitor.health;
import org.apache.streampipes.model.NodeHealthStatus;
+import org.apache.streampipes.node.management.operation.monitor.NodeMonitor;
import org.apache.streampipes.node.management.utils.HttpUtils;
import java.util.concurrent.*;
-public class NodeHealthCheck implements HealthCheck {
+public class NodeHealthCheck implements NodeMonitor<NodeHealthStatus> {
- private static final int HEALTH_CHECK_FUTURE_TIMEOUT_SECS = 3;
+ private static final int FUTURE_TIMEOUT_SECS = 3;
private final String healthCheckEndpoint;
public NodeHealthCheck(String endpoint) {
@@ -32,15 +33,25 @@ public class NodeHealthCheck implements HealthCheck {
}
@Override
+ public Callable<NodeHealthStatus> monitoringCallable() {
+ return () -> {
+ if (!Thread.currentThread().isInterrupted()) {
+ return HttpUtils.get(healthCheckEndpoint, NodeHealthStatus.class);
+ }
+ return new NodeHealthStatus(false);
+ };
+ }
+
+ @Override
public NodeHealthStatus execute() {
ExecutorService executorService = Executors.newSingleThreadExecutor();
- final Future<NodeHealthStatus> future = executorService.submit(nodeHealthStatusCallable());
+ final Future<NodeHealthStatus> future = executorService.submit(monitoringCallable());
NodeHealthStatus nodeHealthStatus;
try {
// blocking call until timeout is reached
- nodeHealthStatus = future.get(HEALTH_CHECK_FUTURE_TIMEOUT_SECS, TimeUnit.SECONDS);
+ nodeHealthStatus = future.get(FUTURE_TIMEOUT_SECS, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException e) {
nodeHealthStatus = new NodeHealthStatus(false);
} catch (TimeoutException e) {
@@ -50,14 +61,4 @@ public class NodeHealthCheck implements HealthCheck {
executorService.shutdown();
return nodeHealthStatus;
}
-
- @Override
- public Callable<NodeHealthStatus> nodeHealthStatusCallable() {
- return () -> {
- if (!Thread.currentThread().isInterrupted()) {
- return HttpUtils.get(healthCheckEndpoint, NodeHealthStatus.class);
- }
- return new NodeHealthStatus(false);
- };
- }
}
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/ClusterResourceMonitor.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/ClusterResourceMonitor.java
new file mode 100644
index 0000000..008c3c4
--- /dev/null
+++ b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/ClusterResourceMonitor.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.streampipes.node.management.operation.monitor.resource;
+
+
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
+import org.apache.streampipes.node.management.operation.monitor.health.ClusterHealthCheckMonitor;
+import org.apache.streampipes.node.management.utils.HttpUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class ClusterResourceMonitor {
+ private static final Logger LOG = LoggerFactory.getLogger(ClusterResourceMonitor.class.getCanonicalName());
+
+ private static final String RESOURCE_COLLECTOR_ROUTE = "/api/v2/node/info/resources";
+ private static final int RESOURCE_COLLECTOR_INTERVAL_SECS = 60;
+ private static ClusterResourceMonitor instance = null;
+
+ private ClusterResourceMonitor() {}
+
+ public static ClusterResourceMonitor getInstance() {
+ if (instance == null) {
+ synchronized (ClusterResourceMonitor.class) {
+ if (instance == null)
+ instance = new ClusterResourceMonitor();
+ }
+ }
+ return instance;
+ }
+
+ public void run() {
+ ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+ executor.scheduleAtFixedRate(resourceCollector, 0, RESOURCE_COLLECTOR_INTERVAL_SECS, TimeUnit.SECONDS);
+ }
+
+ private final Runnable resourceCollector = () -> {
+ // Only collect resources for online nodes, others might not be reachable
+ ClusterHealthCheckMonitor.getInstance().getAllHealthyNodes().forEach(node -> {
+ // collect resources from node
+ String endpoint = HttpUtils.generateEndpoint(node, RESOURCE_COLLECTOR_ROUTE);
+ ResourceMetrics resourceMetrics = new NodeResourceCollector(endpoint).execute();
+
+ // TODO: implement
+ });
+ };
+}
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/NodeHealthCheck.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/NodeResourceCollector.java
similarity index 57%
copy from streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/NodeHealthCheck.java
copy to streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/NodeResourceCollector.java
index a0a0dc3..d2f574b 100644
--- a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/health/NodeHealthCheck.java
+++ b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resource/NodeResourceCollector.java
@@ -15,49 +15,50 @@
* limitations under the License.
*
*/
-package org.apache.streampipes.node.management.operation.monitor.health;
+package org.apache.streampipes.node.management.operation.monitor.resource;
-import org.apache.streampipes.model.NodeHealthStatus;
+import org.apache.streampipes.model.node.monitor.ResourceMetrics;
+import org.apache.streampipes.node.management.operation.monitor.NodeMonitor;
import org.apache.streampipes.node.management.utils.HttpUtils;
import java.util.concurrent.*;
-public class NodeHealthCheck implements HealthCheck {
+public class NodeResourceCollector implements NodeMonitor<ResourceMetrics> {
- private static final int HEALTH_CHECK_FUTURE_TIMEOUT_SECS = 3;
- private final String healthCheckEndpoint;
+ private static final int FUTURE_TIMEOUT_SECS = 3;
+ private final String resourceCollectorEndpoint;
- public NodeHealthCheck(String endpoint) {
- this.healthCheckEndpoint = endpoint;
+ public NodeResourceCollector(String resourceCollectorEndpoint) {
+ this.resourceCollectorEndpoint = resourceCollectorEndpoint;
}
@Override
- public NodeHealthStatus execute() {
+ public Callable<ResourceMetrics> monitoringCallable() {
+ return () -> {
+ if (!Thread.currentThread().isInterrupted()) {
+ return HttpUtils.get(resourceCollectorEndpoint, ResourceMetrics.class);
+ }
+ return new ResourceMetrics();
+ };
+ }
+
+ @Override
+ public ResourceMetrics execute() {
ExecutorService executorService = Executors.newSingleThreadExecutor();
- final Future<NodeHealthStatus> future = executorService.submit(nodeHealthStatusCallable());
+ final Future<ResourceMetrics> future = executorService.submit(monitoringCallable());
- NodeHealthStatus nodeHealthStatus;
+ ResourceMetrics resourceMetrics;
try {
// blocking call until timeout is reached
- nodeHealthStatus = future.get(HEALTH_CHECK_FUTURE_TIMEOUT_SECS, TimeUnit.SECONDS);
+ resourceMetrics = future.get(FUTURE_TIMEOUT_SECS, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException e) {
- nodeHealthStatus = new NodeHealthStatus(false);
+ resourceMetrics = new ResourceMetrics();
} catch (TimeoutException e) {
future.cancel(true);
- nodeHealthStatus = new NodeHealthStatus(false);
+ resourceMetrics = new ResourceMetrics();
}
executorService.shutdown();
- return nodeHealthStatus;
- }
-
- @Override
- public Callable<NodeHealthStatus> nodeHealthStatusCallable() {
- return () -> {
- if (!Thread.currentThread().isInterrupted()) {
- return HttpUtils.get(healthCheckEndpoint, NodeHealthStatus.class);
- }
- return new NodeHealthStatus(false);
- };
+ return resourceMetrics;
}
}
diff --git a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resources/ClusterResourceMonitor.java b/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resources/ClusterResourceMonitor.java
deleted file mode 100644
index 2d1b2ff..0000000
--- a/streampipes-node-management/src/main/java/org/apache/streampipes/node/management/operation/monitor/resources/ClusterResourceMonitor.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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.apache.streampipes.node.management.operation.monitor.resources;
-
-
-import org.apache.streampipes.model.node.NodeInfoDescription;
-import org.apache.streampipes.node.management.utils.StorageUtils;
-
-import java.net.InetSocketAddress;
-import java.net.MalformedURLException;
-import java.net.Socket;
-import java.net.URL;
-import java.util.List;
-
-public class ClusterResourceMonitor {
-
- private static final int RESOURCE_RETRIEVE_FREQUENCY_MS = 60000;
- private static final int SOCKET_TIMEOUT_MS = 500;
- private static ClusterResourceMonitor instance = null;
-
- private ClusterResourceMonitor() {}
-
- public static ClusterResourceMonitor getInstance() {
- if (instance == null) {
- synchronized (ClusterResourceMonitor.class) {
- if (instance == null)
- instance = new ClusterResourceMonitor();
- }
- }
- return instance;
- }
-
- public void run() {
- new Thread(getNodes, "nodes").start();
- }
-
- private final Runnable getNodes = () -> {
- while (true) {
- try {
- List<NodeInfoDescription> nodes = StorageUtils.persistentNodeAPI().getAllNodes();
- if (nodes.size() > 0) {
- nodes.forEach(node -> {
- try {
- URL nodeUrl = generateNodeUrl(node);
- // TODO: gather current resources from all active node controller endpoints
-
- } catch (MalformedURLException e) {
- e.printStackTrace();
- }
- });
- }
- Thread.sleep(RESOURCE_RETRIEVE_FREQUENCY_MS);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- };
-
- private URL generateNodeUrl(NodeInfoDescription desc) throws MalformedURLException {
- return new URL("http", desc.getHostname(), desc.getPort(), "");
- }
-
- private boolean healthCheck(URL url) {
- boolean isAlive = true;
- try {
- InetSocketAddress sa = new InetSocketAddress(url.getHost(), url.getPort());
- Socket ss = new Socket();
- ss.connect(sa, SOCKET_TIMEOUT_MS);
- ss.close();
- } catch(Exception e) {
- isAlive = false;
- }
- return isAlive;
- }
-}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
index bfb209c..f69a3b9 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/MigrationPipelineGenerator.java
@@ -35,7 +35,7 @@ public class MigrationPipelineGenerator {
public static Pipeline generateMigrationPipeline(InvocableStreamPipesEntity entityToMigrate, Pipeline correspondingPipeline){
List<NodeInfoDescription> possibleTargetNodes = new ArrayList<>();
- List<NodeInfoDescription> nodeInfo = NodeManagement.getOnlineNodes();
+ List<NodeInfoDescription> nodeInfo = NodeManagement.getInstance().getOnlineNodes();
nodeInfo.forEach(desc ->{
if(desc.getSupportedElements().stream().anyMatch(element -> element.equals(entityToMigrate.getAppId()))
&& !desc.getNodeControllerId().equals(entityToMigrate.getDeploymentTargetNodeId()))
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/NodeManagementResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/NodeManagementResource.java
index c9acd31..65d58b8 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/NodeManagementResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/NodeManagementResource.java
@@ -39,7 +39,7 @@ public class NodeManagementResource extends AbstractRestResource implements INod
@Produces(MediaType.APPLICATION_JSON)
@Override
public Response addNode(@PathParam("username") String username, NodeInfoDescription desc) {
- return statusMessage(NodeManagement.addOrRejoin(desc));
+ return statusMessage(NodeManagement.getInstance().addOrRejoin(desc));
}
@PUT
@@ -51,7 +51,7 @@ public class NodeManagementResource extends AbstractRestResource implements INod
public Response updateNode(@PathParam("username") String username,
@PathParam("nodeControllerId") String nodeControllerId,
NodeInfoDescription desc) {
- return statusMessage(NodeManagement.updateNode(desc));
+ return statusMessage(NodeManagement.getInstance().updateNode(desc));
}
@POST
@@ -62,7 +62,7 @@ public class NodeManagementResource extends AbstractRestResource implements INod
@Override
public Response syncRemoteUpdateFromNodeController(@PathParam("username") String username,
NodeInfoDescription desc) {
- return statusMessage(NodeManagement.syncRemoteNodeUpdateRequest(desc));
+ return statusMessage(NodeManagement.getInstance().syncRemoteNodeUpdateRequest(desc));
}
@POST
@@ -75,7 +75,7 @@ public class NodeManagementResource extends AbstractRestResource implements INod
@PathParam("username") String username,
@PathParam("nodeControllerId") String nodeControllerId) {
NodeCondition nodeCondition = NodeCondition.valueOf(condition.toUpperCase());
- boolean success = NodeManagement.updateNodeCondition(nodeControllerId, nodeCondition);
+ boolean success = NodeManagement.getInstance().updateNodeCondition(nodeControllerId, nodeCondition);
if (success) {
return statusMessage(Notifications.success(NotificationType.OPERATION_SUCCESS));
} else {
@@ -88,7 +88,7 @@ public class NodeManagementResource extends AbstractRestResource implements INod
@Override
public Response deleteNode(@PathParam("username") String username,
@PathParam("nodeControllerId") String nodeControllerId) {
- NodeManagement.deleteNode(nodeControllerId);
+ NodeManagement.getInstance().deleteNode(nodeControllerId);
return statusMessage(Notifications.success(NotificationType.REMOVED_NODE));
}
@@ -98,7 +98,7 @@ public class NodeManagementResource extends AbstractRestResource implements INod
@Produces(MediaType.APPLICATION_JSON)
@Override
public Response getOnlineNodes() {
- return ok(NodeManagement.getOnlineNodes());
+ return ok(NodeManagement.getInstance().getOnlineNodes());
}
@GET
@@ -106,6 +106,6 @@ public class NodeManagementResource extends AbstractRestResource implements INod
@Produces(MediaType.APPLICATION_JSON)
@Override
public Response getNodes() {
- return ok(NodeManagement.getAllNodes());
+ return ok(NodeManagement.getInstance().getAllNodes());
}
}