[Feature][Engine]Metalake support for data source information storage and management (#9688)

Author: wtybxqm

config/seatunnel-env.cmd

@@ -18,4 +18,13 @@ REM Home directory of spark distribution.
 if "%SPARK_HOME%" == "" set "SPARK_HOME=C:\Program Files\spark"
 
 REM Home directory of flink distribution.
-if "%FLINK_HOME%" == "" set "FLINK_HOME=C:\Program Files\flink"
+if "%FLINK_HOME%" == "" set "FLINK_HOME=C:\Program Files\flink"
+
+REM Whether to enable metalake (true/false).
+if "%METALAKE_ENABLED%" == "" set "META_LAKE_ENABLED=false"
+
+REM Type of metalake implementation. 
+if "%METALAKE_TYPE%" == "" set "METALAKE_TYPE=gravitino"
+
+REM Metalake service URL, format: http://host:port/api/metalakes/{metalake_name}/catalogs/
+if "%METALAKE_URL%" == "" set "METALAKE_URL=http://localhost:8090/api/metalakes/default_metalake_name/catalogs/"

config/seatunnel-env.sh

@@ -20,3 +20,9 @@
 SPARK_HOME=${SPARK_HOME:-/opt/spark}
 # Home directory of flink distribution.
 FLINK_HOME=${FLINK_HOME:-/opt/flink}
+# Whether to enable metalake (true/false).
+METALAKE_ENABLED=${METALAKE_ENABLED:-false}
+# Type of metalake implementation.
+METALAKE_TYPE=${METALAKE_TYPE:-gravitino}
+# Metalake service URL, format: http://host:port/api/metalakes/{metalake_name}/catalogs/.
+METALAKE_URL=${METALAKE_URL:-http://localhost:8090/api/metalakes/default_metalake_name/catalogs/}

docs/en/concept/metalake.md

@@ -0,0 +1,69 @@
+# METALAKE
+
+Since Seatunnel requires database usernames, passwords, and other sensitive information to be written in plaintext within scripts when executing tasks, this may lead to information leakage and is also difficult to maintain. When data source information changes, manual modifications are often required.
+
+To address this, Metalake is introduced. Data source information can be stored in Metalake systems such as Apache Gravitino. Task scripts then use `sourceId` and placeholders instead of actual usernames and passwords. At runtime, the Seatunnel engine retrieves the information from Metalake via HTTP requests and replaces the placeholders accordingly.
+
+To enable Metalake, you first need to modify the environment variables in **seatunnel-env.sh**:
+
+* `METALAKE_ENABLED`
+* `METALAKE_TYPE`
+* `METALAKE_URL`
+
+Set `METALAKE_ENABLED` to `true`. Currently, `METALAKE_TYPE` only supports `gravitino`.
+
+For Apache Gravitino, set `METALAKE_URL` to:
+
+```
+http://host:port/api/metalakes/your_metalake_name/catalogs/
+```
+
+---
+
+## Usage Example
+
+First, create a catalog in Gravitino, for example:
+
+```bash
+curl -L 'http://localhost:8090/api/metalakes/test_metalake/catalogs' \
+-H 'Content-Type: application/json' \
+-H 'Accept: application/vnd.gravitino.v1+json' \
+-d '{
+    "name": "test_catalog",
+    "type": "relational",
+    "provider": "jdbc-mysql",
+    "comment": "for metalake test",
+    "properties": {
+        "jdbc-driver": "com.mysql.cj.jdbc.Driver",
+        "jdbc-url": "not used",
+        "jdbc-user": "root",
+        "jdbc-password": "Abc!@#135_seatunnel"
+    }
+}'
+```
+
+This creates a `test_catalog` under `test_metalake` (note: `metalake` itself must be created in advance).
+
+Thus, `METALAKE_URL` can be set to:
+
+```
+http://localhost:8090/api/metalakes/test_metalake/catalogs/
+```
+
+You can then define the source as:
+
+```hocon
+source {
+    Jdbc {
+        url = "jdbc:mysql://mysql-e2e:3306/seatunnel?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true"
+        driver = "${jdbc-driver}"
+        connection_check_timeout_sec = 100
+        sourceId = "test_catalog"
+        user = "${jdbc-user}"
+        password = "${jdbc-password}"
+        query = "select * from source"
+    }
+}
+```
+
+Here, `sourceId` refers to the catalog name, allowing other fields to use `${}` placeholders. At runtime, they will be automatically replaced. Note that in sinks, the same `sourceId` name is used, and placeholders must always start with `${` and end with `}`. Each item can contain at most one placeholder, and there can be content outside the placeholder as well.

docs/sidebars.js

@@ -94,7 +94,8 @@ const sidebars = {
                 'concept/sql-config',
                 'concept/speed-limit',
                 'concept/event-listener',
-                'concept/schema-evolution'
+                'concept/schema-evolution',
+                'concept/metalake'
             ]
         },
         {

docs/zh/concept/metalake.md

@@ -0,0 +1,69 @@
+# METALAKE
+
+由于Seatunnel在执行任务时，需要将数据库用户名与密码等隐私信息明文写在脚本中，可能会导致信息泄露；并且维护较为困难，数据源信息发生变更时可能需要手动更改。
+
+因此引入了metalake，将数据源的信息存储于Apache Gravitino等metalake中，任务脚本采用`sourceId`和占位符的方法来代替原本的用户名和密码等信息，运行时seatunnel-engine通过http请求从metalake获取信息，根据占位符进行替换。
+
+若要使用metalake，首先要修改**seatunnel-env.sh**中的环境变量：
+
+* `METALAKE_ENABLED`
+* `METALAKE_TYPE`
+* `METALAKE_URL`
+
+将`METALAKE_ENABLED`设为`true`，`METALAKE_TYPE`当前仅支持设为`gravitino`。
+
+对于Apache Gravitino，`METALAKE_URL`设为
+
+```
+http://host:port/api/metalakes/your_metalake_name/catalogs/
+```
+
+---
+
+## 使用示例：
+
+用户需要先在Gravitino中创建catalog，如
+
+```bash
+curl -L 'http://localhost:8090/api/metalakes/test_metalake/catalogs'
+-H 'Content-Type: application/json'
+-H 'Accept: application/vnd.gravitino.v1+json'
+-d '{
+    "name": "test_catalog",
+    "type": "relational",
+    "provider": "jdbc-mysql",
+    "comment": "for metalake test",
+    "properties": {
+        "jdbc-driver": "com.mysql.cj.jdbc.Driver",
+        "jdbc-url": "not used",
+        "jdbc-user": "root",
+        "jdbc-password": "Abc!@#135_seatunnel"
+    }
+}'
+```
+
+这样便在`test_metalake`中创建了一个`test_catalog`（`metalake`需要提前创建）
+
+于是`METALAKE_URL`可以设为
+
+```
+http://localhost:8090/api/metalakes/test_metalake/catalogs/
+```
+
+source可以写为
+
+```
+source {
+    Jdbc {
+        url = "jdbc:mysql://mysql-e2e:3306/seatunnel?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true"
+        driver = "${jdbc-driver}"
+        connection_check_timeout_sec = 100
+        sourceId = "test_catalog"
+        user = "${jdbc-user}"
+        password = "${jdbc-password}"
+        query = "select * from source"
+    }
+}
+```
+
+其中`sourceId`指代catalog的名称，从而其他项可以使用`${}`占位符，运行时会自动替换。注意，在sink中使用时，同样叫`sourceId`；使用占位符时必须以`${`开头，以`}`结尾，每一项最多只能包含一个占位符，占位符以外也可以有内容

seatunnel-api/pom.xml

@@ -29,6 +29,11 @@
     <artifactId>seatunnel-api</artifactId>
     <name>SeaTunnel : Api</name>
 
+    <properties>
+        <httpclient.version>4.5.13</httpclient.version>
+        <httpcore.version>4.4.16</httpcore.version>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
@@ -41,5 +46,15 @@
             <version>${project.version}</version>
             <classifier>optional</classifier>
         </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>${httpclient.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpcore</artifactId>
+            <version>${httpcore.version}</version>
+        </dependency>
     </dependencies>
 </project>

seatunnel-api/src/main/java/org/apache/seatunnel/api/metalake/GravitinoClient.java

@@ -0,0 +1,66 @@
+/*
+ * 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.api.metalake;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+
+import org.apache.seatunnel.common.utils.JsonUtils;
+
+import org.apache.http.HttpEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+import java.io.IOException;
+
+public class GravitinoClient implements MetalakeClient {
+    private final String metalakeUrl;
+
+    public GravitinoClient(String metalakeUrl) {
+        this.metalakeUrl = metalakeUrl;
+    }
+
+    @Override
+    public String getType() {
+        return "gravitino";
+    }
+
+    @Override
+    public JsonNode getMetaInfo(String sourceId) throws IOException {
+        try (CloseableHttpClient client = HttpClients.createDefault()) {
+            HttpGet request = new HttpGet(this.metalakeUrl + sourceId);
+            request.addHeader("Accept", "application/vnd.gravitino.v1+json");
+            try (CloseableHttpResponse response = client.execute(request)) {
+                HttpEntity entity = response.getEntity();
+                if (entity == null) {
+                    throw new RuntimeException("No response entity");
+                }
+                JsonNode rootNode = JsonUtils.readTree(entity.getContent());
+                EntityUtils.consume(entity);
+                JsonNode catalogNode = rootNode.get("catalog");
+                if (catalogNode == null) {
+                    throw new RuntimeException("Response JSON has no 'catalog' field");
+                }
+                JsonNode propertiesNode = catalogNode.get("properties");
+                return propertiesNode;
+            }
+        }
+    }
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/metalake/MetalakeClient.java

@@ -0,0 +1,28 @@
+/*
+ * 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.api.metalake;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+
+import java.io.IOException;
+
+public interface MetalakeClient {
+    String getType();
+
+    JsonNode getMetaInfo(String sourceId) throws IOException;
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/metalake/MetalakeClientFactory.java

@@ -0,0 +1,44 @@
+/*
+ * 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.api.metalake;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+public class MetalakeClientFactory {
+    private static final Map<String, Function<String, MetalakeClient>> REGISTRY = new HashMap<>();
+
+    static {
+        register("gravitino", GravitinoClient::new);
+    }
+
+    private MetalakeClientFactory() {}
+
+    public static void register(String type, Function<String, MetalakeClient> constructor) {
+        REGISTRY.put(type.toLowerCase(), constructor);
+    }
+
+    public static MetalakeClient create(String type, String metalakeUrl) {
+        Function<String, MetalakeClient> constructor = REGISTRY.get(type.toLowerCase());
+        if (constructor == null) {
+            throw new IllegalArgumentException("Unknown MetalakeClient type: " + type);
+        }
+        return constructor.apply(metalakeUrl);
+    }
+}