[GOBBLIN-2271] Add retention (deletion) counts to DatasetMetric in GaaSJobObservabilityEvent

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

@@ -276,6 +276,24 @@
                 "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
+              },
+              {
+                "name": "versionsTotal",
+                "type": ["null","long"],
+                "doc": "Retention (data-deletion) jobs only: number of dataset versions considered under the dataset root (deletable + retained), null for jobs that do not delete",
+                "default": null
+              },
+              {
+                "name": "versionsDeleted",
+                "type": ["null","long"],
+                "doc": "Retention (data-deletion) jobs only: number of dataset versions deleted (or, on a dry run, that would be deleted), null for jobs that do not delete",
+                "default": null
+              },
+              {
+                "name": "directoriesPruned",
+                "type": ["null","long"],
+                "doc": "Retention (data-deletion) jobs only: number of now-empty parent directories pruned after the versions were deleted, null for jobs that do not delete",
+                "default": null
               }
             ]
           }

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

@@ -51,11 +51,27 @@ public class DatasetTaskSummary {
   // run can commit more than one snapshot/partition); null = unsupported/unreported.
   private String snapshotsCommitted;
   private String partitionsCommitted;
+  // Retention (data-deletion) counts, likewise not @NonNull and populated during JSON deserialization of
+  // events that carry them (e.g. DDM retention runs); null = not a deletion job.
+  private Long versionsTotal;
+  private Long versionsDeleted;
+  private Long directoriesPruned;
 
   /**
    * 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(), datasetTaskSummary.getSnapshotsCommitted(), datasetTaskSummary.getPartitionsCommitted());
+    return DatasetMetric.newBuilder()
+        .setDatasetUrn(datasetTaskSummary.getDatasetUrn())
+        .setBytesWritten(datasetTaskSummary.getBytesWritten())
+        .setEntitiesWritten(datasetTaskSummary.getRecordsWritten())
+        .setSuccessfullyCommitted(datasetTaskSummary.isSuccessfullyCommitted())
+        .setDataQualityStatus(datasetTaskSummary.getDataQualityStatus())
+        .setSnapshotsCommitted(datasetTaskSummary.getSnapshotsCommitted())
+        .setPartitionsCommitted(datasetTaskSummary.getPartitionsCommitted())
+        .setVersionsTotal(datasetTaskSummary.getVersionsTotal())
+        .setVersionsDeleted(datasetTaskSummary.getVersionsDeleted())
+        .setDirectoriesPruned(datasetTaskSummary.getDirectoriesPruned())
+        .build();
   }
 }

gobblin-runtime/src/test/java/org/apache/gobblin/runtime/DatasetTaskSummaryTest.java

@@ -0,0 +1,53 @@
+/*
+ * 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.gobblin.runtime;
+
+import java.lang.reflect.Type;
+import java.util.List;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.google.gson.reflect.TypeToken;
+
+import org.apache.gobblin.metrics.DatasetMetric;
+import org.apache.gobblin.runtime.util.GsonUtils;
+
+
+public class DatasetTaskSummaryTest {
+
+  private static final Type SUMMARY_LIST_TYPE = new TypeToken<List<DatasetTaskSummary>>() { }.getType();
+
+  @Test
+  public void testToDatasetMetricMapsRetentionCountsFromJson() {
+    // The shape a DDM retention run emits: nothing written, deletion counts set.
+    String json = "[{\"datasetUrn\":\"hdfs://nn/data/foo\",\"recordsWritten\":0,\"bytesWritten\":0,"
+        + "\"successfullyCommitted\":true,\"versionsTotal\":10,\"versionsDeleted\":4,\"directoriesPruned\":1}]";
+    List<DatasetTaskSummary> summaries = GsonUtils.GSON_WITH_DATE_HANDLING.fromJson(json, SUMMARY_LIST_TYPE);
+
+    DatasetMetric metric = DatasetTaskSummary.toDatasetMetric(summaries.get(0));
+
+    Assert.assertEquals(metric.getDatasetUrn(), "hdfs://nn/data/foo");
+    Assert.assertEquals(metric.getEntitiesWritten(), Long.valueOf(0L));
+    Assert.assertEquals(metric.getBytesWritten(), Long.valueOf(0L));
+    Assert.assertTrue(metric.getSuccessfullyCommitted());
+    Assert.assertEquals(metric.getVersionsTotal(), Long.valueOf(10L));
+    Assert.assertEquals(metric.getVersionsDeleted(), Long.valueOf(4L));
+    Assert.assertEquals(metric.getDirectoriesPruned(), Long.valueOf(1L));
+  }
+}