You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by su...@apache.org on 2019/06/19 23:59:44 UTC
[incubator-gobblin] branch master updated: [GOBBLIN-806] Enable
metrics reporter during dataset discovery for retention job
This is an automated email from the ASF dual-hosted git repository.
suvasude pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-gobblin.git
The following commit(s) were added to refs/heads/master by this push:
new fd38b4b [GOBBLIN-806] Enable metrics reporter during dataset discovery for retention job
fd38b4b is described below
commit fd38b4b48520b2ca9c75de1613169b59de0a3dc2
Author: Zihan Li <zi...@zihli-mn1.linkedin.biz>
AuthorDate: Wed Jun 19 16:59:38 2019 -0700
[GOBBLIN-806] Enable metrics reporter during dataset discovery for retention job
Enable metrics reporter during dataset discovery
for retention job
Enable event submitter for dataset finder imported
by tag
add SuppressWarnings
address comments
Closes #2672 from ZihanLi58/GOBBLIN-806
---
.../data/management/retention/DatasetCleaner.java | 16 ++++++++--------
.../retention/profile/MultiCleanableDatasetFinder.java | 8 +++++++-
.../retention/profile/MultiDatasetFinder.java | 17 +++++++++++------
3 files changed, 26 insertions(+), 15 deletions(-)
diff --git a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/DatasetCleaner.java b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/DatasetCleaner.java
index 6793f5a..848b0d2 100644
--- a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/DatasetCleaner.java
+++ b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/DatasetCleaner.java
@@ -92,6 +92,13 @@ public class DatasetCleaner implements Instrumentable, Closeable {
FileSystem targetFs =
props.containsKey(ConfigurationKeys.WRITER_FILE_SYSTEM_URI) ? WriterUtils.getWriterFs(state) : fs;
this.closer = Closer.create();
+ // TODO -- Remove the dependency on gobblin-core after new Gobblin Metrics does not depend on gobblin-core.
+ List<Tag<?>> tags = Lists.newArrayList();
+ tags.addAll(Tag.fromMap(AzkabanTags.getAzkabanTags()));
+ this.metricContext =
+ this.closer.register(Instrumented.getMetricContext(new State(props), DatasetCleaner.class, tags));
+ this.isMetricEnabled = GobblinMetrics.isEnabled(props);
+ this.eventSubmitter = new EventSubmitter.Builder(this.metricContext, RetentionEvents.NAMESPACE).build();
try {
FileSystem optionalRateControlledFs = targetFs;
if (props.contains(DATASET_CLEAN_HDFS_CALLS_PER_SECOND_LIMIT)) {
@@ -100,7 +107,7 @@ public class DatasetCleaner implements Instrumentable, Closeable {
((RateControlledFileSystem) optionalRateControlledFs).startRateControl();
}
- this.datasetFinder = new MultiCleanableDatasetFinder(optionalRateControlledFs, props);
+ this.datasetFinder = new MultiCleanableDatasetFinder(optionalRateControlledFs, props, eventSubmitter);
} catch (NumberFormatException exception) {
throw new IOException(exception);
} catch (ExecutionException exception) {
@@ -111,13 +118,6 @@ public class DatasetCleaner implements Instrumentable, Closeable {
100, ExecutorsUtils.newThreadFactory(Optional.of(LOG), Optional.of("Dataset-cleaner-pool-%d")));
this.service = ExecutorsUtils.loggingDecorator(executor);
- List<Tag<?>> tags = Lists.newArrayList();
- tags.addAll(Tag.fromMap(AzkabanTags.getAzkabanTags()));
- // TODO -- Remove the dependency on gobblin-core after new Gobblin Metrics does not depend on gobblin-core.
- this.metricContext =
- this.closer.register(Instrumented.getMetricContext(new State(props), DatasetCleaner.class, tags));
- this.isMetricEnabled = GobblinMetrics.isEnabled(props);
- this.eventSubmitter = new EventSubmitter.Builder(this.metricContext, RetentionEvents.NAMESPACE).build();
this.throwables = Lists.newArrayList();
}
diff --git a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiCleanableDatasetFinder.java b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiCleanableDatasetFinder.java
index 93479e5..36007b3 100644
--- a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiCleanableDatasetFinder.java
+++ b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiCleanableDatasetFinder.java
@@ -16,11 +16,13 @@
*/
package org.apache.gobblin.data.management.retention.profile;
+import com.google.common.base.Optional;
import org.apache.gobblin.data.management.retention.DatasetCleaner;
import java.net.URI;
import java.util.Properties;
+import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.hadoop.fs.FileSystem;
import com.typesafe.config.Config;
@@ -49,7 +51,11 @@ public class MultiCleanableDatasetFinder extends MultiDatasetFinder {
public MultiCleanableDatasetFinder(FileSystem fs, Properties jobProps) {
- super(fs, jobProps);
+ this(fs,jobProps,new EventSubmitter.Builder(Optional.absent(),"noMessage").build());
+ }
+
+ public MultiCleanableDatasetFinder(FileSystem fs, Properties jobProps, EventSubmitter eventSubmitter) {
+ super(fs, jobProps, eventSubmitter);
}
@Override
diff --git a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiDatasetFinder.java b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiDatasetFinder.java
index d1bb84e..8c38ae7 100644
--- a/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiDatasetFinder.java
+++ b/gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/profile/MultiDatasetFinder.java
@@ -16,6 +16,7 @@
*/
package org.apache.gobblin.data.management.retention.profile;
+import com.google.common.base.Optional;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.URI;
@@ -26,7 +27,7 @@ import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.reflect.ConstructorUtils;
+import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -72,6 +73,10 @@ public abstract class MultiDatasetFinder implements DatasetsFinder<Dataset> {
@SuppressWarnings({ "rawtypes", "unchecked" })
public MultiDatasetFinder(FileSystem fs, Properties jobProps) {
+ this(fs,jobProps,new EventSubmitter.Builder(Optional.absent(),"noMessage").build());
+ }
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public MultiDatasetFinder(FileSystem fs, Properties jobProps, EventSubmitter eventSubmitter) {
this.jobProps = jobProps;
try {
this.datasetFinders = Lists.newArrayList();
@@ -79,10 +84,9 @@ public abstract class MultiDatasetFinder implements DatasetsFinder<Dataset> {
if (jobProps.containsKey(datasetFinderClassKey())) {
try {
log.info(String.format("Instantiating datasetfinder %s ", jobProps.getProperty(datasetFinderClassKey())));
- this.datasetFinders.add((DatasetsFinder) ConstructorUtils.invokeConstructor(
- Class.forName(jobProps.getProperty(datasetFinderClassKey())), fs, jobProps));
- } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException
- | ClassNotFoundException e) {
+ this.datasetFinders.add((DatasetsFinder) GobblinConstructorUtils.invokeLongestConstructor(
+ Class.forName(jobProps.getProperty(datasetFinderClassKey())), fs, jobProps, eventSubmitter));
+ } catch (ReflectiveOperationException e) {
log.error(
String.format("Retention ignored could not instantiate datasetfinder %s.",
jobProps.getProperty(datasetFinderClassKey())), e);
@@ -106,7 +110,8 @@ public abstract class MultiDatasetFinder implements DatasetsFinder<Dataset> {
try {
this.datasetFinders.add((DatasetsFinder) GobblinConstructorUtils.invokeFirstConstructor(
Class.forName(datasetClassConfig.getString(datasetFinderClassKey())), ImmutableList.of(fs, jobProps,
- datasetClassConfig), ImmutableList.of(fs, jobProps)));
+ datasetClassConfig), ImmutableList.of(fs, jobProps, eventSubmitter),
+ ImmutableList.of(fs, jobProps)));
log.info(String.format("Instantiated datasetfinder %s for %s.",
datasetClassConfig.getString(datasetFinderClassKey()), importedBy));
} catch (InstantiationException | IllegalAccessException | IllegalArgumentException