[Feature][Zeta] Added other metrics info of multi-table

seatunnel-engine/seatunnel-engine-client/src/test/java/org/apache/seatunnel/engine/client/SeaTunnelClientTest.java

@@ -62,8 +62,12 @@
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_BYTES;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_BYTES_PER_SECONDS;
 import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_COUNT;
 import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_QPS;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES_PER_SECONDS;
 import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_COUNT;
 import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_QPS;
 import static org.awaitility.Awaitility.await;
@@ -592,6 +596,23 @@ public void testGetMultiTableJobMetrics() {
                     jobMetrics.contains(SOURCE_RECEIVED_COUNT + "#fake.public.table2"));
             Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake.table1"));
             Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_COUNT + "#fake.public.table2"));
+            Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_BYTES + "#fake.table1"));
+            Assertions.assertTrue(
+                    jobMetrics.contains(SOURCE_RECEIVED_BYTES + "#fake.public.table2"));
+            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_BYTES + "#fake.table1"));
+            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_BYTES + "#fake.public.table2"));
+            Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_QPS + "#fake.table1"));
+            Assertions.assertTrue(jobMetrics.contains(SOURCE_RECEIVED_QPS + "#fake.public.table2"));
+            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_QPS + "#fake.table1"));
+            Assertions.assertTrue(jobMetrics.contains(SINK_WRITE_QPS + "#fake.public.table2"));
+            Assertions.assertTrue(
+                    jobMetrics.contains(SOURCE_RECEIVED_BYTES_PER_SECONDS + "#fake.table1"));
+            Assertions.assertTrue(
+                    jobMetrics.contains(SOURCE_RECEIVED_BYTES_PER_SECONDS + "#fake.public.table2"));
+            Assertions.assertTrue(
+                    jobMetrics.contains(SINK_WRITE_BYTES_PER_SECONDS + "#fake.table1"));
+            Assertions.assertTrue(
+                    jobMetrics.contains(SINK_WRITE_BYTES_PER_SECONDS + "#fake.public.table2"));
 
             log.info("jobMetrics : {}", jobMetrics);
             JsonNode jobMetricsStr = new ObjectMapper().readTree(jobMetrics);
@@ -600,10 +621,6 @@ public void testGetMultiTableJobMetrics() {
                                     Spliterators.spliteratorUnknownSize(
                                             jobMetricsStr.fieldNames(), 0),
                                     false)
-                            .filter(
-                                    metricName ->
-                                            metricName.startsWith(SOURCE_RECEIVED_COUNT)
-                                                    || metricName.startsWith(SINK_WRITE_COUNT))
                             .collect(Collectors.toList());
 
             Map<String, Long> totalCount =
@@ -654,6 +671,31 @@ public void testGetMultiTableJobMetrics() {
                             .filter(e -> e.getKey().startsWith(SINK_WRITE_COUNT))
                             .mapToLong(Map.Entry::getValue)
                             .sum());
+            Assertions.assertEquals(
+                    totalCount.get(SOURCE_RECEIVED_BYTES),
+                    tableCount.entrySet().stream()
+                            .filter(e -> e.getKey().startsWith(SOURCE_RECEIVED_BYTES + "#"))
+                            .mapToLong(Map.Entry::getValue)
+                            .sum());
+            Assertions.assertEquals(
+                    totalCount.get(SINK_WRITE_BYTES),
+                    tableCount.entrySet().stream()
+                            .filter(e -> e.getKey().startsWith(SINK_WRITE_BYTES + "#"))
+                            .mapToLong(Map.Entry::getValue)
+                            .sum());
+            // Instantaneous rates in the same direction are directly added
+            Assertions.assertEquals(
+                    totalCount.get(SOURCE_RECEIVED_QPS),
+                    tableCount.entrySet().stream()
+                            .filter(e -> e.getKey().startsWith(SOURCE_RECEIVED_QPS + "#"))
+                            .mapToLong(Map.Entry::getValue)
+                            .sum());
+            Assertions.assertEquals(
+                    totalCount.get(SINK_WRITE_QPS),
+                    tableCount.entrySet().stream()
+                            .filter(e -> e.getKey().startsWith(SINK_WRITE_QPS + "#"))
+                            .mapToLong(Map.Entry::getValue)
+                            .sum());
 
         } catch (ExecutionException | InterruptedException | JsonProcessingException e) {
             throw new RuntimeException(e);

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/metrics/TaskMetricsCalcContext.java

@@ -0,0 +1,217 @@
+/*
+ * 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.server.metrics;
+
+import org.apache.seatunnel.api.common.metrics.Counter;
+import org.apache.seatunnel.api.common.metrics.Meter;
+import org.apache.seatunnel.api.common.metrics.MetricsContext;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_BYTES;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_BYTES_PER_SECONDS;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_COUNT;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SINK_WRITE_QPS;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_BYTES_PER_SECONDS;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_COUNT;
+import static org.apache.seatunnel.api.common.metrics.MetricNames.SOURCE_RECEIVED_QPS;
+
+public class TaskMetricsCalcContext {
+
+    private final MetricsContext metricsContext;
+
+    private final String type;
+
+    private Counter count;
+
+    private Map<String, Counter> countPerTable = new ConcurrentHashMap<>();
+
+    private Meter QPS;
+
+    private Map<String, Meter> QPSPerTable = new ConcurrentHashMap<>();
+
+    private Counter bytes;
+
+    private Map<String, Counter> bytesPerTable = new ConcurrentHashMap<>();
+
+    private Meter bytesPerSeconds;
+
+    private Map<String, Meter> bytesPerSecondsPerTable = new ConcurrentHashMap<>();
+
+    public TaskMetricsCalcContext(
+            MetricsContext metricsContext, String type, boolean isMulti, List<TablePath> tables) {
+        this.metricsContext = metricsContext;
+        this.type = type;
+        initializeMetrics(isMulti, tables);
+    }
+
+    private void initializeMetrics(boolean isMulti, List<TablePath> tables) {
+        if ("SINK".equalsIgnoreCase(type)) {
+            this.initializeMetrics(
+                    isMulti,
+                    tables,
+                    SINK_WRITE_COUNT,
+                    SINK_WRITE_QPS,
+                    SINK_WRITE_BYTES,
+                    SINK_WRITE_BYTES_PER_SECONDS);
+        } else if ("SOURCE".equalsIgnoreCase(type)) {
+            this.initializeMetrics(
+                    isMulti,
+                    tables,
+                    SOURCE_RECEIVED_COUNT,
+                    SOURCE_RECEIVED_QPS,
+                    SOURCE_RECEIVED_BYTES,
+                    SOURCE_RECEIVED_BYTES_PER_SECONDS);
+        }
+    }
+
+    private void initializeMetrics(
+            boolean isMulti,
+            List<TablePath> tables,
+            String sinkWriteCount,
+            String sinkWriteQps,
+            String sinkWriteBytes,
+            String sinkWriteBytesPerSeconds) {
+        count = metricsContext.counter(sinkWriteCount);
+        QPS = metricsContext.meter(sinkWriteQps);
+        bytes = metricsContext.counter(sinkWriteBytes);
+        bytesPerSeconds = metricsContext.meter(sinkWriteBytesPerSeconds);
+        if (isMulti) {
+            tables.forEach(
+                    tablePath -> {
+                        countPerTable.put(
+                                tablePath.getFullName(),
+                                metricsContext.counter(
+                                        sinkWriteCount + "#" + tablePath.getFullName()));
+                        QPSPerTable.put(
+                                tablePath.getFullName(),
+                                metricsContext.meter(sinkWriteQps + "#" + tablePath.getFullName()));
+                        bytesPerTable.put(
+                                tablePath.getFullName(),
+                                metricsContext.counter(
+                                        sinkWriteBytes + "#" + tablePath.getFullName()));
+                        bytesPerSecondsPerTable.put(
+                                tablePath.getFullName(),
+                                metricsContext.meter(
+                                        sinkWriteBytesPerSeconds + "#" + tablePath.getFullName()));
+                    });
+        }
+    }
+
+    public void updateMetrics(Object data) {
+        count.inc();
+        QPS.markEvent();
+        if (data instanceof SeaTunnelRow) {
+            SeaTunnelRow row = (SeaTunnelRow) data;
+            bytes.inc(row.getBytesSize());
+            bytesPerSeconds.markEvent(row.getBytesSize());
+            String tableId = row.getTableId();
+
+            if (StringUtils.isNotBlank(tableId)) {
+                String tableName = TablePath.of(tableId).getFullName();
+
+                // Processing count
+                processMetrics(
+                        countPerTable,
+                        Counter.class,
+                        tableName,
+                        SINK_WRITE_COUNT,
+                        SOURCE_RECEIVED_COUNT,
+                        Counter::inc);
+
+                // Processing bytes
+                processMetrics(
+                        bytesPerTable,
+                        Counter.class,
+                        tableName,
+                        SINK_WRITE_BYTES,
+                        SOURCE_RECEIVED_BYTES,
+                        counter -> counter.inc(row.getBytesSize()));
+
+                // Processing QPS
+                processMetrics(
+                        QPSPerTable,
+                        Meter.class,
+                        tableName,
+                        SINK_WRITE_QPS,
+                        SOURCE_RECEIVED_QPS,
+                        Meter::markEvent);
+
+                // Processing bytes rate
+                processMetrics(
+                        bytesPerSecondsPerTable,
+                        Meter.class,
+                        tableName,
+                        SINK_WRITE_BYTES_PER_SECONDS,
+                        SOURCE_RECEIVED_BYTES_PER_SECONDS,
+                        meter -> meter.markEvent(row.getBytesSize()));
+            }
+        }
+    }
+
+    private <T> void processMetrics(
+            Map<String, T> metricMap,
+            Class<T> cls,
+            String tableName,
+            String sinkMetric,
+            String sourceMetric,
+            MetricProcessor<T> processor) {
+        T metric = metricMap.get(tableName);
+        if (Objects.nonNull(metric)) {
+            processor.process(metric);
+        } else {
+            String metricName =
+                    "sink".equalsIgnoreCase(type)
+                            ? sinkMetric + "#" + tableName
+                            : sourceMetric + "#" + tableName;
+            T newMetric = createMetric(metricsContext, metricName, cls);
+            processor.process(newMetric);
+            metricMap.put(tableName, newMetric);
+        }
+    }
+
+    private <T> T createMetric(
+            MetricsContext metricsContext, String metricName, Class<T> metricClass) {
+        if (metricClass == Counter.class) {
+            return metricClass.cast(metricsContext.counter(metricName));
+        } else if (metricClass == Meter.class) {
+            return metricClass.cast(metricsContext.meter(metricName));
+        }
+        throw new IllegalArgumentException("Unsupported metric class: " + metricClass.getName());
+    }
+
+    @FunctionalInterface
+    interface MetricProcessor<T> {
+        void process(T t);
+    }
+
+    private String getFullName(TablePath tablePath) {
+        if (StringUtils.isBlank(tablePath.getTableName())) {
+            tablePath =
+                    TablePath.of(tablePath.getDatabaseName(), tablePath.getSchemaName(), "default");
+        }
+        return tablePath.getFullName();
+    }
+}