[Fix][Zeta] Fix job detail metrics NaN with 2.3.13 backend

seatunnel-engine/seatunnel-engine-ui/src/tests/detail-metrics.spec.ts

@@ -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.
+ */
+
+import { describe, expect, test } from 'vitest'
+import {
+  collectVertexMetrics,
+  extractVertexIdentifier,
+  readVertexMetricValue
+} from '@/views/jobs/detail-metrics'
+import type { Vertex } from '@/service/job/types'
+
+describe('detail metrics helpers', () => {
+  const sourceVertex = {
+    vertexId: 1,
+    type: 'source',
+    vertexName: 'pipeline-1 [Source[0]]',
+    tablePaths: ['fake.user_table']
+  } as Vertex
+
+  const sinkVertex = {
+    vertexId: 2,
+    type: 'sink',
+    vertexName: 'pipeline-1 [Sink[1]]',
+    tablePaths: ['fake.user_table']
+  } as Vertex
+
+  test('extracts the vertex identifier from the vertex name', () => {
+    expect(extractVertexIdentifier(sourceVertex.vertexName)).toBe('Source[0]')
+    expect(extractVertexIdentifier(sinkVertex.vertexName)).toBe('Sink[1]')
+  })
+
+  test('prefers 2.3.13 prefixed metric keys over raw table names', () => {
+    const metricMap = {
+      'Source[0].fake.user_table': '10',
+      'Source[1].fake.user_table': '20',
+      'fake.user_table': '999'
+    }
+
+    expect(readVertexMetricValue(metricMap, sourceVertex, 'fake.user_table')).toBe(10)
+  })
+
+  test('falls back to raw table names when prefixed keys are unavailable', () => {
+    const metricMap = {
+      'fake.user_table': '15'
+    }
+
+    expect(readVertexMetricValue(metricMap, sourceVertex, 'fake.user_table')).toBe(15)
+  })
+
+  test('collects only metrics that belong to the focused vertex', () => {
+    const metricMap = {
+      'Sink[0].fake.user_table': '5',
+      'Sink[1].fake.user_table': '8'
+    }
+
+    expect(collectVertexMetrics('TableSinkWriteCount', metricMap, sinkVertex)).toEqual({
+      'TableSinkWriteCount.fake.user_table': '8'
+    })
+  })
+})

seatunnel-engine/seatunnel-engine-ui/src/views/jobs/detail-metrics.ts

@@ -0,0 +1,100 @@
+/*
+ * 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.
+ */
+
+import type { Vertex } from '@/service/job/types'
+
+const VERTEX_IDENTIFIER_PATTERN = /((?:Sink|Source|Transform)\[\d+\])/
+
+type MetricMap = Record<string, string> | undefined
+type MetricVertex = Pick<Vertex, 'vertexName' | 'tablePaths'>
+
+export const extractVertexIdentifier = (vertexName?: string): string | undefined => {
+  return vertexName?.match(VERTEX_IDENTIFIER_PATTERN)?.[1]
+}
+
+const resolveMetricKey = (
+  metricMap: MetricMap,
+  vertex: MetricVertex,
+  path: string
+): string | undefined => {
+  if (!metricMap) {
+    return undefined
+  }
+
+  const identifier = extractVertexIdentifier(vertex.vertexName)
+  if (identifier) {
+    const prefixedKey = `${identifier}.${path}`
+    if (metricMap[prefixedKey] !== undefined) {
+      return prefixedKey
+    }
+  }
+
+  if (metricMap[path] !== undefined) {
+    return path
+  }
+
+  const suffix = `.${path}`
+  const suffixedKeys = Object.keys(metricMap).filter((key) => key.endsWith(suffix))
+
+  if (identifier) {
+    const sameVertexKey = suffixedKeys.find((key) => key.startsWith(`${identifier}.`))
+    if (sameVertexKey) {
+      return sameVertexKey
+    }
+  }
+
+  return suffixedKeys.length === 1 ? suffixedKeys[0] : undefined
+}
+
+export const readVertexMetricValue = (
+  metricMap: MetricMap,
+  vertex: MetricVertex,
+  path: string
+): number => {
+  if (!metricMap) {
+    return 0
+  }
+
+  const metricKey = resolveMetricKey(metricMap, vertex, path)
+  if (!metricKey) {
+    return 0
+  }
+
+  const value = Number(metricMap[metricKey])
+  return Number.isFinite(value) ? value : 0
+}
+
+export const collectVertexMetrics = (
+  metricName: string,
+  metricMap: MetricMap,
+  vertex: MetricVertex
+): Record<string, string> => {
+  const metrics: Record<string, string> = {}
+
+  if (!metricMap) {
+    return metrics
+  }
+
+  vertex.tablePaths.forEach((path) => {
+    const metricKey = resolveMetricKey(metricMap, vertex, path)
+    if (metricKey !== undefined) {
+      metrics[`${metricName}.${path}`] = metricMap[metricKey]
+    }
+  })
+
+  return metrics
+}

seatunnel-engine/seatunnel-engine-ui/src/views/jobs/detail.tsx

@@ -25,7 +25,7 @@ import {
   NDrawer,
   NDrawerContent
 } from 'naive-ui'
-import {computed, defineComponent, onUnmounted, reactive, ref, watch} from 'vue'
+import { computed, defineComponent, onUnmounted, reactive, ref, watch } from 'vue'
 import { getJobInfo } from '@/service/job'
 import { useRoute } from 'vue-router'
 import type { Job, Vertex } from '@/service/job/types'
@@ -37,6 +37,7 @@ import { getColorFromStatus } from '@/utils/getTypeFromStatus'
 import './detail.scss'
 import Configuration from '@/components/configuration'
 import JobLog from '@/components/job-log'
+import { collectVertexMetrics, readVertexMetricValue } from './detail-metrics'
 
 export default defineComponent({
   setup() {
@@ -81,7 +82,7 @@ export default defineComponent({
     })
 
     const isTerminalState = (status: string) => {
-      return ['FINISHED', 'FAILED', 'CANCELED','SAVEPOINT_DONE'].includes(status)
+      return ['FINISHED', 'FAILED', 'CANCELED', 'SAVEPOINT_DONE'].includes(status)
     }
 
     const isRunningState = (status: string) => {
@@ -100,7 +101,10 @@ export default defineComponent({
         | 'TableSourceReceivedBytesPerSeconds'
     ) => {
       if (row.type === 'source') {
-        return row.tablePaths.reduce((s, path) => s + Number(job.metrics?.[key][path]), 0)
+        return row.tablePaths.reduce(
+          (s, path) => s + readVertexMetricValue(job.metrics?.[key], row, path),
+          0
+        )
       }
       return 0
     }
@@ -113,7 +117,10 @@ export default defineComponent({
         | 'TableSinkWriteBytesPerSeconds'
     ) => {
       if (row.type === 'sink') {
-        return row.tablePaths.reduce((s, path) => s + Number(job.metrics?.[key][path]), 0)
+        return row.tablePaths.reduce(
+          (s, path) => s + readVertexMetricValue(job.metrics?.[key], row, path),
+          0
+        )
       }
       return 0
     }
@@ -182,34 +189,42 @@ export default defineComponent({
       const vertex = job.jobDag?.vertexInfoMap?.find((v) => v.vertexId === focusedId.value)
       const metrics = {} as any
       if (vertex?.type === 'source') {
-        Object.keys(job.metrics?.TableSourceReceivedBytes || {}).forEach((key) => {
-          metrics[`TableSourceReceivedBytes.${key}`] = job.metrics?.TableSourceReceivedBytes[key]
-        })
-        Object.keys(job.metrics?.TableSourceReceivedCount || {}).forEach((key) => {
-          metrics[`TableSourceReceivedCount.${key}`] = job.metrics?.TableSourceReceivedCount[key]
-        })
-        Object.keys(job.metrics?.TableSourceReceivedQPS || {}).forEach((key) => {
-          metrics[`TableSourceReceivedQPS.${key}`] = job.metrics?.TableSourceReceivedQPS[key]
-        })
-        Object.keys(job.metrics?.TableSourceReceivedBytesPerSeconds || {}).forEach((key) => {
-          metrics[`TableSourceReceivedBytesPerSeconds.${key}`] =
-            job.metrics?.TableSourceReceivedBytesPerSeconds[key]
-        })
+        Object.assign(
+          metrics,
+          collectVertexMetrics(
+            'TableSourceReceivedBytes',
+            job.metrics?.TableSourceReceivedBytes,
+            vertex
+          ),
+          collectVertexMetrics(
+            'TableSourceReceivedCount',
+            job.metrics?.TableSourceReceivedCount,
+            vertex
+          ),
+          collectVertexMetrics(
+            'TableSourceReceivedQPS',
+            job.metrics?.TableSourceReceivedQPS,
+            vertex
+          ),
+          collectVertexMetrics(
+            'TableSourceReceivedBytesPerSeconds',
+            job.metrics?.TableSourceReceivedBytesPerSeconds,
+            vertex
+          )
+        )
       }
       if (vertex?.type === 'sink') {
-        Object.keys(job.metrics?.TableSinkWriteBytes || {}).forEach((key) => {
-          metrics[`TableSinkWriteBytes.${key}`] = job.metrics?.TableSinkWriteBytes[key]
-        })
-        Object.keys(job.metrics?.TableSinkWriteCount || {}).forEach((key) => {
-          metrics[`TableSinkWriteCount.${key}`] = job.metrics?.TableSinkWriteCount[key]
-        })
-        Object.keys(job.metrics?.TableSinkWriteQPS || {}).forEach((key) => {
-          metrics[`TableSinkWriteQPS.${key}`] = job.metrics?.TableSinkWriteQPS[key]
-        })
-        Object.keys(job.metrics?.TableSinkWriteBytesPerSeconds || {}).forEach((key) => {
-          metrics[`TableSinkWriteBytesPerSeconds.${key}`] =
-            job.metrics?.TableSinkWriteBytesPerSeconds[key]
-        })
+        Object.assign(
+          metrics,
+          collectVertexMetrics('TableSinkWriteBytes', job.metrics?.TableSinkWriteBytes, vertex),
+          collectVertexMetrics('TableSinkWriteCount', job.metrics?.TableSinkWriteCount, vertex),
+          collectVertexMetrics('TableSinkWriteQPS', job.metrics?.TableSinkWriteQPS, vertex),
+          collectVertexMetrics(
+            'TableSinkWriteBytesPerSeconds',
+            job.metrics?.TableSinkWriteBytesPerSeconds,
+            vertex
+          )
+        )
       }
       return Object.assign({}, vertex, metrics)
     })