Draft: [AMORO-4034][Improvement]: MaintainerMetrics Unified interface and Report

amoro-ams/src/main/java/org/apache/amoro/server/optimizing/maintainer/DefaultTableMaintainerContext.java

@@ -23,7 +23,6 @@
 import org.apache.amoro.maintainer.OptimizingInfo;
 import org.apache.amoro.maintainer.TableMaintainerContext;
 import org.apache.amoro.server.table.DefaultTableRuntime;
-import org.apache.amoro.server.table.TableOrphanFilesCleaningMetrics;
 import org.apache.amoro.server.utils.HiveLocationUtil;
 import org.apache.amoro.table.MixedTable;
 
@@ -56,18 +55,11 @@ public TableConfiguration getTableConfiguration() {
 
   @Override
   public MaintainerMetrics getMetrics() {
-    TableOrphanFilesCleaningMetrics metrics = tableRuntime.getOrphanFilesCleaningMetrics();
-    return new MaintainerMetrics() {
-      @Override
-      public void recordOrphanDataFilesCleaned(int expected, int cleaned) {
-        metrics.completeOrphanDataFiles(expected, cleaned);
-      }
-
-      @Override
-      public void recordOrphanMetadataFilesCleaned(int expected, int cleaned) {
-        metrics.completeOrphanMetadataFiles(expected, cleaned);
-      }
-    };
+    // Return the full TableMaintainerMetricsImpl directly
+    // This provides access to all maintainer metrics including orphan files cleaning,
+    // dangling delete files cleaning, snapshot expiration, data expiration, tag creation,
+    // and partition expiration.
+    return tableRuntime.getMaintainerMetrics();
   }
 
   @Override

amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMaintainerMetrics.java

@@ -0,0 +1,230 @@
+/*
+ * 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.amoro.server.table;
+
+import static org.apache.amoro.metrics.MetricDefine.defineCounter;
+import static org.apache.amoro.metrics.MetricDefine.defineGauge;
+
+import org.apache.amoro.ServerTableIdentifier;
+import org.apache.amoro.maintainer.MaintainerMetrics;
+import org.apache.amoro.metrics.MetricDefine;
+
+/**
+ * Abstract base class for table maintenance operation metrics
+ *
+ * <p>Responsibilities:
+ *
+ * <ul>
+ *   <li>Define all MetricDefine constants
+ *   <li>Provide template methods for metrics registration
+ *   <li>Handle tags (catalog, database, table, table_format, operation_type)
+ * </ul>
+ *
+ * <p>Note: All metrics include the table_format tag to distinguish Iceberg and Paimon tables
+ */
+public abstract class AbstractTableMaintainerMetrics extends AbstractTableMetrics
+    implements MaintainerMetrics {
+
+  /** Table format constant: Iceberg native table */
+  protected static final String TABLE_FORMAT_ICEBERG = "iceberg";
+
+  /** Table format constant: Paimon native table */
+  protected static final String TABLE_FORMAT_PAIMON = "paimon";
+
+  /** Table format constant: Mixed table (based on Iceberg) */
+  protected static final String TABLE_FORMAT_MIXED_ICEBERG = "mixed_iceberg";
+
+  /** Table format constant: Hive table */
+  protected static final String TABLE_FORMAT_HIVE = "hive";
+
+  // ========== Orphan Files Related MetricDefine ==========
+
+  /** Count of orphan data files cleaned */
+  public static final MetricDefine TABLE_ORPHAN_DATA_FILES_CLEANED_COUNT =
+      defineCounter("table_orphan_data_files_cleaned_count")
+          .withDescription("Count of orphan data files cleaned")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Expected count of orphan data files to clean */
+  public static final MetricDefine TABLE_ORPHAN_DATA_FILES_CLEANED_EXPECTED_COUNT =
+      defineCounter("table_orphan_data_files_cleaned_expected_count")
+          .withDescription("Expected count of orphan data files to clean")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Count of orphan metadata files cleaned */
+  public static final MetricDefine TABLE_ORPHAN_METADATA_FILES_CLEANED_COUNT =
+      defineCounter("table_orphan_metadata_files_cleaned_count")
+          .withDescription("Count of orphan metadata files cleaned")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Expected count of orphan metadata files to clean */
+  public static final MetricDefine TABLE_ORPHAN_METADATA_FILES_CLEANED_EXPECTED_COUNT =
+      defineCounter("table_orphan_metadata_files_cleaned_expected_count")
+          .withDescription("Expected count of orphan metadata files to clean")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Duration of orphan files cleaning operation (milliseconds) */
+  public static final MetricDefine TABLE_ORPHAN_FILES_CLEANING_DURATION =
+      defineGauge("table_orphan_files_cleaning_duration_millis")
+          .withDescription("Duration of orphan files cleaning operation in milliseconds")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  // ========== Dangling Delete Files Related MetricDefine (Iceberg) ==========
+
+  /** Count of dangling delete files cleaned */
+  public static final MetricDefine TABLE_DANGLING_DELETE_FILES_CLEANED_COUNT =
+      defineCounter("table_dangling_delete_files_cleaned_count")
+          .withDescription("Count of dangling delete files cleaned")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Duration of dangling delete files cleaning operation (milliseconds) */
+  public static final MetricDefine TABLE_DANGLING_DELETE_FILES_CLEANING_DURATION =
+      defineGauge("table_dangling_delete_files_cleaning_duration_millis")
+          .withDescription("Duration of dangling delete files cleaning operation in milliseconds")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  // ========== Snapshot Expiration Related MetricDefine ==========
+
+  /** Count of snapshots expired */
+  public static final MetricDefine TABLE_SNAPSHOTS_EXPIRED_COUNT =
+      defineCounter("table_snapshots_expired_count")
+          .withDescription("Count of snapshots expired")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Count of data files deleted during snapshot expiration */
+  public static final MetricDefine TABLE_SNAPSHOTS_EXPIRED_DATA_FILES_DELETED =
+      defineCounter("table_snapshots_expired_data_files_deleted")
+          .withDescription("Count of data files deleted during snapshot expiration")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Duration of snapshot expiration operation (milliseconds) */
+  public static final MetricDefine TABLE_SNAPSHOTS_EXPIRATION_DURATION =
+      defineGauge("table_snapshots_expiration_duration_millis")
+          .withDescription("Duration of snapshot expiration operation in milliseconds")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  // ========== Data Expiration Related MetricDefine (Iceberg) ==========
+
+  /** Count of data files expired */
+  public static final MetricDefine TABLE_DATA_EXPIRED_DATA_FILES_COUNT =
+      defineCounter("table_data_expired_data_files_count")
+          .withDescription("Count of data files expired")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Count of delete files expired */
+  public static final MetricDefine TABLE_DATA_EXPIRED_DELETE_FILES_COUNT =
+      defineCounter("table_data_expired_delete_files_count")
+          .withDescription("Count of delete files expired")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Duration of data expiration operation (milliseconds) */
+  public static final MetricDefine TABLE_DATA_EXPIRATION_DURATION =
+      defineGauge("table_data_expiration_duration_millis")
+          .withDescription("Duration of data expiration operation in milliseconds")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  // ========== Tag Creation Related MetricDefine (Iceberg) ==========
+
+  /** Count of tags created */
+  public static final MetricDefine TABLE_TAGS_CREATED_COUNT =
+      defineCounter("table_tags_created_count")
+          .withDescription("Count of tags created")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Duration of tag creation operation (milliseconds) */
+  public static final MetricDefine TABLE_TAG_CREATION_DURATION =
+      defineGauge("table_tag_creation_duration_millis")
+          .withDescription("Duration of tag creation operation in milliseconds")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  // ========== Partition Expiration Related MetricDefine (Paimon) ==========
+
+  /** Count of partitions expired */
+  public static final MetricDefine TABLE_PARTITIONS_EXPIRED_COUNT =
+      defineCounter("table_partitions_expired_count")
+          .withDescription("Count of partitions expired")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Count of files expired during partition expiration */
+  public static final MetricDefine TABLE_PARTITIONS_EXPIRED_FILES_COUNT =
+      defineCounter("table_partitions_expired_files_count")
+          .withDescription("Count of files expired during partition expiration")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  /** Duration of partition expiration operation (milliseconds) */
+  public static final MetricDefine TABLE_PARTITION_EXPIRATION_DURATION =
+      defineGauge("table_partition_expiration_duration_millis")
+          .withDescription("Duration of partition expiration operation in milliseconds")
+          .withTags("catalog", "database", "table", "table_format")
+          .build();
+
+  // ========== General Operation Status Related MetricDefine ==========
+
+  /** Count of successful maintainer operations */
+  public static final MetricDefine TABLE_MAINTAINER_OPERATION_SUCCESS_COUNT =
+      defineCounter("table_maintainer_operation_success_count")
+          .withDescription("Count of successful maintainer operations")
+          .withTags("catalog", "database", "table", "table_format", "operation_type")
+          .build();
+
+  /** Count of failed maintainer operations */
+  public static final MetricDefine TABLE_MAINTAINER_OPERATION_FAILURE_COUNT =
+      defineCounter("table_maintainer_operation_failure_count")
+          .withDescription("Count of failed maintainer operations")
+          .withTags("catalog", "database", "table", "table_format", "operation_type")
+          .build();
+
+  /** Duration of maintainer operation (milliseconds) */
+  public static final MetricDefine TABLE_MAINTAINER_OPERATION_DURATION =
+      defineGauge("table_maintainer_operation_duration_millis")
+          .withDescription("Duration of maintainer operation in milliseconds")
+          .withTags("catalog", "database", "table", "table_format", "operation_type")
+          .build();
+
+  /** Table format type */
+  protected final String tableFormat;
+
+  /**
+   * Constructor
+   *
+   * @param identifier Table identifier
+   * @param tableFormat Table format type (iceberg/paimon/mixed_iceberg/hive)
+   */
+  protected AbstractTableMaintainerMetrics(ServerTableIdentifier identifier, String tableFormat) {
+    super(identifier);
+    this.tableFormat = tableFormat;
+  }
+}

amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java

@@ -37,6 +37,15 @@ protected AbstractTableMetrics(ServerTableIdentifier identifier) {
     this.identifier = identifier;
   }
 
+  /**
+   * Get the table identifier.
+   *
+   * @return ServerTableIdentifier
+   */
+  public ServerTableIdentifier getIdentifier() {
+    return identifier;
+  }
+
   protected void registerMetric(MetricRegistry registry, MetricDefine define, Metric metric) {
     MetricKey key =
         registry.register(

amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java

@@ -21,6 +21,7 @@
 import org.apache.amoro.Action;
 import org.apache.amoro.AmoroTable;
 import org.apache.amoro.SupportsProcessPlugins;
+import org.apache.amoro.TableFormat;
 import org.apache.amoro.TableRuntime;
 import org.apache.amoro.api.BlockableOperation;
 import org.apache.amoro.config.OptimizingConfig;
@@ -97,7 +98,7 @@ public class DefaultTableRuntime extends AbstractTableRuntime
 
   private final Map<Action, TableProcessContainer> processContainerMap = Maps.newConcurrentMap();
   private final TableOptimizingMetrics optimizingMetrics;
-  private final TableOrphanFilesCleaningMetrics orphanFilesCleaningMetrics;
+  private final TableMaintainerMetricsImpl maintainerMetrics;
   private final TableSummaryMetrics tableSummaryMetrics;
   private volatile long lastPlanTime;
   private volatile OptimizingProcess optimizingProcess;
@@ -107,11 +108,22 @@ public DefaultTableRuntime(TableRuntimeStore store) {
     super(store);
     this.optimizingMetrics =
         new TableOptimizingMetrics(store.getTableIdentifier(), store.getGroupName());
-    this.orphanFilesCleaningMetrics =
-        new TableOrphanFilesCleaningMetrics(store.getTableIdentifier());
+    this.maintainerMetrics =
+        new TableMaintainerMetricsImpl(store.getTableIdentifier(), getTableFormatString(store));
     this.tableSummaryMetrics = new TableSummaryMetrics(store.getTableIdentifier());
   }
 
+  /**
+   * Get the table format string for metrics.
+   *
+   * @param store TableRuntimeStore
+   * @return lowercase format string (iceberg/paimon/mixed_iceberg/hive)
+   */
+  private String getTableFormatString(TableRuntimeStore store) {
+    TableFormat format = store.getTableIdentifier().getFormat();
+    return format.name().toLowerCase();
+  }
+
   public void recover(OptimizingProcess optimizingProcess) {
     if (!getOptimizingStatus().isProcessing()
         || !Objects.equals(optimizingProcess.getProcessId(), getProcessId())) {
@@ -124,7 +136,7 @@ public void recover(OptimizingProcess optimizingProcess) {
   public void registerMetric(MetricRegistry metricRegistry) {
     // TODO: extract method to interface.
     this.optimizingMetrics.register(metricRegistry);
-    this.orphanFilesCleaningMetrics.register(metricRegistry);
+    this.maintainerMetrics.registerMetrics(metricRegistry);
     this.tableSummaryMetrics.register(metricRegistry);
   }
 
@@ -161,8 +173,13 @@ public List<TableProcessStore> getProcessStates(Action action) {
     return processContainerMap.get(action).getProcessStates();
   }
 
-  public TableOrphanFilesCleaningMetrics getOrphanFilesCleaningMetrics() {
-    return orphanFilesCleaningMetrics;
+  /**
+   * Get the maintainer metrics implementation.
+   *
+   * @return TableMaintainerMetricsImpl instance
+   */
+  public TableMaintainerMetricsImpl getMaintainerMetrics() {
+    return maintainerMetrics;
   }
 
   public long getCurrentSnapshotId() {
@@ -472,7 +489,7 @@ public void beginCommitting() {
   @Override
   public void unregisterMetric() {
     tableSummaryMetrics.unregister();
-    orphanFilesCleaningMetrics.unregister();
+    maintainerMetrics.unregister();
     optimizingMetrics.unregister();
   }