[Feature][Connector-V2] Add prometheus source and sink

.github/workflows/labeler/label-scope-conf.yml

@@ -132,6 +132,11 @@ http:
       - changed-files:
           - any-glob-to-any-file: seatunnel-connectors-v2/connector-http/**
           - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(http)/**'
+prometheus:
+  - all:
+      - changed-files:
+          - any-glob-to-any-file: seatunnel-connectors-v2/connector-prometheus/**
+          - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(prometheus)/**'
 hudi:
   - all:
       - changed-files:

config/plugin_config

@@ -86,4 +86,5 @@ connector-rocketmq
 connector-tdengine
 connector-web3j
 connector-milvus
---end--
+connector-prometheus
+--end--

docs/en/connector-v2/source/prometheus.md

@@ -0,0 +1,152 @@
+# Prometheus
+
+> Prometheus source connector
+
+## Description
+
+Used to read data from Prometheus.
+
+## Key features
+
+- [x] [batch](../../concept/connector-v2-features.md)
+- [ ] [stream](../../concept/connector-v2-features.md)
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+- [ ] [column projection](../../concept/connector-v2-features.md)
+- [ ] [parallelism](../../concept/connector-v2-features.md)
+- [ ] [support user-defined split](../../concept/connector-v2-features.md)
+
+## Options
+
+|            name             |  type   | required |  default value  |
+|-----------------------------|---------|----------|-----------------|
+| url                         | String  | Yes      | -               |
+| query                       | String  | Yes      | -               |
+| query_type                  | String  | Yes      | Instant         |
+| content_field               | String  | Yes      | $.data.result.* |
+| schema.fields               | Config  | Yes      | -               |
+| format                      | String  | No       | json            |
+| params                      | Map     | Yes      | -               |
+| poll_interval_millis        | int     | No       | -               |
+| retry                       | int     | No       | -               |
+| retry_backoff_multiplier_ms | int     | No       | 100             |
+| retry_backoff_max_ms        | int     | No       | 10000           |
+| enable_multi_lines          | boolean | No       | false           |
+| common-options              | config  | No       | -               |
+
+### url [String]
+
+http request url
+
+### query [String]
+
+Prometheus expression query string
+
+### query_type [String]
+
+Instant/Range
+
+1. Instant : The following endpoint evaluates an instant query at a single point in time
+2. Range : The following endpoint evaluates an expression query over a range of time
+
+https://prometheus.io/docs/prometheus/latest/querying/api/
+
+### params [Map]
+
+http params
+
+### poll_interval_millis [int]
+
+request http api interval(millis) in stream mode
+
+### retry [int]
+
+The max retry times if request http return to `IOException`
+
+### retry_backoff_multiplier_ms [int]
+
+The retry-backoff times(millis) multiplier if request http failed
+
+### retry_backoff_max_ms [int]
+
+The maximum retry-backoff times(millis) if request http failed
+
+### format [String]
+
+the format of upstream data, default `json`.
+
+### schema [Config]
+
+Fill in a fixed value
+
+```hocon
+    schema = {
+        fields {
+            metric = "map<string, string>"
+            value = "array<int>"
+            }
+        }
+
+```
+
+#### fields [Config]
+
+the schema fields of upstream data
+
+### common options
+
+Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details
+
+## Example
+
+### Instant:
+
+```hocon
+source {
+  Prometheus {
+    result_table_name = "http"
+    url = "http://mockserver:1080"
+    query = "up"
+    query_type = "Instant"
+    content_field = "$.data.result.*"
+    format = "json"
+    schema = {
+        fields {
+            metric = "map<string, string>"
+            value = "array<int>"
+            }
+        }
+    }
+}
+```
+
+### Range
+
+```hocon
+source {
+  Prometheus {
+    result_table_name = "http"
+    url = "http://mockserver:1080"
+    query = "up"
+    query_type = "Range"
+    content_field = "$.data.result.*"
+    format = "json"
+    start = "2024-07-22T20:10:30.781Z"
+    end = "2024-07-22T20:11:00.781Z"
+    step = "15s"
+    schema = {
+        fields {
+            metric = "map<string, string>"
+            value = "array<int>"
+            }
+        }
+    }
+  }
+```
+
+## Changelog
+
+### next version
+
+- Add Prometheus Source Connector
+- Reduce configuration items
+

plugin-mapping.properties

@@ -129,3 +129,4 @@ seatunnel.source.ObsFile = connector-file-obs
 seatunnel.sink.ObsFile = connector-file-obs
 seatunnel.source.Milvus = connector-milvus
 seatunnel.sink.Milvus = connector-milvus
+seatunnel.source.Prometheus = connector-prometheus

seatunnel-connectors-v2/connector-prometheus/pom.xml

@@ -0,0 +1,58 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.seatunnel</groupId>
+        <artifactId>seatunnel-connectors-v2</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>connector-prometheus</artifactId>
+    <name>SeaTunnel : Connectors V2 : Prometheus</name>
+
+    <properties>
+        <prometheus-client.version>0.16.0</prometheus-client.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-http-base</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>simpleclient</artifactId>
+            <version>${prometheus-client.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.prometheus</groupId>
+            <artifactId>simpleclient_httpserver</artifactId>
+            <version>${prometheus-client.version}</version>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-connectors-v2/connector-prometheus/src/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/config/PrometheusSourceConfig.java

@@ -0,0 +1,85 @@
+/*
+ * 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.connectors.seatunnel.prometheus.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+import org.apache.seatunnel.connectors.seatunnel.http.config.HttpConfig;
+
+/**
+ * SourceConfig is the configuration for the PrometheusSource.
+ *
+ * <p>please see the following link for more details:
+ * https://prometheus.io/docs/prometheus/latest/querying/api/
+ */
+public class PrometheusSourceConfig extends HttpConfig {
+
+    public static final String INSTANT_QUERY_URL = "/api/v1/query";
+
+    public static final String RANGE_QUERY = "Range";
+
+    public static final String RANGE_QUERY_URL = "/api/v1/query_range";
+
+    public static final Option<String> QUERY =
+            Options.key("query")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("Prometheus expression query string");
+
+    public static final Option<String> QUERY_TYPE =
+            Options.key("query_type")
+                    .stringType()
+                    .defaultValue("Instant")
+                    .withDescription("Prometheus expression query string");
+
+    public static final Option<Long> TIMEOUT =
+            Options.key("timeout")
+                    .longType()
+                    .noDefaultValue()
+                    .withDescription("Evaluation timeout");
+
+    public static class RangeConfig {
+
+        public static final Option<String> START =
+                Options.key("start")
+                        .stringType()
+                        .noDefaultValue()
+                        .withDescription("Start timestamp, inclusive.");
+
+        public static final Option<String> END =
+                Options.key("end")
+                        .stringType()
+                        .noDefaultValue()
+                        .withDescription("End timestamp, inclusive.");
+
+        public static final Option<String> STEP =
+                Options.key("step")
+                        .stringType()
+                        .noDefaultValue()
+                        .withDescription(
+                                " Query resolution step width in duration format or float number of seconds.");
+    }
+
+    public static class InstantQueryConfig {
+        public static final Option<Long> TIME =
+                Options.key("time")
+                        .longType()
+                        .noDefaultValue()
+                        .withDescription("Evaluation timestamp,unix_timestamp");
+    }
+}

seatunnel-connectors-v2/connector-prometheus/src/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/config/PrometheusSourceParameter.java

@@ -0,0 +1,74 @@
+/*
+ * 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.connectors.seatunnel.prometheus.config;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter;
+import org.apache.seatunnel.connectors.seatunnel.http.config.HttpRequestMethod;
+
+import java.util.HashMap;
+
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.INSTANT_QUERY_URL;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.InstantQueryConfig.TIME;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.QUERY;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.QUERY_TYPE;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.RANGE_QUERY;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.RANGE_QUERY_URL;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.RangeConfig.END;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.RangeConfig.START;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.RangeConfig.STEP;
+import static org.apache.seatunnel.connectors.seatunnel.prometheus.config.PrometheusSourceConfig.TIMEOUT;
+
+public class PrometheusSourceParameter extends HttpParameter {
+    public void buildWithConfig(Config pluginConfig) {
+        super.buildWithConfig(pluginConfig);
+
+        String query = pluginConfig.getString(QUERY.key());
+
+        String queryType =
+                pluginConfig.hasPath(QUERY_TYPE.key())
+                        ? pluginConfig.getString(QUERY_TYPE.key())
+                        : QUERY_TYPE.defaultValue();
+
+        this.params = this.getParams() == null ? new HashMap<>() : this.getParams();
+
+        params.put(PrometheusSourceConfig.QUERY.key(), query);
+
+        this.setMethod(HttpRequestMethod.GET);
+
+        if (pluginConfig.hasPath(TIMEOUT.key())) {
+            params.put(TIMEOUT.key(), pluginConfig.getString(TIMEOUT.key()));
+        }
+
+        if (RANGE_QUERY.equals(queryType)) {
+            this.setUrl(this.getUrl() + RANGE_QUERY_URL);
+            params.put(START.key(), pluginConfig.getString(START.key()));
+            params.put(END.key(), pluginConfig.getString(END.key()));
+            params.put(STEP.key(), pluginConfig.getString(STEP.key()));
+
+        } else {
+            this.setUrl(this.getUrl() + INSTANT_QUERY_URL);
+            if (pluginConfig.hasPath(TIME.key())) {
+                String time = pluginConfig.getString(TIME.key());
+                params.put(TIME.key(), time);
+            }
+        }
+        this.setParams(params);
+    }
+}