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 366492d0b1f67e346bc4784a3e12b9566517acb2 Author: Patrick Wiener <[email protected]> AuthorDate: Fri Apr 9 08:58:52 2021 +0200 [WIP] minor refactoring to StreamPipesClusterManager --- .../pipeline/AbstractPipelineExecutor.java | 8 +-- ...Manager.java => StreamPipesClusterManager.java} | 10 +-- .../cluster}/AbstractClusterManager.java | 38 +--------- .../cluster}/AvailableNodesFetcher.java | 2 +- .../{ => management/cluster}/NodeSyncOptions.java | 2 +- .../management/healthcheck/ClusterHealthCheck.java | 82 ++++++++++++++++++++++ .../org/apache/streampipes/rest/impl/Node.java | 19 +++-- 7 files changed, 104 insertions(+), 57 deletions(-) diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java index af0aced..e471c04 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/pipeline/AbstractPipelineExecutor.java @@ -21,7 +21,7 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException; import org.apache.streampipes.manager.data.PipelineGraph; import org.apache.streampipes.manager.data.PipelineGraphHelpers; import org.apache.streampipes.manager.execution.http.GraphSubmitter; -import org.apache.streampipes.manager.node.NodeClusterManager; +import org.apache.streampipes.manager.node.StreamPipesClusterManager; import org.apache.streampipes.manager.util.TemporaryGraphStorage; import org.apache.streampipes.model.SpDataSet; import org.apache.streampipes.model.SpDataStream; @@ -85,15 +85,15 @@ public abstract class AbstractPipelineExecutor { } protected void storeDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) { - relays.forEach(NodeClusterManager::persistDataStreamRelay); + relays.forEach(StreamPipesClusterManager::persistDataStreamRelay); } protected void deleteDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) { - relays.forEach(NodeClusterManager::deleteDataStreamRelay); + relays.forEach(StreamPipesClusterManager::deleteDataStreamRelay); } protected void updateDataStreamRelayContainer(List<SpDataStreamRelayContainer> relays) { - relays.forEach(NodeClusterManager::updateDataStreamRelay); + relays.forEach(StreamPipesClusterManager::updateDataStreamRelay); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/NodeClusterManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/StreamPipesClusterManager.java similarity index 93% rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/NodeClusterManager.java rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/StreamPipesClusterManager.java index 4e8b8f6..b3c29d3 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/NodeClusterManager.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/StreamPipesClusterManager.java @@ -17,6 +17,9 @@ */ package org.apache.streampipes.manager.node; +import org.apache.streampipes.manager.node.management.cluster.AbstractClusterManager; +import org.apache.streampipes.manager.node.management.cluster.NodeSyncOptions; +import org.apache.streampipes.manager.node.management.healthcheck.ClusterHealthCheck; import org.apache.streampipes.model.eventrelay.SpDataStreamRelayContainer; import org.apache.streampipes.model.message.Message; import org.apache.streampipes.model.message.NotificationType; @@ -31,15 +34,14 @@ import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.List; import java.util.Optional; -import java.util.stream.Collectors; -public class NodeClusterManager extends AbstractClusterManager { - private static final Logger LOG = LoggerFactory.getLogger(NodeClusterManager.class.getCanonicalName()); +public class StreamPipesClusterManager extends AbstractClusterManager { + private static final Logger LOG = LoggerFactory.getLogger(StreamPipesClusterManager.class.getCanonicalName()); public static List<NodeInfoDescription> getAvailableNodes() { List<NodeInfoDescription> availableAndHealthyNodes = new ArrayList<>(); getNodeStorageApi().getAllActiveNodes().forEach(node -> { - if (syncWithNodeController(node, NodeSyncOptions.HEALTHY)) { + if (ClusterHealthCheck.check(node)) { availableAndHealthyNodes.add(node); } }); diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/AbstractClusterManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/AbstractClusterManager.java similarity index 79% rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/AbstractClusterManager.java rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/AbstractClusterManager.java index 0973c3e..09e7ba0 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/AbstractClusterManager.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/AbstractClusterManager.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.manager.node; +package org.apache.streampipes.manager.node.management.cluster; import com.fasterxml.jackson.core.JsonProcessingException; import org.apache.http.client.fluent.Request; @@ -53,47 +53,11 @@ public abstract class AbstractClusterManager { return sync(element, "/api/v2/node/info", RequestOptions.PUT, true); case RESTART_RELAYS: return sync(element, "/api/v2/node/stream/relay/invoke", RequestOptions.POST, true); - case HEALTHY: - return healthCheck(element, "/healthy"); default: return false; } } - private static <T> boolean healthCheck(T element, String route) { - String url = generateEndpoint(element, route); - // call node controller REST endpoints - //return get(url).contains("PONG"); - - NodeInfoDescription desc = (NodeInfoDescription) element; - String nodeCtlId = desc.getNodeControllerId(); - boolean isAlive = false; - String host = desc.getHostname(); - int port = desc.getPort(); - - int retries = 5; - for (int i = 0 ; i < retries ; i++) { - try { - LOG.info("Trying to health check node controller={} ({})", nodeCtlId, i+1 + "/" + retries); - InetSocketAddress sa = new InetSocketAddress(host, port); - Socket ss = new Socket(); - ss.connect(sa, 500); - ss.close(); - if (ss.isConnected()) { - isAlive = true; - break; - } - Thread.sleep(1000); - } catch (IOException | InterruptedException e) { - continue; - } - isAlive = true; - } - LOG.info(isAlive ? "Successfully health check node controller=" + url : - "Could not perform health check node with controller=" + url); - return isAlive; - } - private static <T> boolean sync(T element, String route, RequestOptions request, boolean withBody) { boolean synced = false; diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/AvailableNodesFetcher.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/AvailableNodesFetcher.java similarity index 97% rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/AvailableNodesFetcher.java rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/AvailableNodesFetcher.java index 2071e12..e0c42a8 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/AvailableNodesFetcher.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/AvailableNodesFetcher.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.manager.node; +package org.apache.streampipes.manager.node.management.cluster; import org.apache.http.client.fluent.Request; import org.apache.streampipes.container.util.ConsulUtil; diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/NodeSyncOptions.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/NodeSyncOptions.java similarity index 93% rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/NodeSyncOptions.java rename to streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/NodeSyncOptions.java index 5ba7567..ae67076 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/NodeSyncOptions.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/cluster/NodeSyncOptions.java @@ -15,7 +15,7 @@ * limitations under the License. * */ -package org.apache.streampipes.manager.node; +package org.apache.streampipes.manager.node.management.cluster; public enum NodeSyncOptions { ACTIVATE_NODE, DEACTIVATE_NODE, UPDATE_NODE,RESTART_RELAYS, HEALTHY; diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/healthcheck/ClusterHealthCheck.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/healthcheck/ClusterHealthCheck.java new file mode 100644 index 0000000..9ca97c4 --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/healthcheck/ClusterHealthCheck.java @@ -0,0 +1,82 @@ +/* + * 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.manager.node.management.healthcheck; + +import org.apache.streampipes.model.node.NodeInfoDescription; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.Socket; + +public class ClusterHealthCheck { + private static final Logger LOG = LoggerFactory.getLogger(ClusterHealthCheck.class.getCanonicalName()); + + private static final String PROTOCOL = "http://";; + private static final String COLON = ":"; + private static final long RETRY_INTERVAL_MS = 5000; + private static final int CONNECT_TIMEOUT = 1000; + + public static boolean check(NodeInfoDescription node) { + return healthCheck(node); + } + + private static boolean healthCheck(NodeInfoDescription desc) { + String url = generateHealthCheckEndpoint(desc); + // call node controller REST endpoints + //return get(url).contains("PONG"); + + String nodeCtlId = desc.getNodeControllerId(); + boolean isAlive = false; + String host = desc.getHostname(); + int port = desc.getPort(); + + int retries = 5; + for (int i = 0 ; i < retries ; i++) { + try { + LOG.info("Trying to health check node controller={} ({})", nodeCtlId, i+1 + "/" + retries); + InetSocketAddress sa = new InetSocketAddress(host, port); + Socket ss = new Socket(); + ss.connect(sa, 500); + ss.close(); + if (ss.isConnected()) { + isAlive = true; + break; + } + Thread.sleep(1000); + } catch (IOException | InterruptedException e) { + continue; + } + isAlive = true; + } + LOG.info(isAlive ? "Successfully health check node controller=" + url : + "Could not perform health check node with controller=" + url); + return isAlive; + } + + // Helpers + + private static String generateHealthCheckEndpoint(NodeInfoDescription desc) { + return PROTOCOL + + desc.getHostname() + + COLON + + desc.getPort() + + "/healthy"; + } +} diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java index 02c0b0a..31f138c 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/Node.java @@ -17,8 +17,7 @@ */ package org.apache.streampipes.rest.impl; -import org.apache.streampipes.manager.info.VersionInfoProvider; -import org.apache.streampipes.manager.node.NodeClusterManager; +import org.apache.streampipes.manager.node.StreamPipesClusterManager; import org.apache.streampipes.model.message.NotificationType; import org.apache.streampipes.model.message.Notifications; import org.apache.streampipes.model.node.NodeInfoDescription; @@ -41,7 +40,7 @@ public class Node extends AbstractRestResource implements INode { @Produces(MediaType.APPLICATION_JSON) @Override public Response addNode(@PathParam("username") String username, NodeInfoDescription desc) { - return statusMessage(NodeClusterManager.addOrRejoin(desc)); + return statusMessage(StreamPipesClusterManager.addOrRejoin(desc)); } @PUT @@ -53,7 +52,7 @@ public class Node extends AbstractRestResource implements INode { public Response updateNode(@PathParam("username") String username, @PathParam("nodeControllerId") String nodeControllerId, NodeInfoDescription desc) { - return statusMessage(NodeClusterManager.updateNode(desc)); + return statusMessage(StreamPipesClusterManager.updateNode(desc)); } @POST @@ -64,7 +63,7 @@ public class Node extends AbstractRestResource implements INode { @Override public Response syncRemoteUpdateFromNodeController(@PathParam("username") String username, NodeInfoDescription desc) { - return statusMessage(NodeClusterManager.syncRemoteNodeUpdateRequest(desc)); + return statusMessage(StreamPipesClusterManager.syncRemoteNodeUpdateRequest(desc)); } @POST @@ -78,9 +77,9 @@ public class Node extends AbstractRestResource implements INode { @PathParam("nodeControllerId") String nodeControllerId) { boolean success = false; if (action.equals(ACTIVATE)) { - success = NodeClusterManager.activateNode(nodeControllerId); + success = StreamPipesClusterManager.activateNode(nodeControllerId); } else if (action.equals(DEACTIVATE)) { - success = NodeClusterManager.deactivateNode(nodeControllerId); + success = StreamPipesClusterManager.deactivateNode(nodeControllerId); } if (success) { return statusMessage(Notifications.success(NotificationType.OPERATION_SUCCESS)); @@ -94,7 +93,7 @@ public class Node extends AbstractRestResource implements INode { @Override public Response deleteNode(@PathParam("username") String username, @PathParam("nodeControllerId") String nodeControllerId) { - NodeClusterManager.deleteNode(nodeControllerId); + StreamPipesClusterManager.deleteNode(nodeControllerId); return statusMessage(Notifications.success(NotificationType.REMOVED_NODE)); } @@ -104,7 +103,7 @@ public class Node extends AbstractRestResource implements INode { @Produces(MediaType.APPLICATION_JSON) @Override public Response getAvailableNodes() { - return ok(NodeClusterManager.getAvailableNodes()); + return ok(StreamPipesClusterManager.getAvailableNodes()); } @GET @@ -112,6 +111,6 @@ public class Node extends AbstractRestResource implements INode { @Produces(MediaType.APPLICATION_JSON) @Override public Response getNodes() { - return ok(NodeClusterManager.getAllNodes()); + return ok(StreamPipesClusterManager.getAllNodes()); } }
