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 a35cc505a7a46163f89ca81bb8bb4003e9f4cedc Author: Patrick Wiener <[email protected]> AuthorDate: Fri Apr 9 11:40:16 2021 +0200 [WIP] add skeleton for cluster resource monitoring --- .../management/pe/InvocableElementManager.java | 3 + .../resources/ClusterResourceManager.java | 96 ++++++++++++++++++++++ 2 files changed, 99 insertions(+) diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java index 0b49d50..2087587 100644 --- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java +++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/container/management/pe/InvocableElementManager.java @@ -158,6 +158,9 @@ public class InvocableElementManager implements PipelineElementLifeCycle { pub.publish(reconfigurationEvent); pub.disconnect(); + // TODO: update RunningInvocableInstances with new reconfiguration value + // remove, then add + response.setSuccess(true); return response; } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/resources/ClusterResourceManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/resources/ClusterResourceManager.java new file mode 100644 index 0000000..7b75c19 --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/node/management/resources/ClusterResourceManager.java @@ -0,0 +1,96 @@ +/* + * 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.resources; + +import org.apache.streampipes.model.node.NodeInfoDescription; +import org.apache.streampipes.storage.api.INodeInfoStorage; +import org.apache.streampipes.storage.management.StorageDispatcher; + +import java.net.InetSocketAddress; +import java.net.MalformedURLException; +import java.net.Socket; +import java.net.URL; +import java.util.List; + +public class ClusterResourceManager { + + private static final int RESOURCE_RETRIEVE_FREQUENCY_MS = 60000; + private static final int SOCKET_TIMEOUT_MS = 500; + private static ClusterResourceManager instance = null; + + private ClusterResourceManager() {} + + public static ClusterResourceManager getInstance() { + if (instance == null) { + synchronized (ClusterResourceManager.class) { + if (instance == null) + instance = new ClusterResourceManager(); + } + } + return instance; + } + + public void run() { + new Thread(getNodes, "nodes").start(); + } + + private final Runnable getNodes = () -> { + while (true) { + try { + List<NodeInfoDescription> nodes = getNodeStorageApi().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; + } + + // Helpers + + private static INodeInfoStorage getNodeStorageApi() { + return StorageDispatcher.INSTANCE.getNoSqlStore().getNodeStorage(); + } +}
