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 <[email protected]> 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()); } }
