[Feature][transform-v2] Support transform metrics

docs/en/seatunnel-engine/rest-api-v1.md

@@ -259,6 +259,7 @@ This API has been deprecated, please use /hazelcast/rest/maps/job-info/:jobId in
     "SinkWriteQPS": "",
     "SinkWriteBytes": "",
     "SinkWriteBytesPerSeconds": "",
+    "TransformOutputCount": {},
     "TableSourceReceivedCount": {},
     "TableSourceReceivedBytes": {},
     "TableSourceReceivedBytesPerSeconds": {},

docs/en/seatunnel-engine/rest-api-v2.md

@@ -165,6 +165,7 @@ seatunnel:
     "SinkWriteQPS": "",
     "SinkWriteBytes": "",
     "SinkWriteBytesPerSeconds": "",
+    "TransformOutputCount": {},
     "TableSourceReceivedCount": {},
     "TableSourceReceivedBytes": {},
     "TableSourceReceivedBytesPerSeconds": {},
@@ -245,6 +246,7 @@ This API has been deprecated, please use /job-info/:jobId instead
     "SinkWriteQPS": "",
     "SinkWriteBytes": "",
     "SinkWriteBytesPerSeconds": "",
+    "TransformOutputCount": {},
     "TableSourceReceivedCount": {},
     "TableSourceReceivedBytes": {},
     "TableSourceReceivedBytesPerSeconds": {},

docs/zh/seatunnel-engine/rest-api-v1.md

@@ -191,6 +191,7 @@ network:
     "SinkWriteQPS": "",
     "SinkWriteBytes": "",
     "SinkWriteBytesPerSeconds": "",
+    "TransformOutputCount": {},
     "TableSourceReceivedCount": {},
     "TableSourceReceivedBytes": {},
     "TableSourceReceivedBytesPerSeconds": {},

docs/zh/seatunnel-engine/rest-api-v2.md

@@ -161,6 +161,7 @@ seatunnel:
     "SinkWriteQPS": "",
     "SinkWriteBytes": "",
     "SinkWriteBytesPerSeconds": "",
+    "TransformOutputCount": {},
     "TableSourceReceivedCount": {},
     "TableSourceReceivedBytes": {},
     "TableSourceReceivedBytesPerSeconds": {},

seatunnel-api/src/main/java/org/apache/seatunnel/api/common/metrics/MetricNames.java

@@ -33,4 +33,5 @@ private MetricNames() {}
     public static final String SINK_WRITE_BYTES = "SinkWriteBytes";
     public static final String SINK_WRITE_QPS = "SinkWriteQPS";
     public static final String SINK_WRITE_BYTES_PER_SECONDS = "SinkWriteBytesPerSeconds";
+    public static final String TRANSFORM_OUTPUT_COUNT = "TransformOutputCount";
 }

seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/TransformExecuteProcessor.java

@@ -20,6 +20,8 @@
 import org.apache.seatunnel.shade.com.typesafe.config.Config;
 
 import org.apache.seatunnel.api.common.JobContext;
+import org.apache.seatunnel.api.common.metrics.MetricNames;
+import org.apache.seatunnel.api.common.metrics.MetricsContext;
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.configuration.util.ConfigValidator;
 import org.apache.seatunnel.api.table.factory.TableTransformFactory;
@@ -32,12 +34,16 @@
 import org.apache.seatunnel.core.starter.execution.PluginUtil;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelFactoryDiscovery;
 import org.apache.seatunnel.plugin.discovery.seatunnel.SeaTunnelTransformPluginDiscovery;
+import org.apache.seatunnel.translation.flink.metric.FlinkMetricContext;
 
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.api.common.functions.RichMapFunction;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
 import org.apache.flink.util.Collector;
 
 import java.net.URL;
@@ -103,24 +109,27 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
                                         LinkedHashMap::new));
 
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+        int index = 0;
         for (int i = 0; i < plugins.size(); i++) {
             try {
                 Config pluginConfig = pluginConfigs.get(i);
+                ReadonlyConfig options = ReadonlyConfig.fromConfig(pluginConfig);
                 DataStreamTableInfo stream =
                         fromSourceTable(pluginConfig, new ArrayList<>(outputTables.values()))
                                 .orElse(input);
                 TableTransformFactory factory = plugins.get(i);
                 TableTransformFactoryContext context =
                         new TableTransformFactoryContext(
-                                stream.getCatalogTables(),
-                                ReadonlyConfig.fromConfig(pluginConfig),
-                                classLoader);
+                                stream.getCatalogTables(), options, classLoader);
                 ConfigValidator.of(context.getOptions()).validate(factory.optionRule());
                 SeaTunnelTransform transform = factory.createTransform(context).createTransform();
 
                 transform.setJobContext(jobContext);
+                String metricName =
+                        String.format("Transform[%s]-%s", transform.getPluginName(), index++);
+                String pluginOutput = options.get(PLUGIN_OUTPUT);
                 DataStream<SeaTunnelRow> inputStream =
-                        flinkTransform(transform, stream.getDataStream());
+                        flinkTransform(transform, stream.getDataStream(), metricName, pluginOutput);
                 String pluginOutputIdentifier =
                         ReadonlyConfig.fromConfig(pluginConfig).get(PLUGIN_OUTPUT);
                 // TODO transform support multi tables
@@ -142,10 +151,14 @@ public List<DataStreamTableInfo> execute(List<DataStreamTableInfo> upstreamDataS
     }
 
     protected DataStream<SeaTunnelRow> flinkTransform(
-            SeaTunnelTransform transform, DataStream<SeaTunnelRow> stream) {
+            SeaTunnelTransform transform,
+            DataStream<SeaTunnelRow> stream,
+            String metricName,
+            String pluginOutput) {
         if (transform instanceof SeaTunnelFlatMapTransform) {
             return stream.flatMap(
-                    new ArrayFlatMap(transform), TypeInformation.of(SeaTunnelRow.class));
+                    new ArrayFlatMap(transform, metricName, pluginOutput),
+                    TypeInformation.of(SeaTunnelRow.class));
         }
 
         return stream.transform(
@@ -155,20 +168,59 @@ protected DataStream<SeaTunnelRow> flinkTransform(
                                 flinkRuntimeEnvironment
                                         .getStreamExecutionEnvironment()
                                         .clean(
-                                                row ->
-                                                        ((SeaTunnelMapTransform<SeaTunnelRow>)
-                                                                        transform)
-                                                                .map(row))))
+                                                new FlinkRichMapFunction(
+                                                        transform, metricName, pluginOutput))))
                 // null value shouldn't be passed to downstream
                 .filter(Objects::nonNull);
     }
 
-    public static class ArrayFlatMap implements FlatMapFunction<SeaTunnelRow, SeaTunnelRow> {
+    public static class FlinkRichMapFunction extends RichMapFunction<SeaTunnelRow, SeaTunnelRow> {
+        private MetricsContext metricsContext;
+        private SeaTunnelTransform transform;
+        private final String metricName;
+        private final String pluginOutput;
+
+        public FlinkRichMapFunction(
+                SeaTunnelTransform transform, String metricName, String pluginOutput) {
+            this.transform = transform;
+            this.metricName = metricName;
+            this.pluginOutput = pluginOutput;
+        }
+
+        @Override
+        public void open(Configuration parameters) throws Exception {
+            metricsContext = new FlinkMetricContext((StreamingRuntimeContext) getRuntimeContext());
+        }
 
+        @Override
+        public SeaTunnelRow map(SeaTunnelRow row) throws Exception {
+            if (Objects.isNull(row)) {
+                return null;
+            }
+            SeaTunnelRow rowResult = ((SeaTunnelMapTransform<SeaTunnelRow>) transform).map(row);
+            if (rowResult != null) {
+                String tableId = pluginOutput == null ? rowResult.getTableId() : pluginOutput;
+                updateMetric(metricName, tableId, metricsContext);
+            }
+            return rowResult;
+        }
+    }
+
+    public static class ArrayFlatMap extends RichFlatMapFunction<SeaTunnelRow, SeaTunnelRow> {
+        private MetricsContext metricsContext;
         private SeaTunnelTransform transform;
+        private final String metricName;
+        private final String pluginOutput;
 
-        public ArrayFlatMap(SeaTunnelTransform transform) {
+        public ArrayFlatMap(SeaTunnelTransform transform, String metricName, String pluginOutput) {
             this.transform = transform;
+            this.metricName = metricName;
+            this.pluginOutput = pluginOutput;
+        }
+
+        @Override
+        public void open(Configuration parameters) throws Exception {
+            metricsContext = new FlinkMetricContext((StreamingRuntimeContext) getRuntimeContext());
         }
 
         @Override
@@ -177,9 +229,25 @@ public void flatMap(SeaTunnelRow row, Collector<SeaTunnelRow> collector) {
                     ((SeaTunnelFlatMapTransform<SeaTunnelRow>) transform).flatMap(row);
             if (CollectionUtils.isNotEmpty(rows)) {
                 for (SeaTunnelRow rowResult : rows) {
+                    String tableId = pluginOutput == null ? rowResult.getTableId() : pluginOutput;
+                    updateMetric(metricName, tableId, metricsContext);
                     collector.collect(rowResult);
                 }
             }
         }
     }
+
+    private static void updateMetric(
+            String metricName, String tableId, MetricsContext metricsContext) {
+        StringBuilder metricNameBuilder = new StringBuilder();
+        metricNameBuilder
+                .append(MetricNames.TRANSFORM_OUTPUT_COUNT)
+                .append("#")
+                .append(metricName)
+                .append("#")
+                .append(tableId);
+        if (metricsContext != null) {
+            metricsContext.counter(metricNameBuilder.toString()).inc();
+        }
+    }
 }

seatunnel-core/seatunnel-starter/src/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java

@@ -43,6 +43,7 @@
 import org.apache.seatunnel.engine.core.job.JobStatus;
 import org.apache.seatunnel.engine.server.SeaTunnelNodeContext;
 
+import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.StringUtils;
 
 import com.hazelcast.client.config.ClientConfig;
@@ -56,6 +57,8 @@
 import java.time.LocalDateTime;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.Random;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -229,7 +232,8 @@ public void execute() throws CommandExecuteException {
         } finally {
             if (jobMetricsSummary != null) {
                 // print job statistics information when job finished
-                log.info(
+                StringBuilder logMessage = new StringBuilder();
+                logMessage.append(
                         StringFormatUtils.formatTable(
                                 "Job Statistic Information",
                                 "Start Time",
@@ -247,6 +251,28 @@ public void execute() throws CommandExecuteException {
                                 "Total Failed Count",
                                 jobMetricsSummary.getSourceReadCount()
                                         - jobMetricsSummary.getSinkWriteCount()));
+
+                if (MapUtils.isNotEmpty(jobMetricsSummary.getTransformMetricsMaps())) {
+                    jobMetricsSummary
+                            .getTransformMetricsMaps()
+                            .forEach(
+                                    (tableName, metrics) -> {
+                                        String[] transformInfos =
+                                                new String[metrics.entrySet().size() * 2 + 2];
+                                        transformInfos[0] = tableName + " Information";
+                                        int index = 0;
+                                        for (Map.Entry<String, Object> entry : metrics.entrySet()) {
+                                            transformInfos[++index] = entry.getKey();
+                                            transformInfos[++index] =
+                                                    String.valueOf(entry.getValue());
+                                        }
+                                        if (Objects.nonNull(transformInfos)) {
+                                            logMessage.append(
+                                                    StringFormatUtils.formatTable(transformInfos));
+                                        }
+                                    });
+                }
+                log.info("{}", logMessage);
             }
             closeClient();
         }

seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/MultiTableMetricsIT.java

@@ -132,7 +132,25 @@ public void multiTableMetrics() {
                                             equalTo(String.valueOf(dataSize * 10)))
                                     .body(
                                             "metrics.TableSinkWriteBytes.'fake.public.table2'",
-                                            equalTo(String.valueOf(dataSize * 5)));
+                                            equalTo(String.valueOf(dataSize * 5)))
+                                    .body(
+                                            "metrics.TransformOutputCount.'Transform[0]-Sql'.'fake3'",
+                                            equalTo("10"))
+                                    .body(
+                                            "metrics.TransformOutputCount.'Transform[1]-Sql'.'fake4'",
+                                            equalTo("10"))
+                                    .body(
+                                            "metrics.TransformOutputCount.'Transform[2]-Sql'.'fake5'",
+                                            equalTo("10"))
+                                    .body(
+                                            "metrics.TransformOutputCount.'Transform[3]-Sql'.'fake6'",
+                                            equalTo("5"))
+                                    .body(
+                                            "metrics.TransformOutputCount.'Transform[4]-Sql'.'fake7'",
+                                            equalTo("5"))
+                                    .body(
+                                            "metrics.TransformOutputCount.'Transform[5]-Sql'.'fake8'",
+                                            equalTo("5"));
                             Assertions.assertTrue(
                                     Double.parseDouble(response.path("metrics.SourceReceivedQPS"))
                                                     > 0
@@ -178,6 +196,35 @@ public void multiTableMetrics() {
                                             && Double.parseDouble(
                                                             response.path(
                                                                     "metrics.TableSinkWriteBytesPerSeconds.'fake.public.table2'"))
+                                                    > 0
+                                            && Double.parseDouble(
+                                                            response.path(
+                                                                    "metrics.TransformOutputCount.'Transform[0]-Sql'.'fake3'"))
+                                                    > 0
+                                            && Double.parseDouble(
+                                                            response.path(
+                                                                    "metrics.TransformOutputCount.'Transform[1]-Sql'"
+                                                                            + ".'fake4'"))
+                                                    > 0
+                                            && Double.parseDouble(
+                                                            response.path(
+                                                                    "metrics.TransformOutputCount.'Transform[2]-Sql'"
+                                                                            + ".'fake5'"))
+                                                    > 0
+                                            && Double.parseDouble(
+                                                            response.path(
+                                                                    "metrics.TransformOutputCount.'Transform[3]-Sql'"
+                                                                            + ".'fake6'"))
+                                                    > 0
+                                            && Double.parseDouble(
+                                                            response.path(
+                                                                    "metrics.TransformOutputCount.'Transform[4]-Sql'"
+                                                                            + ".'fake7'"))
+                                                    > 0
+                                            && Double.parseDouble(
+                                                            response.path(
+                                                                    "metrics.TransformOutputCount.'Transform[5]-Sql'"
+                                                                            + ".'fake8'"))
                                                     > 0);
                         });
     }

seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/batch_fake_multi_table_to_console.conf

@@ -116,13 +116,44 @@ source {
 }
 
 transform {
+  Sql {
+    plugin_input = "fake1"
+    plugin_output = "fake3"
+    query = "select * from fake1"
+  }
+    Sql {
+      plugin_input = "fake3"
+      plugin_output = "fake4"
+      query = "select * from fake3"
+    }
+    Sql {
+      plugin_input = "fake4"
+      plugin_output = "fake5"
+      query = "select * from fake4"
+    }
+  Sql {
+    plugin_input = "fake2"
+    plugin_output = "fake6"
+    query = "select * from fake2"
+  }
+    Sql {
+      plugin_input = "fake6"
+      plugin_output = "fake7"
+      query = "select * from fake6"
+    }
+    Sql {
+      plugin_input = "fake7"
+      plugin_output = "fake8"
+      query = "select * from fake7"
+    }
+
 }
 
 sink {
   console {
-    plugin_input = "fake1"
+    plugin_input = "fake5"
   }
   console {
-    plugin_input = "fake2"
+    plugin_input = "fake8"
   }
-}
+}