[Feature][Zeta] Add tag to node used to filter worker when submit job (apache#7045)

Author: liunaijie

README.md

@@ -61,7 +61,7 @@ SeaTunnel addresses common data integration challenges:
 
 ## SeaTunnel Workflow
 
-![SeaTunnel Workflow](docs/en/images/architecture_diagram.png)
+![SeaTunnel Workflow](docs/images/architecture_diagram.png)
 
 Configure jobs, select execution engines, and parallelize data using Source Connectors. Easily develop and extend connectors to meet your needs.
 

docs/en/about.md

@@ -34,7 +34,7 @@ SeaTunnel focuses on data integration and data synchronization, and is mainly de
 
 ## SeaTunnel work flowchart
 
-![SeaTunnel work flowchart](images/architecture_diagram.png)
+![SeaTunnel work flowchart](../images/architecture_diagram.png)
 
 The runtime process of SeaTunnel is shown in the figure above.
 

docs/en/faq.md

@@ -65,9 +65,9 @@ Refer to: [lightbend/config#456](https://github.com/lightbend/config/issues/456)
 
 Of course! See the screenshot below:
 
-![workflow.png](images/workflow.png)
+![workflow.png](../images/workflow.png)
 
-![azkaban.png](images/azkaban.png)
+![azkaban.png](../images/azkaban.png)
 
 ## Does SeaTunnel have a case for configuring multiple sources, such as configuring elasticsearch and hdfs in source at the same time?
 
@@ -184,7 +184,7 @@ The following conclusions can be drawn:
 
 3. In general, both M and N are determined, and the conclusion can be drawn from 2: The size of `spark.streaming.kafka.maxRatePerPartition` is positively correlated with the size of `spark.executor.cores` * `spark.executor.instances`, and it can be increased while increasing the resource `maxRatePerPartition` to speed up consumption.
 
-![kafka](images/kafka.png)
+![kafka](../images/kafka.png)
 
 ## How can I solve the Error `Exception in thread "main" java.lang.NoSuchFieldError: INSTANCE`?
 

docs/en/seatunnel-engine/resource-isolation.md

@@ -0,0 +1,83 @@
+---
+
+sidebar_position: 9
+-------------------
+
+After version 2.3.6. SeaTunnel can add `tag` to each worker node, when you submit job you can use `tag_filter` to filter the node you want run this job.
+
+# How to archive this:
+
+1. update the config in `hazelcast.yaml`,
+
+```yaml
+hazelcast:
+  cluster-name: seatunnel
+  network:
+    rest-api:
+      enabled: true
+      endpoint-groups:
+        CLUSTER_WRITE:
+          enabled: true
+        DATA:
+          enabled: true
+    join:
+      tcp-ip:
+        enabled: true
+        member-list:
+          - localhost
+    port:
+      auto-increment: false
+      port: 5801
+  properties:
+    hazelcast.invocation.max.retry.count: 20
+    hazelcast.tcp.join.port.try.count: 30
+    hazelcast.logging.type: log4j2
+    hazelcast.operation.generic.thread.count: 50
+  member-attributes:
+    group:
+      type: string
+      value: platform
+    team:
+      type: string
+      value: team1
+```
+
+In this config, we specify the tag by `member-attributes`, the node has `group=platform, team=team1` tags.
+
+2. add `tag_filter` to your job config
+
+```hacon
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+  tag_filter {
+    group = "platform"
+    team = "team1"
+  }
+}
+source {
+  FakeSource {
+    result_table_name = "fake"
+    parallelism = 1
+    schema = {
+      fields {
+        name = "string"
+      }
+    }
+  }
+}
+transform {
+}
+sink {
+  console {
+    source_table_name="fake"
+  }
+}
+```
+
+**Notice:**
+- If not set `tag_filter` in job config, it will random choose the node in all active nodes.
+- When you add multiple tag in `tag_filter`, it need all key exist and value match. if all node not match, you will get `NoEnoughResourceException` exception.
+
+![img.png](../../images/resource-isolation.png)
+

docs/en/images/architecture_diagram.png docs/images/architecture_diagram.png

@@ -178,7 +178,8 @@ const sidebars = {
                 "seatunnel-engine/checkpoint-storage",
                 "seatunnel-engine/rest-api",
                 "seatunnel-engine/tcp",
-                "seatunnel-engine/engine-jar-storage-mode"
+                "seatunnel-engine/engine-jar-storage-mode",
+                "seatunnel-engine/resource-isolation",
             ]
         },
         {

docs/en/images/azkaban.png docs/images/azkaban.png

@@ -32,7 +32,7 @@ SeaTunnel专注于数据集成和数据同步，主要旨在解决数据集成
 
 ## SeaTunnel work flowchart
 
-![SeaTunnel work flowchart](images/architecture_diagram.png)
+![SeaTunnel work flowchart](../images/architecture_diagram.png)
 
 SeaTunnel的运行流程如上图所示。
 

docs/en/images/checkstyle.png docs/images/checkstyle.png

@@ -65,9 +65,9 @@ your string 1
 
 当然！ 请参阅下面的屏幕截图：
 
-![工作流程.png](images/workflow.png)
+![工作流程.png](../images/workflow.png)
 
-![azkaban.png](images/azkaban.png)
+![azkaban.png](../images/azkaban.png)
 
 ## SeaTunnel是否有配置多个源的情况，例如同时在源中配置elasticsearch和hdfs？
 
@@ -185,7 +185,7 @@ sink {
 
 3、一般来说，M和N都确定了，从2可以得出结论：`spark.streaming.kafka.maxRatePerPartition`的大小与`spark.executor.cores` * `spark的大小正相关 .executor.instances`，可以在增加资源`maxRatePerPartition`的同时增加，以加快消耗。
 
-![kafka](images/kafka.png)
+![kafka](../images/kafka.png)
 
 ## 如何解决错误 `Exception in thread "main" java.lang.NoSuchFieldError: INSTANCE`？
 

docs/en/images/kafka.png docs/images/kafka.png

@@ -0,0 +1,83 @@
+---
+
+sidebar_position: 9
+-------------------
+
+在2.3.6版本之后, SeaTunnel支持对每个实例添加`tag`, 然后在提交任务时可以在配置文件中使用`tag_filter`来选择任务将要运行的节点.
+
+# 如何实现改功能
+
+1. 更新`hazelcast.yaml`文件
+
+```yaml
+hazelcast:
+  cluster-name: seatunnel
+  network:
+    rest-api:
+      enabled: true
+      endpoint-groups:
+        CLUSTER_WRITE:
+          enabled: true
+        DATA:
+          enabled: true
+    join:
+      tcp-ip:
+        enabled: true
+        member-list:
+          - localhost
+    port:
+      auto-increment: false
+      port: 5801
+  properties:
+    hazelcast.invocation.max.retry.count: 20
+    hazelcast.tcp.join.port.try.count: 30
+    hazelcast.logging.type: log4j2
+    hazelcast.operation.generic.thread.count: 50
+  member-attributes:
+    group:
+      type: string
+      value: platform
+    team:
+      type: string
+      value: team1
+```
+
+在这个配置中, 我们通过`member-attributes`设置了`group=platform, team=team1`这样两个`tag`
+
+2. 在任务的配置中添加`tag_filter`来选择你需要运行该任务的节点
+
+```hacon
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+  tag_filter {
+    group = "platform"
+    team = "team1"
+  }
+}
+source {
+  FakeSource {
+    result_table_name = "fake"
+    parallelism = 1
+    schema = {
+      fields {
+        name = "string"
+      }
+    }
+  }
+}
+transform {
+}
+sink {
+  console {
+    source_table_name="fake"
+  }
+}
+```
+
+**注意:**
+- 当在任务的配置中, 没有添加`tag_filter`时, 会从所有节点中随机选择节点来运行任务.
+- 当`tag_filter`中存在多个过滤条件时, 会根据key存在以及value相等的全部匹配的节点, 当没有找到匹配的节点时, 会抛出 `NoEnoughResourceException`异常.
+
+![img.png](../../images/resource-isolation.png)
+

docs/images/resource-isolation.png

@@ -69,8 +69,8 @@ public <T> T get(Option<T> option) {
     /**
      * Transform to Config todo: This method should be removed after we remove Config
      *
-     * @deprecated Please use ReadonlyConfig directly
      * @return Config
+     * @deprecated Please use ReadonlyConfig directly
      */
     @Deprecated
     public Config toConfig() {
@@ -96,6 +96,10 @@ public void toMap(Map<String, String> result) {
         }
     }
 
+    public Map<String, Object> getSourceMap() {
+        return confData;
+    }
+
     public <T> Optional<T> getOptional(Option<T> option) {
         if (option == null) {
             throw new NullPointerException("Option not be null.");

docs/en/images/seatunnel-workflow.svg docs/images/seatunnel-workflow.svg

@@ -94,4 +94,10 @@ public interface EnvCommonOptions {
                     .mapType()
                     .noDefaultValue()
                     .withDescription("custom parameters for run engine");
+
+    Option<Map<String, String>> NODE_TAG_FILTER =
+            Options.key("tag_filter")
+                    .mapType()
+                    .noDefaultValue()
+                    .withDescription("Define the worker where the job runs by tag");
 }

docs/en/images/seatunnel_architecture.png docs/images/seatunnel_architecture.png

@@ -1,7 +1,7 @@
 # Introduction
 
 This module is the seatunnel job entrypoint. SeaTunnel jobs are started by the below process.
-![seatunnel-workflow.svg](../docs/en/images/seatunnel_starter.png)
+![seatunnel-workflow.svg](../docs/images/seatunnel_starter.png)
 
 - seatunnel-core-flink: The flink job starter.
 - seatunnel-core-flink-sql: The flink sql job starter.

docs/en/images/seatunnel_starter.png docs/images/seatunnel_starter.png

@@ -2,7 +2,7 @@
 
 This module is the base start module for SeaTunnel new connector API.
 
-![seatunnel_architecture.png](../../docs/en/images/seatunnel_architecture.png)
+![seatunnel_architecture.png](../../docs/images/seatunnel_architecture.png)
 
 # SeaTunnel Job Execute Process
 

docs/en/images/workflow.png docs/images/workflow.png

@@ -0,0 +1,64 @@
+/*
+ * 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.seatunnel.engine.e2e.resourceIsolation;
+
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+
+import org.apache.commons.lang3.StringUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+
+import java.io.IOException;
+
+public class ResourceIsolationIT extends TestSuiteBase {
+
+    @TestTemplate
+    @DisabledOnContainer(
+            value = {},
+            type = {EngineType.SPARK, EngineType.FLINK},
+            disabledReason = "only work on Zeta")
+    public void testTagMatch(TestContainer container) throws IOException, InterruptedException {
+        Container.ExecResult execResult =
+                container.executeJob("/resource-isolation/fakesource_to_console.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());
+    }
+
+    @TestTemplate
+    @DisabledOnContainer(
+            value = {},
+            type = {EngineType.SPARK, EngineType.FLINK},
+            disabledReason = "only work on Zeta")
+    public void testTagNotMatch(TestContainer container) throws IOException, InterruptedException {
+        Container.ExecResult execResult =
+                container.executeJob(
+                        "/resource-isolation/fakesource_to_console_tag_not_match.conf");
+        Assertions.assertNotEquals(0, execResult.getExitCode());
+        Assertions.assertTrue(
+                StringUtils.isNotBlank(execResult.getStderr())
+                        && execResult
+                                .getStderr()
+                                .contains(
+                                        "org.apache.seatunnel.engine.server.resourcemanager.NoEnoughResourceException"));
+    }
+}

docs/sidebars.js

@@ -41,3 +41,10 @@ hazelcast:
     hazelcast.slow.operation.detector.stacktrace.logging.enabled: true
     hazelcast.logging.type: log4j2
     hazelcast.operation.generic.thread.count: 200
+  member-attributes:
+    group:
+      type: string
+      value: platform
+    team:
+      type: string
+      value: team1