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