feat: add parquet source metrics

dagger-core/build.gradle

@@ -61,6 +61,7 @@ configurations {
 dependencies {
     minimalJar project(path: ':dagger-common', configuration: 'minimalCommonJar')
     minimalJar project(path: ':dagger-functions', configuration: 'minimalFunctionsJar')
+    minimalJar 'io.odpf:depot:0.1.5'
 
     compileOnly 'org.projectlombok:lombok:1.18.8'
     annotationProcessor 'org.projectlombok:lombok:1.18.8'

dagger-core/src/main/java/io/odpf/dagger/core/StreamManager.java

@@ -1,5 +1,6 @@
 package io.odpf.dagger.core;
 
+import io.odpf.dagger.core.metrics.reporters.statsd.DaggerStatsDReporter;
 import io.odpf.dagger.core.source.Stream;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -220,6 +221,8 @@ private void addSink(StreamInfo streamInfo) {
     }
 
     List<Stream> getStreams() {
-        return StreamsFactory.getStreams(configuration, stencilClientOrchestrator);
+        org.apache.flink.configuration.Configuration flinkConfiguration = (org.apache.flink.configuration.Configuration) this.executionEnvironment.getConfiguration();
+        DaggerStatsDReporter daggerStatsDReporter = DaggerStatsDReporter.Provider.provide(flinkConfiguration, configuration);
+        return StreamsFactory.getStreams(configuration, stencilClientOrchestrator, daggerStatsDReporter);
     }
 }

dagger-core/src/main/java/io/odpf/dagger/core/deserializer/DaggerDeserializerFactory.java

@@ -4,6 +4,8 @@
 import io.odpf.dagger.common.core.StencilClientOrchestrator;
 import io.odpf.dagger.common.serde.DaggerDeserializer;
 import io.odpf.dagger.core.exception.DaggerConfigurationException;
+import io.odpf.dagger.core.metrics.reporters.statsd.SerializedStatsDReporterSupplier;
+import io.odpf.dagger.core.metrics.reporters.statsd.StatsDErrorReporter;
 import io.odpf.dagger.core.source.config.StreamConfig;
 import org.apache.flink.types.Row;
 
@@ -12,12 +14,17 @@
 import java.util.stream.Stream;
 
 public class DaggerDeserializerFactory {
-    public static DaggerDeserializer<Row> create(StreamConfig streamConfig, Configuration configuration, StencilClientOrchestrator stencilClientOrchestrator) {
+    public static DaggerDeserializer<Row> create(StreamConfig streamConfig, Configuration configuration, StencilClientOrchestrator stencilClientOrchestrator, SerializedStatsDReporterSupplier statsDReporterSupplier) {
         return getDaggerDeserializerProviders(streamConfig, configuration, stencilClientOrchestrator)
                 .stream()
                 .filter(DaggerDeserializerProvider::canProvide)
                 .findFirst()
-                .orElseThrow(() -> new DaggerConfigurationException("No suitable deserializer could be constructed for the given stream configuration."))
+                .orElseThrow(() -> {
+                    StatsDErrorReporter statsDErrorReporter = new StatsDErrorReporter(statsDReporterSupplier);
+                    DaggerConfigurationException ex = new DaggerConfigurationException("No suitable deserializer could be constructed for the given stream configuration.");
+                    statsDErrorReporter.reportFatalException(ex);
+                    return ex;
+                })
                 .getDaggerDeserializer();
     }
 

dagger-core/src/main/java/io/odpf/dagger/core/metrics/aspects/ChronologyOrderedSplitAssignerAspects.java

@@ -0,0 +1,28 @@
+package io.odpf.dagger.core.metrics.aspects;
+
+import io.odpf.dagger.common.metrics.aspects.AspectType;
+import io.odpf.dagger.common.metrics.aspects.Aspects;
+
+public enum ChronologyOrderedSplitAssignerAspects implements Aspects {
+    TOTAL_SPLITS_DISCOVERED("total_splits_discovered", AspectType.Gauge),
+    TOTAL_SPLITS_RECORDED("total_splits_recorded", AspectType.Gauge),
+    SPLITS_AWAITING_ASSIGNMENT("splits_awaiting_assignment", AspectType.Counter);
+
+    ChronologyOrderedSplitAssignerAspects(String value, AspectType aspectType) {
+        this.value = value;
+        this.aspectType = aspectType;
+    }
+
+    private final String value;
+    private final AspectType aspectType;
+
+    @Override
+    public String getValue() {
+        return value;
+    }
+
+    @Override
+    public AspectType getAspectType() {
+        return aspectType;
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/aspects/ParquetReaderAspects.java

@@ -0,0 +1,30 @@
+package io.odpf.dagger.core.metrics.aspects;
+
+import io.odpf.dagger.common.metrics.aspects.AspectType;
+import io.odpf.dagger.common.metrics.aspects.Aspects;
+
+public enum ParquetReaderAspects implements Aspects {
+    READER_CREATED("reader_created", AspectType.Counter),
+    READER_CLOSED("reader_closed", AspectType.Counter),
+    READER_ROWS_EMITTED("reader_rows_emitted", AspectType.Counter),
+    READER_ROW_DESERIALIZATION_TIME("reader_row_deserialization_time", AspectType.Histogram),
+    READER_ROW_READ_TIME("reader_row_read_time", AspectType.Histogram);
+
+    private final String value;
+    private final AspectType aspectType;
+
+    ParquetReaderAspects(String value, AspectType aspectType) {
+        this.value = value;
+        this.aspectType = aspectType;
+    }
+
+    @Override
+    public String getValue() {
+        return value;
+    }
+
+    @Override
+    public AspectType getAspectType() {
+        return aspectType;
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/DaggerMetricsConfig.java

@@ -0,0 +1,43 @@
+package io.odpf.dagger.core.metrics.reporters.statsd;
+
+import io.odpf.depot.config.MetricsConfig;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+
+public class DaggerMetricsConfig implements MetricsConfig {
+    private static final String FLINK_STATSD_HOST_CONFIG_KEY = "metrics.reporter.stsd.host";
+    private static final String DEFAULT_STATSD_HOST_VALUE = "localhost";
+    private static final String FLINK_STATSD_PORT_CONFIG_KEY = "metrics.reporter.stsd.port";
+    private static final int DEFAULT_STATSD_PORT_VALUE = 8125;
+    private final String hostName;
+    private final int port;
+
+    public DaggerMetricsConfig(Configuration flinkConfiguration) {
+        ConfigOption<String> hostConfigOption = ConfigOptions
+                .key(FLINK_STATSD_HOST_CONFIG_KEY)
+                .stringType()
+                .defaultValue(DEFAULT_STATSD_HOST_VALUE);
+        ConfigOption<Integer> portConfigOption = ConfigOptions
+                .key(FLINK_STATSD_PORT_CONFIG_KEY)
+                .intType()
+                .defaultValue(DEFAULT_STATSD_PORT_VALUE);
+        this.hostName = flinkConfiguration.getString(hostConfigOption);
+        this.port = flinkConfiguration.getInteger(portConfigOption);
+    }
+
+    @Override
+    public String getMetricStatsDHost() {
+        return hostName;
+    }
+
+    @Override
+    public Integer getMetricStatsDPort() {
+        return port;
+    }
+
+    @Override
+    public String getMetricStatsDTags() {
+        return "";
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/DaggerStatsDReporter.java

@@ -0,0 +1,59 @@
+package io.odpf.dagger.core.metrics.reporters.statsd;
+
+import io.odpf.dagger.core.metrics.reporters.statsd.tags.GlobalTags;
+import io.odpf.dagger.core.metrics.reporters.statsd.tags.StatsDTag;
+import io.odpf.depot.metrics.StatsDReporter;
+import io.odpf.depot.metrics.StatsDReporterBuilder;
+import org.apache.flink.configuration.Configuration;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+import static io.odpf.dagger.core.utils.Constants.FLINK_JOB_ID_DEFAULT;
+import static io.odpf.dagger.core.utils.Constants.FLINK_JOB_ID_KEY;
+
+public class DaggerStatsDReporter implements SerializedStatsDReporterSupplier {
+    private static StatsDReporter statsDReporter;
+    private final Configuration flinkConfiguration;
+    private final io.odpf.dagger.common.configuration.Configuration daggerConfiguration;
+
+    private DaggerStatsDReporter(Configuration flinkConfiguration, io.odpf.dagger.common.configuration.Configuration daggerConfiguration) {
+        this.flinkConfiguration = flinkConfiguration;
+        this.daggerConfiguration = daggerConfiguration;
+    }
+
+    private String[] generateGlobalTags() {
+        StatsDTag[] globalTags = new StatsDTag[]{
+                new StatsDTag(GlobalTags.JOB_ID, daggerConfiguration.getString(FLINK_JOB_ID_KEY, FLINK_JOB_ID_DEFAULT))};
+        return Arrays.stream(globalTags)
+                .map(StatsDTag::getFormattedTag)
+                .toArray(String[]::new);
+    }
+
+    @Override
+    public StatsDReporter buildStatsDReporter() {
+        if (statsDReporter == null) {
+            DaggerMetricsConfig daggerMetricsConfig = new DaggerMetricsConfig(flinkConfiguration);
+            String[] globalTags = generateGlobalTags();
+            statsDReporter = StatsDReporterBuilder
+                    .builder()
+                    .withMetricConfig(daggerMetricsConfig)
+                    .withExtraTags(globalTags)
+                    .build();
+        }
+        return statsDReporter;
+    }
+
+    protected static void close() throws IOException {
+        if (statsDReporter != null) {
+            statsDReporter.close();
+            statsDReporter = null;
+        }
+    }
+
+    public static class Provider {
+        public static DaggerStatsDReporter provide(Configuration flinkConfiguration, io.odpf.dagger.common.configuration.Configuration daggerConfiguration) {
+            return new DaggerStatsDReporter(flinkConfiguration, daggerConfiguration);
+        }
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/SerializedStatsDReporterSupplier.java

@@ -0,0 +1,15 @@
+package io.odpf.dagger.core.metrics.reporters.statsd;
+
+import io.odpf.depot.metrics.StatsDReporter;
+
+import java.io.Serializable;
+
+/* Flink requires that all objects which are needed to prepare the Job Graph should be serializable along with their
+properties/fields. StatsDReporter and its fields are not serializable. Hence, in order to mitigate job graph creation
+failure, we create a serializable interface around the reporter as below. This is a common idiom to make un-serializable
+fields serializable in Java 8: https://stackoverflow.com/a/22808112 */
+
+@FunctionalInterface
+public interface SerializedStatsDReporterSupplier extends Serializable {
+    StatsDReporter buildStatsDReporter();
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/StatsDErrorReporter.java

@@ -0,0 +1,39 @@
+package io.odpf.dagger.core.metrics.reporters.statsd;
+
+import io.odpf.dagger.core.metrics.reporters.ErrorReporter;
+import io.odpf.dagger.core.metrics.reporters.statsd.tags.StatsDTag;
+import io.odpf.depot.metrics.StatsDReporter;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+
+import java.io.Serializable;
+
+import static io.odpf.dagger.core.utils.Constants.FATAL_EXCEPTION_METRIC_GROUP_KEY;
+import static io.odpf.dagger.core.utils.Constants.NONFATAL_EXCEPTION_METRIC_GROUP_KEY;
+
+public class StatsDErrorReporter implements ErrorReporter, Serializable {
+    private static final String FATAL_EXCEPTION_TAG_KEY = "fatal_exception_type";
+    private static final String NON_FATAL_EXCEPTION_TAG_KEY = "non_fatal_exception_type";
+    private final StatsDReporter statsDReporter;
+
+    public StatsDErrorReporter(SerializedStatsDReporterSupplier statsDReporterSupplier) {
+        this.statsDReporter = statsDReporterSupplier.buildStatsDReporter();
+    }
+
+    @Override
+    public void reportFatalException(Exception exception) {
+        StatsDTag statsDTag = new StatsDTag(FATAL_EXCEPTION_TAG_KEY, exception.getClass().getName());
+        statsDReporter.captureCount(FATAL_EXCEPTION_METRIC_GROUP_KEY, 1L, statsDTag.getFormattedTag());
+    }
+
+    @Override
+    public void reportNonFatalException(Exception exception) {
+        StatsDTag statsDTag = new StatsDTag(NON_FATAL_EXCEPTION_TAG_KEY, exception.getClass().getName());
+        statsDReporter.captureCount(NONFATAL_EXCEPTION_METRIC_GROUP_KEY, 1L, statsDTag.getFormattedTag());
+    }
+
+    @Override
+    public Counter addExceptionToCounter(Exception exception, MetricGroup metricGroup, String metricGroupKey) {
+        throw new UnsupportedOperationException("This operation is not supported on StatsDErrorReporter");
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/manager/DaggerCounterManager.java

@@ -0,0 +1,47 @@
+package io.odpf.dagger.core.metrics.reporters.statsd.manager;
+
+import io.odpf.dagger.common.metrics.aspects.Aspects;
+import io.odpf.dagger.core.metrics.reporters.statsd.SerializedStatsDReporterSupplier;
+import io.odpf.dagger.core.metrics.reporters.statsd.measurement.Counter;
+import io.odpf.dagger.core.metrics.reporters.statsd.tags.StatsDTag;
+import io.odpf.depot.metrics.StatsDReporter;
+
+import java.util.ArrayList;
+
+public class DaggerCounterManager implements MeasurementManager, Counter {
+    private final StatsDReporter statsDReporter;
+    private String[] formattedTags;
+
+    public DaggerCounterManager(SerializedStatsDReporterSupplier statsDReporterSupplier) {
+        this.statsDReporter = statsDReporterSupplier.buildStatsDReporter();
+    }
+
+    @Override
+    public void register(StatsDTag[] tags) {
+        ArrayList<String> tagList = new ArrayList<>();
+        for (StatsDTag measurementTag : tags) {
+            tagList.add(measurementTag.getFormattedTag());
+        }
+        this.formattedTags = tagList.toArray(new String[0]);
+    }
+
+    @Override
+    public void increment(Aspects aspect) {
+        increment(aspect, 1L);
+    }
+
+    @Override
+    public void increment(Aspects aspect, long positiveCount) {
+        statsDReporter.captureCount(aspect.getValue(), positiveCount, formattedTags);
+    }
+
+    @Override
+    public void decrement(Aspects aspect) {
+        decrement(aspect, -1L);
+    }
+
+    @Override
+    public void decrement(Aspects aspect, long negativeCount) {
+        statsDReporter.captureCount(aspect.getValue(), negativeCount, formattedTags);
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/manager/DaggerGaugeManager.java

@@ -0,0 +1,32 @@
+package io.odpf.dagger.core.metrics.reporters.statsd.manager;
+
+import io.odpf.dagger.common.metrics.aspects.Aspects;
+import io.odpf.dagger.core.metrics.reporters.statsd.SerializedStatsDReporterSupplier;
+import io.odpf.dagger.core.metrics.reporters.statsd.measurement.Gauge;
+import io.odpf.dagger.core.metrics.reporters.statsd.tags.StatsDTag;
+import io.odpf.depot.metrics.StatsDReporter;
+
+import java.util.ArrayList;
+
+public class DaggerGaugeManager implements MeasurementManager, Gauge {
+    private final StatsDReporter statsDReporter;
+    private String[] formattedTags;
+
+    public DaggerGaugeManager(SerializedStatsDReporterSupplier statsDReporterSupplier) {
+        this.statsDReporter = statsDReporterSupplier.buildStatsDReporter();
+    }
+
+    @Override
+    public void register(StatsDTag[] tags) {
+        ArrayList<String> tagList = new ArrayList<>();
+        for (StatsDTag measurementTag : tags) {
+            tagList.add(measurementTag.getFormattedTag());
+        }
+        this.formattedTags = tagList.toArray(new String[0]);
+    }
+
+    @Override
+    public void markValue(Aspects aspect, int gaugeValue) {
+        statsDReporter.gauge(aspect.getValue(), gaugeValue, formattedTags);
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/manager/DaggerHistogramManager.java

@@ -0,0 +1,32 @@
+package io.odpf.dagger.core.metrics.reporters.statsd.manager;
+
+import io.odpf.dagger.common.metrics.aspects.Aspects;
+import io.odpf.dagger.core.metrics.reporters.statsd.SerializedStatsDReporterSupplier;
+import io.odpf.dagger.core.metrics.reporters.statsd.measurement.Histogram;
+import io.odpf.dagger.core.metrics.reporters.statsd.tags.StatsDTag;
+import io.odpf.depot.metrics.StatsDReporter;
+
+import java.util.ArrayList;
+
+public class DaggerHistogramManager implements MeasurementManager, Histogram {
+    private final StatsDReporter statsDReporter;
+    private String[] formattedTags;
+
+    public DaggerHistogramManager(SerializedStatsDReporterSupplier statsDReporterSupplier) {
+        this.statsDReporter = statsDReporterSupplier.buildStatsDReporter();
+    }
+
+    @Override
+    public void register(StatsDTag[] tags) {
+        ArrayList<String> tagList = new ArrayList<>();
+        for (StatsDTag measurementTag : tags) {
+            tagList.add(measurementTag.getFormattedTag());
+        }
+        this.formattedTags = tagList.toArray(new String[0]);
+    }
+
+    @Override
+    public void recordValue(Aspects aspect, long value) {
+        statsDReporter.captureHistogram(aspect.getValue(), value, formattedTags);
+    }
+}

dagger-core/src/main/java/io/odpf/dagger/core/metrics/reporters/statsd/manager/MeasurementManager.java

@@ -0,0 +1,9 @@
+package io.odpf.dagger.core.metrics.reporters.statsd.manager;
+
+import io.odpf.dagger.core.metrics.reporters.statsd.tags.StatsDTag;
+
+import java.io.Serializable;
+
+public interface MeasurementManager extends Serializable {
+    void register(StatsDTag[] tags);
+}