[WIP][Feature] yarn application mode support prometheus

streampark-common/src/main/scala/org/apache/streampark/common/conf/CommonConfig.scala

@@ -118,4 +118,17 @@ object CommonConfig {
     description = "Use the database type"
   )
 
+  val STREAMPARK_PROMETHEUS_ENABLE: InternalOption = InternalOption(
+    key = "streampark.prometheus.enable",
+    defaultValue = false,
+    classType = classOf[Boolean],
+    description = "Enable prometheus"
+  )
+
+  val STREAMPARK_PROMGATEWAY_HOST_URL: InternalOption = InternalOption(
+    key = "streampark.promgateway.hostUrl",
+    defaultValue = "http://localhost:9091",
+    classType = classOf[String],
+    description = "Promgateway hostUrl"
+  )
 }

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/resources/prometheus.md

@@ -0,0 +1,90 @@
+
+### 1. linux install pushgateway
+
+#### 1.1. Download pushgateway
+
+```shell
+wget https://github.com/prometheus/pushgateway/releases/download/v1.6.2/pushgateway-1.6.2.linux-amd64.tar.gz
+
+tar -zxvf pushgateway-1.6.2.linux-amd64.tar.gz
+
+mv pushgateway-1.6.2.linux-amd64 pushgateway
+```
+
+#### 1.2. Start pushgateway
+
+```shell
+cd pushgateway
+
+nohup ./pushgateway --web.listen-address :9091 > ./pushgateway.log 2>&1 &
+```
+
+### 2. linux install prometheus
+
+#### 2.1. Download prometheus
+
+```shell
+wget https://github.com/prometheus/prometheus/releases/download/v2.47.1/prometheus-2.47.1.linux-amd64.tar.gz
+
+tar -zxvf prometheus-2.47.1.linux-amd64.tar.gz
+
+mv prometheus-2.47.1.linux-amd64 prometheus
+```
+
+#### 2.2. Modified prometheus.yml, added
+
+```text
+  - job_name: "pushgateway"
+    static_configs:
+      - targets: ["ip:9091"]
+        labels:
+          instance: pushgateway
+```
+
+#### 2.3. Start prometheus
+
+```shell
+cd prometheus
+
+nohup ./prometheus --config.file=prometheus.yml > ./prometheus.log 2>&1 &
+```
+
+### 3. Modify the streampark configuration
+
+```text
+org.apache.streampark.common.conf.CommonConfig.STREAMPARK_PROMETHEUS_ENABLE
+streampark.prometheus.enable: true
+
+org.apache.streampark.common.conf.CommonConfig.STREAMPARK_PROMGATEWAY_HOST_URL
+streampark.promgateway.hostUrl: http://ip:9091
+```
+Current only yarn-application is supported. Other modes have not been tested
+
+#### 4. Stop
+
+#### 4.1. Stop prometheus
+
+```shell
+pgrep prometheus
+
+kill {pid}
+```
+
+#### 4.2. Stop pushgateway
+
+```shell
+pgrep pushgateway
+
+kill {pid}
+```
+
+#### 5. Reference document
+
+```text
+https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/deployment/metric_reporters/#prometheuspushgateway
+
+https://prometheus.io/docs/prometheus/latest/getting_started/
+```
+
+
+

streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala

@@ -17,7 +17,7 @@
 
 package org.apache.streampark.flink.client.`trait`
 
-import org.apache.streampark.common.conf.{ConfigConst, Workspace}
+import org.apache.streampark.common.conf.{CommonConfig, ConfigConst, InternalConfigHolder, Workspace}
 import org.apache.streampark.common.conf.ConfigConst._
 import org.apache.streampark.common.enums.{ApplicationType, FlinkDevelopmentMode, FlinkExecutionMode, FlinkRestoreMode}
 import org.apache.streampark.common.fs.FsOperator
@@ -156,6 +156,27 @@ trait FlinkClientTrait extends Logger {
           })
     }
 
+    // set prometheus
+    val enablePrometheus: Boolean =
+      InternalConfigHolder.get(CommonConfig.STREAMPARK_PROMETHEUS_ENABLE)
+    val promgatewayHostUrl: String =
+      InternalConfigHolder.get(CommonConfig.STREAMPARK_PROMGATEWAY_HOST_URL)
+
+    if (
+      submitRequest.executionMode == FlinkExecutionMode.YARN_APPLICATION && enablePrometheus && StringUtils
+        .isNotBlank(promgatewayHostUrl)
+    ) {
+      flinkConfig.setString(
+        "metrics.reporter.promgateway.factory.class",
+        "org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory")
+      flinkConfig.setString("metrics.reporter.promgateway.hostUrl", promgatewayHostUrl)
+      flinkConfig.setString("metrics.reporter.promgateway.jobName", submitRequest.appName)
+      flinkConfig.setString("metrics.reporter.promgateway.randomJobNameSuffix", "true")
+      flinkConfig.setString("metrics.reporter.promgateway.deleteOnShutdown", "false")
+      flinkConfig.setString("metrics.reporter.promgateway.groupingKey", "k1=v1;k2=v2")
+      flinkConfig.setString("metrics.reporter.promgateway.interval", "10 SECONDS")
+    }
+
     setConfig(submitRequest, flinkConfig)
 
     doSubmit(submitRequest, flinkConfig)