You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by go...@apache.org on 2021/05/10 21:03:14 UTC
[incubator-streampipes] 03/07: [WIP] Check for Hardware requirement
to filter supported entities
This is an automated email from the ASF dual-hosted git repository.
gomm pushed a commit to branch edge-extensions
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 657b34c91a32e9bed4895827e236244670c4f9b9
Author: Daniel Gomm <da...@outlook.de>
AuthorDate: Sun May 2 13:35:46 2021 +0200
[WIP] Check for Hardware requirement to filter supported entities
---
.../extensions/ExtensionsModelSubmitter.java | 24 +++++++++++++--------
.../standalone/init/StandaloneModelSubmitter.java | 25 ++++++++++++++--------
.../model/node/InvocableRegistration.java | 15 +++++++------
.../container/util/NodeControllerUtil.java | 15 +++++++------
.../offloading/OffloadingPolicyManager.java | 18 ++++++++++++++++
.../offloading/model/OffloadingStrategy.java | 18 ++++++++++++++++
.../model/property/CPULoadResourceProperty.java | 18 ++++++++++++++++
.../property/FreeDiskSpaceResourceProperty.java | 18 ++++++++++++++++
.../model/property/FreeMemoryResourceProperty.java | 18 ++++++++++++++++
.../model/property/ResourceProperty.java | 18 ++++++++++++++++
.../model/selection/CPULoadSelectionStrategy.java | 18 ++++++++++++++++
.../model/selection/RandomSelectionStrategy.java | 18 ++++++++++++++++
.../model/selection/SelectionStrategy.java | 18 ++++++++++++++++
.../management/pe/InvocableElementManager.java | 21 +++++++++++++++---
.../management/resource/ResourceManager.java | 17 +++++++--------
15 files changed, 236 insertions(+), 43 deletions(-)
diff --git a/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java b/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java
index 56a6004..07ee0d8 100644
--- a/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java
+++ b/streampipes-container-extensions/src/main/java/org/apache/streampipes/container/extensions/ExtensionsModelSubmitter.java
@@ -22,12 +22,12 @@ import org.apache.streampipes.connect.adapter.model.generic.Protocol;
import org.apache.streampipes.connect.container.worker.management.MasterRestClient;
import org.apache.streampipes.connect.container.worker.management.NodeControllerRestClient;
import org.apache.streampipes.connect.init.AdapterDeclarerSingleton;
+import org.apache.streampipes.container.declarer.Declarer;
import org.apache.streampipes.container.init.DeclarersSingleton;
import org.apache.streampipes.container.init.ModelSubmitter;
import org.apache.streampipes.container.init.RunningInstances;
import org.apache.streampipes.container.locales.LabelGenerator;
import org.apache.streampipes.container.model.EdgeExtensionsConfig;
-import org.apache.streampipes.container.model.ExtensionsConfig;
import org.apache.streampipes.container.util.ConsulUtil;
import org.apache.streampipes.container.util.NodeControllerUtil;
import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory;
@@ -37,6 +37,8 @@ import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory;
import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
+import org.apache.streampipes.model.base.ConsumableStreamPipesEntity;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
import org.apache.streampipes.model.connect.adapter.GenericAdapterDescription;
@@ -52,6 +54,7 @@ import org.springframework.context.annotation.Import;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.*;
+import java.util.stream.Collectors;
@Configuration
@EnableAutoConfiguration
@@ -97,7 +100,7 @@ public abstract class ExtensionsModelSubmitter extends ModelSubmitter<EdgeExtens
if (System.getenv(NODE_CONTROLLER_ID) != null) {
// secondary
// register pipeline element service via node controller
- NodeControllerUtil.register(conf, generateSupportedAppIds());
+ NodeControllerUtil.register(conf, generateSupportedEntities());
String nodeControllerUrl = PROTOCOL + conf.getNodeControllerHost() + COLON + conf.getNodeControllerPort();
@@ -146,16 +149,19 @@ public abstract class ExtensionsModelSubmitter extends ModelSubmitter<EdgeExtens
}
}
- private List<String> generateSupportedAppIds() {
- List<String> supportedAppIds = new ArrayList<>();
+ private List<ConsumableStreamPipesEntity> generateSupportedEntities() {
+ List<ConsumableStreamPipesEntity> supportedEntities = new ArrayList<>();
- List<String> dataProcessorsAppIds = new ArrayList<>(DeclarersSingleton.getInstance().getEpaDeclarers().keySet());
- List<String> dataSinksAppIds = new ArrayList<>(DeclarersSingleton.getInstance().getConsumerDeclarers().keySet());
+ List<ConsumableStreamPipesEntity> dataProcessors =
+ DeclarersSingleton.getInstance().getEpaDeclarers().values()
+ .stream().map(Declarer::declareModel).collect(Collectors.toList());
+ List<ConsumableStreamPipesEntity> dataSinks = DeclarersSingleton.getInstance().getConsumerDeclarers().values()
+ .stream().map(Declarer::declareModel).collect(Collectors.toList());
- supportedAppIds.addAll(dataProcessorsAppIds);
- supportedAppIds.addAll(dataSinksAppIds);
+ supportedEntities.addAll(dataProcessors);
+ supportedEntities.addAll(dataSinks);
- return supportedAppIds;
+ return supportedEntities;
}
private ConnectWorkerContainer getContainerDescription(EdgeExtensionsConfig conf, boolean runsOnEdgeNode) {
diff --git a/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java b/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java
index f8e25ce..0086dba 100644
--- a/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java
+++ b/streampipes-container-standalone/src/main/java/org/apache/streampipes/container/standalone/init/StandaloneModelSubmitter.java
@@ -19,7 +19,10 @@
package org.apache.streampipes.container.standalone.init;
+import org.apache.streampipes.container.declarer.Declarer;
import org.apache.streampipes.container.util.NodeControllerUtil;
+import org.apache.streampipes.model.base.ConsumableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
@@ -35,6 +38,7 @@ import org.apache.streampipes.container.util.ConsulUtil;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.stream.Collectors;
import javax.annotation.PreDestroy;
@@ -59,7 +63,7 @@ public abstract class StandaloneModelSubmitter extends ModelSubmitter<PeConfig>
app.setDefaultProperties(Collections.singletonMap("server.port", peConfig.getPort()));
app.run();
- // check wether pipeline element is managed by node controller
+ // check whether pipeline element is managed by node controller
if (System.getenv("SP_NODE_CONTROLLER_ID") != null) {
// secondary
// register pipeline element service via node controller
@@ -67,7 +71,7 @@ public abstract class StandaloneModelSubmitter extends ModelSubmitter<PeConfig>
peConfig.getId(),
peConfig.getHost(),
peConfig.getPort(),
- generateSupportedAppIds());
+ generateSupportedEntities());
} else {
// primary
ConsulUtil.registerPeService(
@@ -77,16 +81,19 @@ public abstract class StandaloneModelSubmitter extends ModelSubmitter<PeConfig>
}
}
- private List<String> generateSupportedAppIds() {
- List<String> supportedAppIds = new ArrayList<>();
+ private List<ConsumableStreamPipesEntity> generateSupportedEntities() {
+ List<ConsumableStreamPipesEntity> supportedEntities = new ArrayList<>();
- List<String> dataProcessorsAppIds = new ArrayList<>(DeclarersSingleton.getInstance().getEpaDeclarers().keySet());
- List<String> dataSinksAppIds = new ArrayList<>(DeclarersSingleton.getInstance().getConsumerDeclarers().keySet());
+ List<ConsumableStreamPipesEntity> dataProcessors =
+ DeclarersSingleton.getInstance().getEpaDeclarers().values()
+ .stream().map(Declarer::declareModel).collect(Collectors.toList());
+ List<ConsumableStreamPipesEntity> dataSinks = DeclarersSingleton.getInstance().getConsumerDeclarers().values()
+ .stream().map(Declarer::declareModel).collect(Collectors.toList());
- supportedAppIds.addAll(dataProcessorsAppIds);
- supportedAppIds.addAll(dataSinksAppIds);
+ supportedEntities.addAll(dataProcessors);
+ supportedEntities.addAll(dataSinks);
- return supportedAppIds;
+ return supportedEntities;
}
@PreDestroy
diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/model/node/InvocableRegistration.java b/streampipes-container/src/main/java/org/apache/streampipes/container/model/node/InvocableRegistration.java
index 4605b6c..3c162f4 100644
--- a/streampipes-container/src/main/java/org/apache/streampipes/container/model/node/InvocableRegistration.java
+++ b/streampipes-container/src/main/java/org/apache/streampipes/container/model/node/InvocableRegistration.java
@@ -17,21 +17,22 @@ package org.apache.streampipes.container.model.node;/*
*/
import org.apache.streampipes.container.model.consul.ConsulServiceRegistrationBody;
+import org.apache.streampipes.model.base.ConsumableStreamPipesEntity;
import java.util.List;
public class InvocableRegistration {
private ConsulServiceRegistrationBody consulServiceRegistrationBody;
- private List<String> supportedPipelineElementAppIds;
+ private List<ConsumableStreamPipesEntity> supportedPipelineElements;
public InvocableRegistration() {
}
public InvocableRegistration(ConsulServiceRegistrationBody consulServiceRegistrationBody,
- List<String> supportedPipelineElementAppIds) {
+ List<ConsumableStreamPipesEntity> supportedPipelineElementAppIds) {
this.consulServiceRegistrationBody = consulServiceRegistrationBody;
- this.supportedPipelineElementAppIds = supportedPipelineElementAppIds;
+ this.supportedPipelineElements = supportedPipelineElementAppIds;
}
public ConsulServiceRegistrationBody getConsulServiceRegistrationBody() {
@@ -42,11 +43,11 @@ public class InvocableRegistration {
this.consulServiceRegistrationBody = consulServiceRegistrationBody;
}
- public List<String> getSupportedPipelineElementAppIds() {
- return supportedPipelineElementAppIds;
+ public List<ConsumableStreamPipesEntity> getSupportedPipelineElements() {
+ return supportedPipelineElements;
}
- public void setSupportedPipelineElementAppIds(List<String> supportedPipelineElementAppIds) {
- this.supportedPipelineElementAppIds = supportedPipelineElementAppIds;
+ public void setSupportedPipelineElements(List<ConsumableStreamPipesEntity> supportedPipelineElements) {
+ this.supportedPipelineElements = supportedPipelineElements;
}
}
diff --git a/streampipes-container/src/main/java/org/apache/streampipes/container/util/NodeControllerUtil.java b/streampipes-container/src/main/java/org/apache/streampipes/container/util/NodeControllerUtil.java
index 4295c04..9f5f8ed 100644
--- a/streampipes-container/src/main/java/org/apache/streampipes/container/util/NodeControllerUtil.java
+++ b/streampipes-container/src/main/java/org/apache/streampipes/container/util/NodeControllerUtil.java
@@ -23,6 +23,9 @@ import org.apache.streampipes.container.model.EdgeExtensionsConfig;
import org.apache.streampipes.container.model.node.InvocableRegistration;
import org.apache.streampipes.container.model.consul.ConsulServiceRegistrationBody;
import org.apache.streampipes.container.model.consul.HealthCheckConfiguration;
+import org.apache.streampipes.model.base.ConsumableStreamPipesEntity;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,18 +47,18 @@ public class NodeControllerUtil {
private static final String NODE_CONTROLLER_REGISTER_SVC_URL = "api/v2/node/element/register";
// StandaloneModelSubmitter for pipeline-elements-all-jvm
- public static void register(String id, String host, int port, List<String> supportedAppIds) {
+ public static void register(String id, String host, int port, List<ConsumableStreamPipesEntity> supportedEntities) {
String body = createSvcBody(PE_TAG, makeSvcId(host, id), host,
- port, Arrays.asList(PE_TAG, SECONDARY_PE_IDENTIFIER_TAG), supportedAppIds);
+ port, Arrays.asList(PE_TAG, SECONDARY_PE_IDENTIFIER_TAG), supportedEntities);
String endpoint = generateNodeControllerEndpointFromEnv();
registerService(endpoint, body);
}
// ExtensionsModelSubmitter for extensions-all
- public static void register(EdgeExtensionsConfig conf, List<String> supportedAppIds) {
+ public static void register(EdgeExtensionsConfig conf, List<ConsumableStreamPipesEntity> supportedEntities) {
String body = createSvcBody(PE_TAG, makeSvcId(conf.getHost(), conf.getId()), conf.getHost(),
- conf.getPort(), Arrays.asList(PE_TAG, SECONDARY_PE_IDENTIFIER_TAG), supportedAppIds);
+ conf.getPort(), Arrays.asList(PE_TAG, SECONDARY_PE_IDENTIFIER_TAG), supportedEntities);
String endpoint = generateNodeControllerEndpointFromConfig(conf);
registerService(endpoint, body);
@@ -97,7 +100,7 @@ public class NodeControllerUtil {
}
private static String createSvcBody(String name, String id, String host, int port, List<String> tags,
- List<String> supportedAppIds) {
+ List<ConsumableStreamPipesEntity> supportedEntities) {
try {
ConsulServiceRegistrationBody body = new ConsulServiceRegistrationBody();
String healthCheckURL = HTTP_PROTOCOL + host + COLON + port;
@@ -111,7 +114,7 @@ public class NodeControllerUtil {
InvocableRegistration svcBody = new InvocableRegistration();
svcBody.setConsulServiceRegistrationBody(body);
- svcBody.setSupportedPipelineElementAppIds(supportedAppIds);
+ svcBody.setSupportedPipelineElements(supportedEntities);
return JacksonSerializer.getObjectMapper().writeValueAsString(svcBody);
} catch (JsonProcessingException e) {
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java
index 07e896b..f041695 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/OffloadingPolicyManager.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading;
import org.apache.streampipes.node.controller.management.offloading.model.OffloadingStrategy;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/OffloadingStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/OffloadingStrategy.java
index 433362a..8a6218d 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/OffloadingStrategy.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/OffloadingStrategy.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model;
import org.apache.streampipes.node.controller.management.offloading.model.policies.OffloadingPolicy;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java
index 169a755..7beb720 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/CPULoadResourceProperty.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model.property;
import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java
index 9cf4d00..840de8d 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeDiskSpaceResourceProperty.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model.property;
import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java
index fcabeb7..11c5ce3 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/FreeMemoryResourceProperty.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model.property;
import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java
index 3378ea3..9b61663 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/property/ResourceProperty.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model.property;
import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/CPULoadSelectionStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/CPULoadSelectionStrategy.java
index efd055a..1022e64 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/CPULoadSelectionStrategy.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/CPULoadSelectionStrategy.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model.selection;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/RandomSelectionStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/RandomSelectionStrategy.java
index 0dcd551..6f214df 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/RandomSelectionStrategy.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/RandomSelectionStrategy.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model.selection;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/SelectionStrategy.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/SelectionStrategy.java
index 54a1859..23f1931 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/SelectionStrategy.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/offloading/model/selection/SelectionStrategy.java
@@ -1,3 +1,21 @@
+/*
+ * 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.controller.management.offloading.model.selection;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
diff --git a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java
index 401abc7..8f25aa7 100644
--- a/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java
+++ b/streampipes-node-controller/src/main/java/org/apache/streampipes/node/controller/management/pe/InvocableElementManager.java
@@ -29,10 +29,13 @@ import org.apache.streampipes.messaging.jms.ActiveMQPublisher;
import org.apache.streampipes.messaging.kafka.SpKafkaProducer;
import org.apache.streampipes.messaging.mqtt.MqttPublisher;
import org.apache.streampipes.model.Response;
+import org.apache.streampipes.model.base.ConsumableStreamPipesEntity;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.grounding.*;
import org.apache.streampipes.model.node.NodeInfoDescription;
import org.apache.streampipes.model.pipeline.PipelineElementReconfigurationEntity;
+import org.apache.streampipes.model.resource.Hardware;
import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
import org.apache.streampipes.model.staticproperty.StaticProperty;
import org.apache.streampipes.node.controller.config.NodeConfiguration;
@@ -49,6 +52,7 @@ import java.net.Socket;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.util.*;
+import java.util.stream.Collectors;
public class InvocableElementManager implements IPipelineElementLifeCycle {
@@ -167,10 +171,9 @@ public class InvocableElementManager implements IPipelineElementLifeCycle {
try {
String url = generateBackendOffloadEndpoint();
String desc = toJson(instanceToOffload);
- org.apache.http.client.fluent.Response resp = Request.Post(url)
+ Request.Post(url)
.bodyString(desc, ContentType.APPLICATION_JSON)
.execute();
- resp.returnContent();
} catch (IOException e) {
e.printStackTrace();
}
@@ -258,7 +261,8 @@ public class InvocableElementManager implements IPipelineElementLifeCycle {
}
private void updateAndSyncNodeInfoDescription(InvocableRegistration registration) {
- setSupportedPipelineElements(registration.getSupportedPipelineElementAppIds());
+ setSupportedPipelineElements(getSupportedEntities(registration).stream()
+ .map(NamedStreamPipesEntity::getAppId).collect(Collectors.toList()));
try {
String url = generateBackendEndpoint();
String desc = toJson(getNodeInfoDescription());
@@ -270,6 +274,17 @@ public class InvocableElementManager implements IPipelineElementLifeCycle {
}
}
+ private List<ConsumableStreamPipesEntity> getSupportedEntities(InvocableRegistration registration){
+ //Check if the node supports the entity (atm checks if it has a GPU) TODO: simplify; check for other Hardware
+ return registration.getSupportedPipelineElements().stream().filter(nse -> (!nse.getResourceRequirements().stream()
+ .filter(req -> req instanceof Hardware).filter(req -> ((Hardware) req).isGpu()).map(req -> ((Hardware) req)
+ .isGpu()).findFirst().orElse(false) ||
+ nse.getResourceRequirements().stream().filter(req -> req instanceof Hardware)
+ .filter(req -> ((Hardware) req).isGpu()).map(req -> ((Hardware) req)
+ .isGpu()).findFirst().orElse(false) == getNodeInfoDescription()
+ .getNodeResources().getHardwareResource().getGpu().getCores()>0)).collect(Collectors.toList());
+ }
+
private List<InvocableStreamPipesEntity> getAllInvocables() {
return RunningInvocableInstances.INSTANCE.getAll();
}
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 1e3db98..a65039a 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
@@ -19,11 +19,13 @@ package org.apache.streampipes.node.controller.management.resource;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.node.controller.config.NodeControllerConfig;
-import org.apache.streampipes.node.controller.management.offloading.AutoOffloadingManager;
-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.config.NodeConfiguration;
+import org.apache.streampipes.node.controller.management.offloading.OffloadingPolicyManager;
+import org.apache.streampipes.node.controller.management.offloading.model.OffloadingStrategy;
+import org.apache.streampipes.node.controller.management.offloading.model.policies.Comparator;
+import org.apache.streampipes.node.controller.management.offloading.model.policies.ThresholdViolationOffloadingPolicy;
+import org.apache.streampipes.node.controller.management.offloading.model.property.CPULoadResourceProperty;
+import org.apache.streampipes.node.controller.management.offloading.model.selection.RandomSelectionStrategy;
import org.apache.streampipes.node.controller.management.resource.model.ResourceMetrics;
import org.apache.streampipes.node.controller.management.resource.utils.DiskSpace;
import org.apache.streampipes.node.controller.management.resource.utils.ResourceUtils;
@@ -48,14 +50,11 @@ public class ResourceManager {
private final OperatingSystem os = si.getOperatingSystem();
private final Calendar cal = Calendar.getInstance();
private final ResourceMetrics resourceMetrics = new ResourceMetrics();
- // Offloading policy
- private final OffloadingPolicy<Float> multiStepThresholdPolicy = new MultiOccurrenceThresholdViolationPolicy<>(5,
- Comparator.GREATER,99f, 5);
private ResourceManager() {
//Offloading Policy
OffloadingPolicyManager.getInstance().addOffloadingStrategy(new OffloadingStrategy<Float>(new
- ThresholdViolationOffloadingPolicy<>(5, Comparator.GREATER,0.6f, 5),
+ ThresholdViolationOffloadingPolicy<>(5, Comparator.GREATER,60f, 5),
new CPULoadResourceProperty(), new RandomSelectionStrategy()));
}