Add retention job summary events

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/dataset/DummyDataset.java

@@ -17,21 +17,21 @@
 
 package org.apache.gobblin.data.management.dataset;
 
-import lombok.RequiredArgsConstructor;
-
 import java.io.IOException;
 import java.util.Collection;
 
-import org.apache.gobblin.dataset.FileSystemDataset;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
 import com.google.common.collect.ImmutableList;
 
+import lombok.RequiredArgsConstructor;
+
 import org.apache.gobblin.data.management.copy.CopyConfiguration;
-import org.apache.gobblin.data.management.copy.CopyableDataset;
 import org.apache.gobblin.data.management.copy.CopyEntity;
+import org.apache.gobblin.data.management.copy.CopyableDataset;
 import org.apache.gobblin.data.management.retention.dataset.CleanableDataset;
+import org.apache.gobblin.dataset.FileSystemDataset;
 
 
 /**
@@ -42,8 +42,9 @@ public class DummyDataset implements CopyableDataset, CleanableDataset, FileSyst
 
   private final Path datasetRoot;
 
-  @Override public void clean() throws IOException {
+  @Override public int clean() throws IOException {
     // Do nothing
+    return 0;
   }
 
   @Override public Collection<? extends CopyEntity> getCopyableFiles(FileSystem targetFs,

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/DatasetCleaner.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -47,6 +48,7 @@
 import org.apache.gobblin.configuration.State;
 import org.apache.gobblin.data.management.retention.dataset.CleanableDataset;
 import org.apache.gobblin.data.management.retention.profile.MultiCleanableDatasetFinder;
+import org.apache.gobblin.data.management.retention.version.DatasetRetentionSummary;
 import org.apache.gobblin.dataset.Dataset;
 import org.apache.gobblin.dataset.DatasetsFinder;
 import org.apache.gobblin.instrumented.Instrumentable;
@@ -56,6 +58,7 @@
 import org.apache.gobblin.metrics.Tag;
 import org.apache.gobblin.metrics.event.EventSubmitter;
 import org.apache.gobblin.runtime.DynamicConfigGeneratorFactory;
+import org.apache.gobblin.runtime.util.GsonUtils;
 import org.apache.gobblin.util.AzkabanTags;
 import org.apache.gobblin.util.ConfigUtils;
 import org.apache.gobblin.util.ExecutorsUtils;
@@ -88,6 +91,7 @@ public class DatasetCleaner implements Instrumentable, Closeable {
   private Optional<Meter> datasetsCleanSuccessMeter = Optional.absent();
   private Optional<Meter> datasetsCleanFailureMeter = Optional.absent();
   private Optional<CountDownLatch> finishCleanSignal;
+  private final ConcurrentHashMap<String, DatasetRetentionSummary> datasetRetentionSummaries = new ConcurrentHashMap<>();
   private final List<Throwable> throwables;
 
   public DatasetCleaner(FileSystem fs, Properties props) throws IOException {
@@ -141,21 +145,24 @@ public void clean() throws IOException {
     List<Dataset> dataSets = this.datasetFinder.findDatasets();
     this.finishCleanSignal = Optional.of(new CountDownLatch(dataSets.size()));
     for (final Dataset dataset : dataSets) {
-      ListenableFuture<Void> future = this.service.submit(new Callable<Void>() {
+      ListenableFuture<Integer> future = this.service.submit(new Callable<Integer>(){
         @Override
-        public Void call() throws Exception {
+        public Integer call() throws Exception {
           if (dataset instanceof CleanableDataset) {
-            ((CleanableDataset) dataset).clean();
+            int versionsDeleted = ((CleanableDataset) dataset).clean();
+            return versionsDeleted;
           }
-          return null;
+          return 0;
         }
       });
-      Futures.addCallback(future, new FutureCallback<Void>() {
+      Futures.addCallback(future, new FutureCallback<Integer>() {
         @Override
         public void onFailure(Throwable throwable) {
           DatasetCleaner.this.finishCleanSignal.get().countDown();
           LOG.warn("Exception caught when cleaning " + dataset.datasetURN() + ".", throwable);
           DatasetCleaner.this.throwables.add(throwable);
+          DatasetRetentionSummary summary = new DatasetRetentionSummary(dataset.datasetURN(), 0, false);
+          DatasetCleaner.this.datasetRetentionSummaries.put(dataset.datasetURN(), summary);
           Instrumented.markMeter(DatasetCleaner.this.datasetsCleanFailureMeter);
           DatasetCleaner.this.eventSubmitter.submit(RetentionEvents.CleanFailed.EVENT_NAME,
               ImmutableMap.of(RetentionEvents.CleanFailed.FAILURE_CONTEXT_METADATA_KEY,
@@ -164,9 +171,11 @@ public void onFailure(Throwable throwable) {
         }
 
         @Override
-        public void onSuccess(Void arg0) {
+        public void onSuccess(Integer datasetsDeleted) {
           DatasetCleaner.this.finishCleanSignal.get().countDown();
-          LOG.info("Successfully cleaned: " + dataset.datasetURN());
+          DatasetRetentionSummary summary = new DatasetRetentionSummary(dataset.datasetURN(), datasetsDeleted, true);
+          DatasetCleaner.this.datasetRetentionSummaries.put(dataset.datasetURN(), summary);
+          LOG.info("Successfully cleaned: {} with {} versions marked for deletion", dataset.datasetURN(), datasetsDeleted);
           Instrumented.markMeter(DatasetCleaner.this.datasetsCleanSuccessMeter);
         }
       });
@@ -187,8 +196,12 @@ public void close() throws IOException {
       }
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
-      throw new IOException("Not all datasets finish cleanning", e);
+      throw new IOException("Not all datasets finished cleaning", e);
     } finally {
+      // Submit a job summary event
+      String serializedSummary = GsonUtils.gsonBuilderWithSerializationSupport().disableHtmlEscaping().create().toJson(this.datasetRetentionSummaries.values());
+      this.eventSubmitter.submit(RetentionEvents.RetentionJobSummary.EVENT_NAME,
+          ImmutableMap.of(RetentionEvents.RetentionJobSummary.DATASET_VERSIONS_CLEANED_METADATA_KEY, serializedSummary));
       ExecutorsUtils.shutdownExecutorService(this.service, Optional.of(LOG));
       this.closer.close();
     }

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/RetentionEvents.java

@@ -19,6 +19,7 @@
 
 import org.apache.gobblin.data.management.retention.dataset.CleanableDataset;
 
+
 /**
  * Holds event names and constants used in events submitted by a retention job.
  */
@@ -34,6 +35,12 @@ static class CleanFailed {
      */
     static final String FAILURE_CONTEXT_METADATA_KEY = "failureContext";
   }
+
+  static class RetentionJobSummary {
+    static final String EVENT_NAME = "RetentionJobSummary";
+    static final String DATASET_VERSIONS_CLEANED_METADATA_KEY = "datasetsMarkedForCleaning";
+  }
+
   static final String NAMESPACE = "gobblin.data.management.retention";
   static final String DATASET_URN_METADATA_KEY = "datasetUrn";
 }

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/CleanableDataset.java

@@ -18,6 +18,7 @@
 package org.apache.gobblin.data.management.retention.dataset;
 
 import java.io.IOException;
+
 import org.apache.gobblin.dataset.Dataset;
 
 
@@ -28,10 +29,11 @@
 public interface CleanableDataset extends Dataset {
 
   /**
-   * Cleans the {@link CleanableDataset}. In general, this means to apply a
+   * Cleans the {@link CleanableDataset}.
+   * Returns the number of versions marked for deletion
    * {@link org.apache.gobblin.data.management.retention.policy.RetentionPolicy} and delete files and directories that need deleting.
    * @throws IOException
    */
-  public void clean() throws IOException;
+  public int clean() throws IOException;
 
 }

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/CleanableDatasetStoreDataset.java

@@ -21,13 +21,15 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+
+import com.google.common.collect.Lists;
+
 import org.apache.gobblin.data.management.policy.VersionSelectionPolicy;
 import org.apache.gobblin.data.management.version.DatasetStateStoreVersion;
 import org.apache.gobblin.data.management.version.DatasetVersion;
 import org.apache.gobblin.data.management.version.finder.VersionFinder;
 import org.apache.gobblin.metastore.DatasetStoreDataset;
 import org.apache.gobblin.metastore.metadata.DatasetStateStoreEntryManager;
-import com.google.common.collect.Lists;
 
 
 /**
@@ -44,7 +46,7 @@ public CleanableDatasetStoreDataset(DatasetStoreDataset.Key key, List<DatasetSta
   public abstract VersionSelectionPolicy<T> getVersionSelectionPolicy();
 
   @Override
-  public void clean() throws IOException {
+  public int clean() throws IOException {
 
     List<T> versions = Lists.newArrayList(this.getVersionFinder().findDatasetVersions(this));
 
@@ -55,5 +57,7 @@ public void clean() throws IOException {
     for (Object version : deletableVersions) {
       ((DatasetStateStoreVersion) version).getEntry().delete();
     }
+
+    return deletableVersions.size();
   }
 }

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/CleanableHiveDataset.java

@@ -120,13 +120,13 @@ public CleanableHiveDataset(FileSystem fs, HiveMetastoreClientPool clientPool, T
    * </p>
    */
   @Override
-  public void clean() throws IOException {
+  public int clean() throws IOException {
 
     List versions = Lists.newArrayList(this.hiveDatasetVersionFinder.findDatasetVersions(this));
 
     if (versions.isEmpty()) {
       log.warn(String.format("No dataset version can be found. Ignoring %s", this.getTable().getCompleteName()));
-      return;
+      return 0;
     }
 
     Collections.sort(versions, Collections.reverseOrder());
@@ -159,6 +159,7 @@ public void clean() throws IOException {
     if (!exceptions.isEmpty()) {
       throw new RuntimeException(String.format("Deletion failed for %s partitions", exceptions.size()));
     }
+    return deletableVersions.size();
   }
 
   @Override

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/CleanableIcebergDataset.java

@@ -17,10 +17,6 @@
 
 package org.apache.gobblin.data.management.retention.dataset;
 
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.typesafe.config.Config;
-import com.typesafe.config.ConfigFactory;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -30,6 +26,18 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.stream.Collectors;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
 import org.apache.gobblin.configuration.ConfigurationKeys;
 import org.apache.gobblin.configuration.State;
 import org.apache.gobblin.data.management.policy.VersionSelectionPolicy;
@@ -39,11 +47,6 @@
 import org.apache.gobblin.iceberg.GobblinMCEProducer;
 import org.apache.gobblin.metadata.OperationType;
 import org.apache.gobblin.metadata.SchemaSource;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.iceberg.catalog.TableIdentifier;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 
 /**
@@ -75,15 +78,15 @@ public CleanableIcebergDataset(FileSystem fs, Properties props, Path datasetRoot
   }
 
   @Override
-  public void clean() throws IOException {
+  public int clean() throws IOException {
 
     if (this.isDatasetBlacklisted) {
       this.log.info("Dataset blacklisted. Cleanup skipped for " + datasetRoot());
-      return;
+      return 0;
     }
 
     boolean atLeastOneFailureSeen = false;
-
+    int totalVersionsDeleted = 0;
     for (VersionFinderAndPolicy<T> versionFinderAndPolicy : getVersionFindersAndPolicies()) {
       Config retentionConfig = versionFinderAndPolicy.getConfig();
       Preconditions.checkArgument(retentionConfig != null,
@@ -108,7 +111,7 @@ public void clean() throws IOException {
       Collections.sort(versions, Collections.reverseOrder());
 
       Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
+      totalVersionsDeleted += deletableVersions.size();
       cleanImpl(deletableVersions, retentionConfig);
     }
 
@@ -125,6 +128,7 @@ public void clean() throws IOException {
       log.error("interrupted while sleep");
       throw new IOException(e);
     }
+    return totalVersionsDeleted;
   }
 
   /**

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/MultiVersionCleanableDatasetBase.java

@@ -23,11 +23,6 @@
 import java.util.List;
 import java.util.Properties;
 
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.Singular;
-
 import org.apache.hadoop.fs.FileSystem;
 import org.slf4j.Logger;
 
@@ -37,6 +32,11 @@
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
 
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Singular;
+
 import org.apache.gobblin.data.management.policy.EmbeddedRetentionSelectionPolicy;
 import org.apache.gobblin.data.management.policy.SelectNothingPolicy;
 import org.apache.gobblin.data.management.policy.VersionSelectionPolicy;
@@ -255,15 +255,16 @@ public MultiVersionCleanableDatasetBase(FileSystem fs, Properties properties, bo
    *
    */
   @Override
-  public void clean() throws IOException {
+  public int clean() throws IOException {
 
     if (this.isDatasetBlacklisted) {
       this.log.info("Dataset blacklisted. Cleanup skipped for " + datasetRoot());
-      return;
+      return 0;
     }
 
     boolean atLeastOneFailureSeen = false;
 
+    int totalVersionsDeleted = 0;
     for (VersionFinderAndPolicy<T> versionFinderAndPolicy : getVersionFindersAndPolicies()) {
 
       VersionSelectionPolicy<T> selectionPolicy = versionFinderAndPolicy.getVersionSelectionPolicy();
@@ -277,7 +278,6 @@ public void clean() throws IOException {
           versionFinder.getClass().getName(), selectionPolicy));
 
       List<T> versions = Lists.newArrayList(versionFinder.findDatasetVersions(this));
-
       if (versions.isEmpty()) {
         this.log.warn("No dataset version can be found. Ignoring.");
         continue;
@@ -286,7 +286,7 @@ public void clean() throws IOException {
       Collections.sort(versions, Collections.reverseOrder());
 
       Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
+      totalVersionsDeleted += deletableVersions.size();
       cleanImpl(deletableVersions);
 
       List<DatasetVersion> allVersions = Lists.newArrayList();
@@ -308,6 +308,7 @@ public void clean() throws IOException {
       throw new RuntimeException(String.format(
           "At least one failure happened while processing %s. Look for previous logs for failures", datasetRoot()));
     }
+    return totalVersionsDeleted;
   }
 
   protected void cleanImpl(Collection<T> deletableVersions) throws IOException {