[GOBBLIN-2269] Surface snapshot/partition copy metrics in observability

gobblin-metrics-libs/gobblin-metrics-base/src/main/avro/GaaSJobObservabilityEvent.avsc

@@ -264,6 +264,18 @@
                 "type": ["null","string"],
                 "doc": "Whether the dataset passed the overall data quality check",
                 "default": null
+              },
+              {
+                "name": "snapshotsCommitted",
+                "type": ["null","string"],
+                "doc": "Comma-separated list of the ids of the table snapshots committed for the dataset in this run (e.g. Iceberg snapshot-replication commits), null if unsupported/unreported by the writer",
+                "default": null
+              },
+              {
+                "name": "partitionsCommitted",
+                "type": ["null","string"],
+                "doc": "Comma-separated list of the partitions committed for the dataset in this run, null if the writer does not commit partitions or does not report them",
+                "default": null
               }
             ]
           }

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/pegasus/org/apache/gobblin/service/JobStatistics.pdl

@@ -29,4 +29,35 @@ record JobStatistics {
    * estimate of time left until job completion
    */
   estimatedSecondsToCompletion: long
+
+  /**
+   * Number of bytes written/copied for the dataset in this job execution.
+   * Optional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported.
+   */
+  bytesWritten: optional long
+
+  /**
+   * Number of records (or files, for file/blob copies) written for the dataset in this job execution.
+   * Optional; -1 if unsupported by the writer.
+   */
+  recordsWritten: optional long
+
+  /**
+   * Number of files committed for the dataset in this job execution.
+   * Optional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise.
+   */
+  filesCommitted: optional long
+
+  /**
+   * Comma-separated list of the ids of the table snapshots committed for the dataset in this job
+   * execution (e.g. Iceberg snapshot-replication commits).
+   * Optional; absent when the writer does not commit snapshots or does not report them.
+   */
+  snapshotsCommitted: optional string
+
+  /**
+   * Comma-separated list of the partitions committed for the dataset in this job execution.
+   * Optional; absent when the writer does not commit partitions or does not report them.
+   */
+  partitionsCommitted: optional string
 }

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowexecutions.snapshot.json

@@ -186,6 +186,31 @@
                 "name" : "estimatedSecondsToCompletion",
                 "type" : "long",
                 "doc" : "estimate of time left until job completion"
+              }, {
+                "name" : "bytesWritten",
+                "type" : "long",
+                "doc" : "Number of bytes written/copied for the dataset in this job execution.\nOptional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported.",
+                "optional" : true
+              }, {
+                "name" : "recordsWritten",
+                "type" : "long",
+                "doc" : "Number of records (or files, for file/blob copies) written for the dataset in this job execution.\nOptional; -1 if unsupported by the writer.",
+                "optional" : true
+              }, {
+                "name" : "filesCommitted",
+                "type" : "long",
+                "doc" : "Number of files committed for the dataset in this job execution.\nOptional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise.",
+                "optional" : true
+              }, {
+                "name" : "snapshotsCommitted",
+                "type" : "string",
+                "doc" : "Comma-separated list of the ids of the table snapshots committed for the dataset in this job\nexecution (e.g. Iceberg snapshot-replication commits).\nOptional; absent when the writer does not commit snapshots or does not report them.",
+                "optional" : true
+              }, {
+                "name" : "partitionsCommitted",
+                "type" : "string",
+                "doc" : "Comma-separated list of the partitions committed for the dataset in this job execution.\nOptional; absent when the writer does not commit partitions or does not report them.",
+                "optional" : true
               } ]
             },
             "doc" : "Statistics from the job execution. The values may be updated during the run of a job."

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-api/src/main/snapshot/org.apache.gobblin.service.flowstatuses.snapshot.json

@@ -188,6 +188,31 @@
                 "name" : "estimatedSecondsToCompletion",
                 "type" : "long",
                 "doc" : "estimate of time left until job completion"
+              }, {
+                "name" : "bytesWritten",
+                "type" : "long",
+                "doc" : "Number of bytes written/copied for the dataset in this job execution.\nOptional; populated by writers that report byte counts (e.g. DDM file/blob replication), -1 if unsupported.",
+                "optional" : true
+              }, {
+                "name" : "recordsWritten",
+                "type" : "long",
+                "doc" : "Number of records (or files, for file/blob copies) written for the dataset in this job execution.\nOptional; -1 if unsupported by the writer.",
+                "optional" : true
+              }, {
+                "name" : "filesCommitted",
+                "type" : "long",
+                "doc" : "Number of files committed for the dataset in this job execution.\nOptional; populated by file/blob writers (e.g. DDM Iceberg replication), absent otherwise.",
+                "optional" : true
+              }, {
+                "name" : "snapshotsCommitted",
+                "type" : "string",
+                "doc" : "Comma-separated list of the ids of the table snapshots committed for the dataset in this job\nexecution (e.g. Iceberg snapshot-replication commits).\nOptional; absent when the writer does not commit snapshots or does not report them.",
+                "optional" : true
+              }, {
+                "name" : "partitionsCommitted",
+                "type" : "string",
+                "doc" : "Comma-separated list of the partitions committed for the dataset in this job execution.\nOptional; absent when the writer does not commit partitions or does not report them.",
+                "optional" : true
               } ]
             },
             "doc" : "Statistics from the job execution. The values may be updated during the run of a job."

gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java

@@ -184,17 +184,35 @@ public static FlowExecution convertFlowStatus(FlowStatus monitoringFlowStatus,
       Long timeLeft = estimateCopyTimeLeft(queriedJobStatus.getLastProgressEventTime(), queriedJobStatus.getStartTime(),
           queriedJobStatus.getProgressPercentage());
 
+      JobStatistics jobStatistics = new JobStatistics()
+          .setExecutionStartTime(queriedJobStatus.getStartTime())
+          .setExecutionEndTime(queriedJobStatus.getEndTime())
+          .setProcessedCount(queriedJobStatus.getProcessedCount())
+          .setJobProgress(queriedJobStatus.getProgressPercentage())
+          .setEstimatedSecondsToCompletion(timeLeft);
+      // Optional per-dataset copy metrics (e.g. DDM file/blob replication); set only when reported.
+      if (queriedJobStatus.getBytesWritten() >= 0) {
+        jobStatistics.setBytesWritten(queriedJobStatus.getBytesWritten());
+      }
+      if (queriedJobStatus.getRecordsWritten() >= 0) {
+        jobStatistics.setRecordsWritten(queriedJobStatus.getRecordsWritten());
+      }
+      if (queriedJobStatus.getFilesCommitted() >= 0) {
+        jobStatistics.setFilesCommitted(queriedJobStatus.getFilesCommitted());
+      }
+      if (queriedJobStatus.getSnapshotsCommitted() != null) {
+        jobStatistics.setSnapshotsCommitted(queriedJobStatus.getSnapshotsCommitted());
+      }
+      if (queriedJobStatus.getPartitionsCommitted() != null) {
+        jobStatistics.setPartitionsCommitted(queriedJobStatus.getPartitionsCommitted());
+      }
+
       jobStatus.setFlowId(flowId)
           .setJobId(new JobId()
               .setJobName(queriedJobStatus.getJobName())
               .setJobGroup(queriedJobStatus.getJobGroup()))
           .setJobTag(queriedJobStatus.getJobTag(), SetMode.IGNORE_NULL)
-          .setExecutionStatistics(new JobStatistics()
-              .setExecutionStartTime(queriedJobStatus.getStartTime())
-              .setExecutionEndTime(queriedJobStatus.getEndTime())
-              .setProcessedCount(queriedJobStatus.getProcessedCount())
-              .setJobProgress(queriedJobStatus.getProgressPercentage())
-              .setEstimatedSecondsToCompletion(timeLeft))
+          .setExecutionStatistics(jobStatistics)
           .setExecutionStatus(ExecutionStatus.valueOf(queriedJobStatus.getEventName()))
           .setMessage(queriedJobStatus.getMessage())
           .setJobState(new JobState()

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/DatasetTaskSummary.java

@@ -18,6 +18,7 @@
 package org.apache.gobblin.runtime;
 
 import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.NonNull;
@@ -34,6 +35,7 @@
 @Data
 @Setter(AccessLevel.NONE) // NOTE: non-`final` members solely to enable deserialization
 @RequiredArgsConstructor
+@AllArgsConstructor
 @NoArgsConstructor
 @ToString
 public class DatasetTaskSummary {
@@ -42,11 +44,18 @@ public class DatasetTaskSummary {
   @NonNull private long bytesWritten;
   @NonNull private boolean successfullyCommitted;
   @NonNull private String dataQualityStatus;
+  // NOTE: intentionally NOT @NonNull so the 5-arg @RequiredArgsConstructor stays intact for native
+  // Gobblin call sites (AbstractJobLauncher). Usually populated via reflection during JSON
+  // deserialization of events that carry them (e.g. DDM Iceberg snapshot replication); the
+  // @AllArgsConstructor gives native producers a direct path. Both are comma-separated lists (a
+  // run can commit more than one snapshot/partition); null = unsupported/unreported.
+  private String snapshotsCommitted;
+  private String partitionsCommitted;
 
   /**
    * Convert a {@link DatasetTaskSummary} to a {@link DatasetMetric}.
    */
   public static DatasetMetric toDatasetMetric(DatasetTaskSummary datasetTaskSummary) {
-    return new DatasetMetric(datasetTaskSummary.getDatasetUrn(), datasetTaskSummary.getBytesWritten(), datasetTaskSummary.getRecordsWritten(), datasetTaskSummary.isSuccessfullyCommitted(), datasetTaskSummary.getDataQualityStatus());
+    return new DatasetMetric(datasetTaskSummary.getDatasetUrn(), datasetTaskSummary.getBytesWritten(), datasetTaskSummary.getRecordsWritten(), datasetTaskSummary.isSuccessfullyCommitted(), datasetTaskSummary.getDataQualityStatus(), datasetTaskSummary.getSnapshotsCommitted(), datasetTaskSummary.getPartitionsCommitted());
   }
 }

gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatus.java

@@ -60,4 +60,12 @@ public class JobStatus {
   private final Supplier<List<Issue>> issues;
   private final int progressPercentage;
   private final long lastProgressEventTime;
+  // Optional per-dataset copy metrics surfaced on the job-status REST API (JobStatistics).
+  // Populated for writers that report them (e.g. DDM file/blob replication); -1 = unset/unsupported
+  // for the counts. The snapshot ids and partitions are comma-separated lists; null = unreported.
+  @Builder.Default private final long bytesWritten = -1L;
+  @Builder.Default private final long recordsWritten = -1L;
+  @Builder.Default private final long filesCommitted = -1L;
+  @Builder.Default private final String snapshotsCommitted = null;
+  @Builder.Default private final String partitionsCommitted = null;
 }

gobblin-runtime/src/main/java/org/apache/gobblin/service/monitoring/JobStatusRetriever.java

@@ -60,6 +60,16 @@ public abstract class JobStatusRetriever implements LatestFlowExecutionIdTracker
   public static final String EVENT_NAME_FIELD = "eventName";
   public static final String NA_KEY = "NA";
 
+  // File/blob movement writers report per-dataset copy metrics on the JobSummary event via these
+  // flat metadata keys; the job-status monitor persists them into the job state, and they are
+  // surfaced here on the JobStatistics REST surface. -1 = absent/unsupported for the counts;
+  // the snapshot ids and partitions are comma-separated lists, null = absent/unreported.
+  public static final String BYTES_WRITTEN_FIELD = "bytesCopied";
+  public static final String RECORDS_WRITTEN_FIELD = "rowsCopied";
+  public static final String FILES_COMMITTED_FIELD = "filesCommitted";
+  public static final String SNAPSHOTS_COMMITTED_FIELD = "snapshotsCommitted";
+  public static final String PARTITIONS_COMMITTED_FIELD = "partitionsCommitted";
+
   @Getter
   protected final MetricContext metricContext;
 
@@ -156,13 +166,20 @@ public static JobStatus.JobStatusBuilder createJobStatusBuilderFromState(State j
     boolean shouldRetry = Boolean.parseBoolean(jobState.getProp(TimingEvent.FlowEventConstants.SHOULD_RETRY_FIELD, "false"));
     int progressPercentage = jobState.getPropAsInt(TimingEvent.JOB_COMPLETION_PERCENTAGE, 0);
     long lastProgressEventTime = jobState.getPropAsLong(TimingEvent.JOB_LAST_PROGRESS_EVENT_TIME, 0);
+    long bytesWritten = jobState.getPropAsLong(BYTES_WRITTEN_FIELD, -1L);
+    long recordsWritten = jobState.getPropAsLong(RECORDS_WRITTEN_FIELD, -1L);
+    long filesCommitted = jobState.getPropAsLong(FILES_COMMITTED_FIELD, -1L);
+    String snapshotsCommitted = jobState.getProp(SNAPSHOTS_COMMITTED_FIELD, null);
+    String partitionsCommitted = jobState.getProp(PARTITIONS_COMMITTED_FIELD, null);
 
     return JobStatus.builder().flowName(flowName).flowGroup(flowGroup).flowExecutionId(flowExecutionId).jobName(jobName)
         .jobGroup(jobGroup).jobTag(jobTag).jobExecutionId(jobExecutionId).eventName(eventName).lowWatermark(lowWatermark)
         .highWatermark(highWatermark).orchestratedTime(orchestratedTime).startTime(startTime).endTime(endTime)
         .message(message).processedCount(processedCount).maxAttempts(maxAttempts).currentAttempts(currentAttempts)
         .currentGeneration(currentGeneration).shouldRetry(shouldRetry).progressPercentage(progressPercentage)
-        .lastProgressEventTime(lastProgressEventTime);
+        .lastProgressEventTime(lastProgressEventTime)
+        .bytesWritten(bytesWritten).recordsWritten(recordsWritten).filesCommitted(filesCommitted)
+        .snapshotsCommitted(snapshotsCommitted).partitionsCommitted(partitionsCommitted);
   }
 
   protected static final String getFlowGroup(State jobState) {