[GOBBLIN-2209] Emit GaaS Executor Otel Metrics (#4118)

Author: Blazer-007

gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java

@@ -934,7 +934,10 @@ public class ConfigurationKeys {
 
   // Opentelemetry based metrics reporting
   public static final String METRICS_REPORTING_OPENTELEMETRY_PREFIX = "metrics.reporting.opentelemetry.";
+  public static final String METRICS_REPORTING_OPENTELEMETRY_CLASSNAME = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "className";
+  public static final String DEFAULT_METRICS_REPORTING_OPENTELEMETRY_CLASSNAME = "org.apache.gobblin.metrics.OpenTelemetryMetrics";
   public static final String METRICS_REPORTING_OPENTELEMETRY_ENABLED = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "enabled";
+  public static final Boolean DEFAULT_METRICS_REPORTING_OPENTELEMETRY_ENABLED = false;
 
   public static final String METRICS_REPORTING_OPENTELEMETRY_LOGEXPORTER_ENABLED = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "logexporter.enabled";
 
@@ -943,14 +946,19 @@ public class ConfigurationKeys {
   public static final String METRICS_REPORTING_OPENTELEMETRY_LOGEXPORTER_CLASSNAME = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "logexporter.className";
 
   public static final String METRICS_REPORTING_OPENTELEMETRY_CONFIGS_PREFIX = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "configs.";
-  public static final Boolean DEFAULT_METRICS_REPORTING_OPENTELEMETRY_ENABLED = false;
 
   public static final String METRICS_REPORTING_OPENTELEMETRY_ENDPOINT = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "endpoint";
   public static final String METRICS_REPORTING_OPENTELEMETRY_FABRIC = METRICS_REPORTING_OPENTELEMETRY_CONFIGS_PREFIX + "fabric";
   // Headers to add to the OpenTelemetry HTTP Exporter, formatted as a JSON String with string keys and values
   public static final String METRICS_REPORTING_OPENTELEMETRY_HEADERS = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "headers";
 
-  public static final String METRICS_REPORTING_OPENTELEMETRY_INTERVAL_MILLIS = METRICS_CONFIGURATIONS_PREFIX + "interval.millis";
+  public static final String METRICS_REPORTING_OPENTELEMETRY_INTERVAL_MILLIS = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "interval.millis";
+  public static final String METRICS_REPORTING_OPENTELEMETRY_HISTOGRAM_MAX_BUCKETS = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "histogram.max.buckets";
+  public static final String METRICS_REPORTING_OPENTELEMETRY_HISTOGRAM_MAX_SCALE = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "histogram.max.scale";
+  // A comma-separated list of dimensions to add to the OpenTelemetry metrics
+  public static final String METRICS_REPORTING_OPENTELEMETRY_DIMENSIONS = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "dimensions";
+  public static final String METRICS_REPORTING_OPENTELEMETRY_GROUP_NAME = METRICS_REPORTING_OPENTELEMETRY_PREFIX + "group.name";
+  public static final String DEFAULT_METRICS_REPORTING_OPENTELEMETRY_GROUP_NAME = "org.apache.gobblin.metrics";
 
   /**
    * Rest server configuration properties.

gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/OpenTelemetryMetrics.java

@@ -32,6 +32,11 @@
 import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter;
 import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.metrics.Aggregation;
+import io.opentelemetry.sdk.metrics.InstrumentSelector;
+import io.opentelemetry.sdk.metrics.InstrumentType;
+import io.opentelemetry.sdk.metrics.View;
+import io.opentelemetry.sdk.metrics.internal.view.Base2ExponentialHistogramAggregation;
 import io.opentelemetry.sdk.metrics.SdkMeterProvider;
 import io.opentelemetry.sdk.metrics.export.AggregationTemporalitySelector;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
@@ -52,8 +57,10 @@
 @Slf4j
 public class OpenTelemetryMetrics extends OpenTelemetryMetricsBase {
 
-  private static OpenTelemetryMetrics GLOBAL_INSTANCE;
+  private static volatile OpenTelemetryMetrics GLOBAL_INSTANCE;
   private static final Long DEFAULT_OPENTELEMETRY_REPORTING_INTERVAL_MILLIS = 10000L;
+  private static final int DEFAULT_OPENTELEMETRY_HISTOGRAM_MAX_BUCKETS = 256;
+  private static final int DEFAULT_OPENTELEMETRY_HISTOGRAM_MAX_SCALE = 3;
 
   private OpenTelemetryMetrics(State state) {
     super(state);
@@ -94,7 +101,12 @@ protected MetricExporter initializeMetricExporter(State state) {
   public static OpenTelemetryMetrics getInstance(State state) {
     if (state.getPropAsBoolean(ConfigurationKeys.METRICS_REPORTING_OPENTELEMETRY_ENABLED,
         ConfigurationKeys.DEFAULT_METRICS_REPORTING_OPENTELEMETRY_ENABLED) && GLOBAL_INSTANCE == null) {
-      GLOBAL_INSTANCE = new OpenTelemetryMetrics(state);
+      synchronized (OpenTelemetryMetrics.class) {
+        if (GLOBAL_INSTANCE == null) {
+          log.info("Creating OpenTelemetryMetrics instance");
+          GLOBAL_INSTANCE = new OpenTelemetryMetrics(state);
+        }
+      }
     }
     return GLOBAL_INSTANCE;
   }
@@ -115,6 +127,13 @@ protected void initialize(State state) {
       }
       metricsResource = Resource.getDefault().merge(Resource.create(attributesBuilder.build()));
     }
+
+    Aggregation histogramAggregation = Base2ExponentialHistogramAggregation.create(
+        state.getPropAsInt(ConfigurationKeys.METRICS_REPORTING_OPENTELEMETRY_HISTOGRAM_MAX_BUCKETS,
+            DEFAULT_OPENTELEMETRY_HISTOGRAM_MAX_BUCKETS),
+        state.getPropAsInt(ConfigurationKeys.METRICS_REPORTING_OPENTELEMETRY_HISTOGRAM_MAX_SCALE,
+            DEFAULT_OPENTELEMETRY_HISTOGRAM_MAX_SCALE));
+
     SdkMeterProvider meterProvider = SdkMeterProvider.builder()
         .setResource(metricsResource)
         .registerMetricReader(
@@ -123,6 +142,9 @@ protected void initialize(State state) {
                     state.getPropAsLong(ConfigurationKeys.METRICS_REPORTING_OPENTELEMETRY_INTERVAL_MILLIS,
                         DEFAULT_OPENTELEMETRY_REPORTING_INTERVAL_MILLIS)))
                 .build())
+        .registerView(
+            InstrumentSelector.builder().setType(InstrumentType.HISTOGRAM).build(),
+            View.builder().setAggregation(histogramAggregation).build())
         .build();
 
     this.openTelemetry = OpenTelemetrySdk.builder().setMeterProvider(meterProvider).build();

gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/opentelemetry/GobblinOpenTelemetryMetrics.java

@@ -0,0 +1,56 @@
+/*
+ * 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.metrics.opentelemetry;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.Meter;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+
+@Getter
+@AllArgsConstructor
+public enum GobblinOpenTelemetryMetrics {
+  /**
+   * Metric to track the count of Gobblin Jobs for each of its state (GenerateWorkUnit, ProcessWorkUnit, CommitStep).
+   * Metric Unit: 1 represents each increment will add one data point to the counter.
+   * */
+  GOBBLIN_JOB_STATE("gobblin.job.state", "Gobblin job state counter", "1", OpenTelemetryMetricType.LONG_COUNTER),
+
+  /**
+   * Metric to track the latency of each Gobblin Job state (GenerateWorkUnit, ProcessWorkUnit, CommitStep).
+   * Metric Unit: seconds (s) represents the time taken for each state.
+   * */
+  GOBBLIN_JOB_STATE_LATENCY("gobblin.job.state.latency", "Gobblin job state latency", "s", OpenTelemetryMetricType.DOUBLE_HISTOGRAM);
+
+  private final String metricName;
+  private final String metricDescription;
+  private final String metricUnit;
+  private final OpenTelemetryMetricType metricType;
+
+  @SuppressWarnings("unchecked")
+  public <T extends OpenTelemetryMetric> T createMetric(Attributes attributes, Meter meter) {
+    return (T) this.metricType.getFactory().newMetric(this.metricName, this.metricDescription, this.metricUnit, attributes, meter);
+  }
+
+  @Override
+  public String toString() {
+    return String.format("Metric{name='%s', description='%s', unit='%s', type=%s}", metricName, metricDescription, metricUnit, metricType);
+  }
+
+}

gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/opentelemetry/GobblinOpenTelemetryMetricsConstants.java

@@ -0,0 +1,40 @@
+/*
+ * 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.metrics.opentelemetry;
+
+public class GobblinOpenTelemetryMetricsConstants {
+
+  public static class DimensionKeys {
+    public static final String STATE = "state";
+    public static final String CURR_STATE = "currState";
+  }
+
+  public static class DimensionValues {
+    public static final String GENERATE_WU = "generateWU";
+    public static final String PROCESS_WU = "processWU";
+    public static final String COMMIT_STEP = "commitStep";
+    public static final String JOB_START = "jobStart";
+    public static final String JOB_COMPLETE = "jobComplete";
+    public static final String GENERATE_WU_START = "generateWUStart";
+    public static final String GENERATE_WU_COMPLETE = "generateWUComplete";
+    public static final String PROCESS_WU_START = "processWUStart";
+    public static final String PROCESS_WU_COMPLETE = "processWUComplete";
+    public static final String COMMIT_STEP_START = "commitStepStart";
+    public static final String COMMIT_STEP_COMPLETE = "commitStepComplete";
+  }
+}

gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/opentelemetry/OpenTelemetryDoubleHistogram.java

@@ -0,0 +1,87 @@
+/*
+ * 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.metrics.opentelemetry;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+
+
+/**
+ * Implementation of {@link OpenTelemetryMetric} that wraps an OpenTelemetry {@link DoubleHistogram}.
+ *
+ * <p>This class provides a histogram for recording the distribution of double values.
+ * It supports recording values with optional additional attributes that can be merged with base attributes.</p>
+ *
+ */
+@Slf4j
+@AllArgsConstructor
+public class OpenTelemetryDoubleHistogram implements OpenTelemetryMetric {
+  private String name;
+  private Attributes baseAttributes;
+  private DoubleHistogram doubleHistogram;
+
+  /**
+   * Records the specified value in the histogram with the base attributes.
+   *
+   * @param value the double value to record in the histogram
+   */
+  public void record(double value) {
+    log.debug("Emitting double histogram metric: {}, value: {}, attributes: {}", this.name, value, this.baseAttributes);
+    this.doubleHistogram.record(value, this.baseAttributes);
+  }
+
+  /**
+   * Records the specified value in the histogram with a combination of base attributes and additional attributes.
+   *
+   * @param value the double value to record in the histogram
+   * @param additionalAttributes the additional attributes to be merged with base attributes
+   */
+  public void record(double value, Attributes additionalAttributes) {
+    log.debug("Emitting double histogram metric: {}, value: {}, base attributes: {}, additional attributes: {}",
+        this.name, value, this.baseAttributes, additionalAttributes);
+    this.doubleHistogram.record(value, OpenTelemetryHelper.mergeAttributes(this.baseAttributes, additionalAttributes));
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public String getMetricName() {
+    return this.name;
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public OpenTelemetryMetricType getMetricType() {
+    return OpenTelemetryMetricType.DOUBLE_HISTOGRAM;
+  }
+
+  /**
+   * Returns a string representation of this histogram with its name.
+   */
+  @Override
+  public String toString() {
+    return "OpenTelemetryDoubleHistogram{name='" + name + "'}";
+  }
+
+}

gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/opentelemetry/OpenTelemetryHelper.java

@@ -0,0 +1,94 @@
+/*
+ * 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.metrics.opentelemetry;
+
+import java.util.Map;
+
+import lombok.experimental.UtilityClass;
+import org.apache.commons.lang3.StringUtils;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+
+/**
+ * Utility class for OpenTelemetry related operations.
+ *
+ * <p>Provides methods to handle OpenTelemetry attributes, including merging multiple
+ * {@link Attributes} instances and converting maps to {@link Attributes}.
+ */
+@UtilityClass
+public class OpenTelemetryHelper {
+
+  private static final String DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE = "UNKNOWN";
+
+  /**
+   * Returns the provided attribute value when it is non-null and non-empty;
+   * otherwise returns the default OpenTelemetry attribute placeholder.
+   *
+   * @param value candidate attribute value to check
+   * @return the original value if not empty, or DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE otherwise
+   */
+  public static String getOrDefaultOpenTelemetryAttrValue(String value) {
+    return StringUtils.defaultIfBlank(value, DEFAULT_OPENTELEMETRY_ATTRIBUTE_VALUE);
+  }
+
+  /**
+   * Merges multiple {@link Attributes} instances into a single {@link Attributes}.
+   *
+   * <p>Any {@code null} or empty ({@link Attributes#isEmpty()}) instances are ignored.
+   * The resulting {@link Attributes} contains all key-value pairs from the
+   * provided non-null, non-empty inputs in the order they are given.
+   * For duplicate keys, the last occurrence in the array will take precedence.
+   *
+   * @param attributesArray array of {@link Attributes} to merge; may contain {@code null} or empty entries
+   * @return a new {@link Attributes} instance containing all entries from the non-null,
+   *         non-empty inputs; never {@code null}
+   */
+  public static Attributes mergeAttributes(Attributes... attributesArray) {
+    AttributesBuilder builder = Attributes.builder();
+    for (Attributes attrs : attributesArray) {
+      if (attrs != null && !attrs.isEmpty()) {
+        builder.putAll(attrs);
+      }
+    }
+    return builder.build();
+  }
+
+  /**
+   * Converts a map of string attributes to an OpenTelemetry {@link Attributes} instance.
+   *
+   * <p>Each entry in the map is converted to an OpenTelemetry attribute, using
+   * {@link #getOrDefaultOpenTelemetryAttrValue(String)} to handle empty values.
+   *
+   * @param attributes map of string attributes to convert; may be {@code null}
+   * @return a new {@link Attributes} instance containing the converted attributes;
+   *         never {@code null}
+   */
+  public static Attributes toOpenTelemetryAttributes(Map<String, String> attributes) {
+    AttributesBuilder builder = Attributes.builder();
+    if (attributes != null) {
+      for (Map.Entry<String, String> entry : attributes.entrySet()) {
+        String key = entry.getKey();
+        String value = getOrDefaultOpenTelemetryAttrValue(entry.getValue());
+        builder.put(key, value);
+      }
+    }
+    return builder.build();
+  }
+
+}