Skip to content

Commit 395b02b

Browse files
m-nagarajanm-nagarajan
committed
add async gauge to Otel
1 parent 8fc4ecc commit 395b02b

16 files changed

Lines changed: 570 additions & 110 deletions

File tree

‎internal/venice-client-common/src/main/java/com/linkedin/venice/stats/VeniceOpenTelemetryMetricsRepository.java‎

Lines changed: 52 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,15 @@
1616
import io.opentelemetry.api.OpenTelemetry;
1717
import io.opentelemetry.api.common.Attributes;
1818
import io.opentelemetry.api.common.AttributesBuilder;
19-
import io.opentelemetry.api.metrics.DoubleGauge;
20-
import io.opentelemetry.api.metrics.DoubleGaugeBuilder;
2119
import io.opentelemetry.api.metrics.DoubleHistogram;
2220
import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
2321
import io.opentelemetry.api.metrics.LongCounter;
2422
import io.opentelemetry.api.metrics.LongCounterBuilder;
23+
import io.opentelemetry.api.metrics.LongGauge;
24+
import io.opentelemetry.api.metrics.LongGaugeBuilder;
2525
import io.opentelemetry.api.metrics.Meter;
2626
import io.opentelemetry.api.metrics.MeterProvider;
27+
import io.opentelemetry.api.metrics.ObservableLongGauge;
2728
import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter;
2829
import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder;
2930
import io.opentelemetry.sdk.OpenTelemetrySdk;
@@ -48,6 +49,8 @@
4849
import java.util.Map;
4950
import java.util.Set;
5051
import java.util.concurrent.TimeUnit;
52+
import java.util.function.LongSupplier;
53+
import javax.annotation.Nonnull;
5154
import org.apache.logging.log4j.LogManager;
5255
import org.apache.logging.log4j.Logger;
5356

@@ -163,7 +166,8 @@ public VeniceOpenTelemetryMetricsRepository(VeniceMetricsConfig metricsConfig) {
163166
*/
164167
private final VeniceConcurrentHashMap<String, DoubleHistogram> histogramMap = new VeniceConcurrentHashMap<>();
165168
private final VeniceConcurrentHashMap<String, LongCounter> counterMap = new VeniceConcurrentHashMap<>();
166-
private final VeniceConcurrentHashMap<String, DoubleGauge> gaugeMap = new VeniceConcurrentHashMap<>();
169+
private final VeniceConcurrentHashMap<String, LongGauge> gaugeMap = new VeniceConcurrentHashMap<>();
170+
private final VeniceConcurrentHashMap<String, ObservableLongGauge> asyncGaugeMap = new VeniceConcurrentHashMap<>();
167171

168172
MetricExporter getOtlpHttpMetricExporter(VeniceMetricsConfig metricsConfig) {
169173
OtlpHttpMetricExporterBuilder exporterBuilder =
@@ -282,20 +286,52 @@ public LongCounter createCounter(MetricEntity metricEntity) {
282286
});
283287
}
284288

285-
public DoubleGauge createGuage(MetricEntity metricEntity) {
289+
public LongGauge createGuage(MetricEntity metricEntity) {
286290
if (!emitOpenTelemetryMetrics()) {
287291
return null;
288292
}
289293
return gaugeMap.computeIfAbsent(metricEntity.getMetricName(), key -> {
290294
String fullMetricName = getFullMetricName(metricEntity);
291-
DoubleGaugeBuilder builder = meter.gaugeBuilder(fullMetricName)
295+
LongGaugeBuilder builder = meter.gaugeBuilder(fullMetricName)
292296
.setUnit(metricEntity.getUnit().name())
293-
.setDescription(getMetricDescription(metricEntity, metricsConfig));
297+
.setDescription(getMetricDescription(metricEntity, metricsConfig))
298+
.ofLongs();
294299
return builder.build();
295300
});
296301
}
297302

298-
public Object createInstrument(MetricEntity metricEntity) {
303+
/**
304+
* Asynchronous gauge that will call the supplier during metrics collection.
305+
* This is useful for metrics that are not updated frequently or require expensive computation.
306+
* For now, the attributes are passed in as a parameter while creating the gauge.
307+
*/
308+
public ObservableLongGauge createAsyncGauge(
309+
MetricEntity metricEntity,
310+
@Nonnull LongSupplier asyncCallback,
311+
@Nonnull Attributes attributes) {
312+
if (!emitOpenTelemetryMetrics()) {
313+
return null;
314+
}
315+
return asyncGaugeMap.computeIfAbsent(metricEntity.getMetricName(), key -> {
316+
String fullMetricName = getFullMetricName(metricEntity);
317+
LongGaugeBuilder builder = meter.gaugeBuilder(fullMetricName)
318+
.setUnit(metricEntity.getUnit().name())
319+
.setDescription(getMetricDescription(metricEntity, metricsConfig))
320+
.ofLongs();
321+
322+
return builder.buildWithCallback(measurement -> {
323+
long v;
324+
try {
325+
v = asyncCallback.getAsLong();
326+
} catch (Exception e) {
327+
return;
328+
}
329+
measurement.record(v, attributes);
330+
});
331+
});
332+
}
333+
334+
public Object createInstrument(MetricEntity metricEntity, LongSupplier asyncCallback, Attributes attributes) {
299335
MetricType metricType = metricEntity.getMetricType();
300336
switch (metricType) {
301337
case HISTOGRAM:
@@ -304,14 +340,23 @@ public Object createInstrument(MetricEntity metricEntity) {
304340

305341
case COUNTER:
306342
return createCounter(metricEntity);
343+
307344
case GAUGE:
308345
return createGuage(metricEntity);
309346

347+
case ASYNC_GAUGE:
348+
return createAsyncGauge(metricEntity, asyncCallback, attributes);
349+
310350
default:
311351
throw new VeniceException("Unknown metric type: " + metricType);
312352
}
313353
}
314354

355+
@VisibleForTesting
356+
public Object createInstrument(MetricEntity metricEntity) {
357+
return createInstrument(metricEntity, null, null);
358+
}
359+
315360
public String getDimensionName(VeniceMetricsDimensions dimension) {
316361
return dimension.getDimensionName(getMetricFormat());
317362
}

‎internal/venice-client-common/src/main/java/com/linkedin/venice/stats/metrics/MetricEntityState.java‎

Lines changed: 77 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,17 +5,19 @@
55
import com.linkedin.venice.stats.dimensions.VeniceMetricsDimensions;
66
import io.opentelemetry.api.common.AttributeKey;
77
import io.opentelemetry.api.common.Attributes;
8-
import io.opentelemetry.api.metrics.DoubleGauge;
98
import io.opentelemetry.api.metrics.DoubleHistogram;
109
import io.opentelemetry.api.metrics.LongCounter;
10+
import io.opentelemetry.api.metrics.LongGauge;
1111
import io.tehuti.metrics.MeasurableStat;
1212
import io.tehuti.metrics.Sensor;
13+
import io.tehuti.metrics.stats.AsyncGauge;
1314
import java.util.Arrays;
1415
import java.util.HashSet;
1516
import java.util.List;
1617
import java.util.Map;
1718
import java.util.Objects;
1819
import java.util.Set;
20+
import java.util.function.LongSupplier;
1921
import org.apache.logging.log4j.LogManager;
2022
import org.apache.logging.log4j.Logger;
2123

@@ -55,11 +57,31 @@ public MetricEntityState(
5557
TehutiSensorRegistrationFunction registerTehutiSensorFn,
5658
TehutiMetricNameEnum tehutiMetricNameEnum,
5759
List<MeasurableStat> tehutiMetricStats) {
60+
this(
61+
metricEntity,
62+
otelRepository,
63+
baseDimensionsMap,
64+
registerTehutiSensorFn,
65+
tehutiMetricNameEnum,
66+
tehutiMetricStats,
67+
null,
68+
null);
69+
}
70+
71+
public MetricEntityState(
72+
MetricEntity metricEntity,
73+
VeniceOpenTelemetryMetricsRepository otelRepository,
74+
Map<VeniceMetricsDimensions, String> baseDimensionsMap,
75+
TehutiSensorRegistrationFunction registerTehutiSensorFn,
76+
TehutiMetricNameEnum tehutiMetricNameEnum,
77+
List<MeasurableStat> tehutiMetricStats,
78+
LongSupplier asyncCallback,
79+
Attributes asyncAttributes) {
5880
this.metricEntity = metricEntity;
5981
this.emitOpenTelemetryMetrics = otelRepository != null && otelRepository.emitOpenTelemetryMetrics();
6082
this.otelRepository = otelRepository;
6183
this.baseDimensionsMap = baseDimensionsMap;
62-
createMetric(tehutiMetricNameEnum, tehutiMetricStats, registerTehutiSensorFn);
84+
createMetric(tehutiMetricNameEnum, tehutiMetricStats, registerTehutiSensorFn, asyncCallback, asyncAttributes);
6385
}
6486

6587
public void setOtelMetric(Object otelMetric) {
@@ -78,12 +100,55 @@ public interface TehutiSensorRegistrationFunction {
78100
Sensor register(String sensorName, MeasurableStat... stats);
79101
}
80102

103+
/**
104+
* Validates the tehuti metrics stats against the otel metric type for a given metric entity.
105+
* If tehutiMetricStats contains AsyncGauge, then the metric type should be ASYNC_GAUGE and tehutiMetricStats
106+
* should contain only one stat. If tehutiMetricStats does not contain AsyncGauge, then the metric type should not
107+
* be ASYNC_GAUGE.
108+
*
109+
* @param tehutiMetricStats the tehuti metrics stats for the given metric entity.
110+
*/
111+
private void validateMetric(List<MeasurableStat> tehutiMetricStats, LongSupplier asyncCallback) {
112+
if (asyncCallback != null && !metricEntity.getMetricType().isAsyncMetric()) {
113+
throw new IllegalArgumentException(
114+
"Async callback is provided, but the metric type is not async for metric: " + metricEntity.getMetricName());
115+
} else if (asyncCallback == null && metricEntity.getMetricType().isAsyncMetric()) {
116+
throw new IllegalArgumentException(
117+
"Async callback is not provided, but the metric type is async for metric: " + metricEntity.getMetricName());
118+
}
119+
120+
// ASYNC_GAUGE specific: If both tehuti and Otel are present, validate if all are nothing is async
121+
if (tehutiMetricStats == null || tehutiMetricStats.isEmpty()) {
122+
return;
123+
}
124+
// if tehutiMetricStats has AsyncGauge() then the metric type should be ASYNC_GAUGE
125+
if (tehutiMetricStats.stream().anyMatch(stat -> stat instanceof AsyncGauge)) {
126+
if (tehutiMetricStats.size() > 1) {
127+
throw new IllegalArgumentException(
128+
"Tehuti metric stats contains AsyncGauge, but it should be the only stat for metric: "
129+
+ metricEntity.getMetricName());
130+
}
131+
if (metricEntity.getMetricType() != MetricType.ASYNC_GAUGE) {
132+
throw new IllegalArgumentException(
133+
"Tehuti metric stats contains AsyncGauge, but the otel metric type is not ASYNC_GAUGE for metric: "
134+
+ metricEntity.getMetricName());
135+
}
136+
} else if (metricEntity.getMetricType() == MetricType.ASYNC_GAUGE) {
137+
throw new IllegalArgumentException(
138+
"Tehuti metric stats does not contain AsyncGauge, but the otel metric type is ASYNC_GAUGE for metric: "
139+
+ metricEntity.getMetricName());
140+
}
141+
}
142+
81143
public void createMetric(
82144
TehutiMetricNameEnum tehutiMetricNameEnum,
83145
List<MeasurableStat> tehutiMetricStats,
84-
TehutiSensorRegistrationFunction registerTehutiSensorFn) {
146+
TehutiSensorRegistrationFunction registerTehutiSensorFn,
147+
LongSupplier asyncCallback,
148+
Attributes asyncAttributes) {
149+
validateMetric(tehutiMetricStats, asyncCallback);
85150
if (emitOpenTelemetryMetrics()) {
86-
setOtelMetric(otelRepository.createInstrument(this.metricEntity));
151+
setOtelMetric(otelRepository.createInstrument(this.metricEntity, asyncCallback, asyncAttributes));
87152
}
88153
// tehuti metric
89154
if (tehutiMetricStats != null && !tehutiMetricStats.isEmpty()) {
@@ -108,7 +173,10 @@ public void recordOtelMetric(double value, Attributes attributes) {
108173
((LongCounter) otelMetric).add((long) value, attributes);
109174
break;
110175
case GAUGE:
111-
((DoubleGauge) otelMetric).set((long) value, attributes);
176+
((LongGauge) otelMetric).set((long) value, attributes);
177+
break;
178+
case ASYNC_GAUGE:
179+
// Async gauge is not recorded directly, it is updated by the callback function.
112180
break;
113181
default:
114182
throw new IllegalArgumentException("Unsupported metric type: " + metricType);
@@ -128,6 +196,10 @@ final void record(long value, Attributes attributes) {
128196
}
129197

130198
final void record(double value, Attributes attributes) {
199+
if (metricEntity.getMetricType() == MetricType.ASYNC_GAUGE) {
200+
// Async gauge metrics are not recorded directly, they are updated by the callback function.
201+
return;
202+
}
131203
recordOtelMetric(value, attributes);
132204
recordTehutiMetric(value);
133205
}

‎internal/venice-client-common/src/main/java/com/linkedin/venice/stats/metrics/MetricEntityStateBase.java‎

Lines changed: 78 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@
77
import java.util.Collections;
88
import java.util.List;
99
import java.util.Map;
10+
import java.util.function.LongSupplier;
11+
import javax.annotation.Nonnull;
1012
import org.apache.commons.lang.Validate;
1113

1214

@@ -18,16 +20,34 @@
1820
public class MetricEntityStateBase extends MetricEntityState {
1921
private final Attributes attributes;
2022

21-
/** should not be called directly, call {@link #create} instead */
23+
/** should not be called directly, call {@link #createAsyncMetric} instead */
2224
private MetricEntityStateBase(
2325
MetricEntity metricEntity,
2426
VeniceOpenTelemetryMetricsRepository otelRepository,
2527
Map<VeniceMetricsDimensions, String> baseDimensionsMap,
2628
Attributes baseAttributes) {
27-
this(metricEntity, otelRepository, null, null, Collections.EMPTY_LIST, baseDimensionsMap, baseAttributes);
29+
this(metricEntity, otelRepository, baseDimensionsMap, baseAttributes, null);
2830
}
2931

30-
/** should not be called directly, call {@link #create} instead */
32+
/** should not be called directly, call {@link #createAsyncMetric} instead */
33+
private MetricEntityStateBase(
34+
MetricEntity metricEntity,
35+
VeniceOpenTelemetryMetricsRepository otelRepository,
36+
Map<VeniceMetricsDimensions, String> baseDimensionsMap,
37+
Attributes baseAttributes,
38+
LongSupplier asyncCallback) {
39+
this(
40+
metricEntity,
41+
otelRepository,
42+
null,
43+
null,
44+
Collections.EMPTY_LIST,
45+
baseDimensionsMap,
46+
baseAttributes,
47+
asyncCallback);
48+
}
49+
50+
/** should not be called directly, call {@link #createAsyncMetric} instead */
3151
private MetricEntityStateBase(
3252
MetricEntity metricEntity,
3353
VeniceOpenTelemetryMetricsRepository otelRepository,
@@ -36,13 +56,36 @@ private MetricEntityStateBase(
3656
List<MeasurableStat> tehutiMetricStats,
3757
Map<VeniceMetricsDimensions, String> baseDimensionsMap,
3858
Attributes baseAttributes) {
59+
this(
60+
metricEntity,
61+
otelRepository,
62+
registerTehutiSensorFn,
63+
tehutiMetricNameEnum,
64+
tehutiMetricStats,
65+
baseDimensionsMap,
66+
baseAttributes,
67+
null);
68+
}
69+
70+
/** should not be called directly, call {@link #createAsyncMetric} instead */
71+
private MetricEntityStateBase(
72+
MetricEntity metricEntity,
73+
VeniceOpenTelemetryMetricsRepository otelRepository,
74+
TehutiSensorRegistrationFunction registerTehutiSensorFn,
75+
TehutiMetricNameEnum tehutiMetricNameEnum,
76+
List<MeasurableStat> tehutiMetricStats,
77+
Map<VeniceMetricsDimensions, String> baseDimensionsMap,
78+
Attributes baseAttributes,
79+
LongSupplier asyncCallback) {
3980
super(
4081
metricEntity,
4182
otelRepository,
4283
baseDimensionsMap,
4384
registerTehutiSensorFn,
4485
tehutiMetricNameEnum,
45-
tehutiMetricStats);
86+
tehutiMetricStats,
87+
asyncCallback,
88+
baseAttributes);
4689
validateRequiredDimensions(metricEntity, baseAttributes, baseDimensionsMap);
4790
// directly using the Attributes as multiple MetricEntityState can reuse the same base attributes object.
4891
// If we want to fully abstract the Attribute creation inside these classes, we can create it here instead.
@@ -63,6 +106,16 @@ public static MetricEntityStateBase create(
63106
return new MetricEntityStateBase(metricEntity, otelRepository, baseDimensionsMap, baseAttributes);
64107
}
65108

109+
/** Factory method to keep the API consistent with other subclasses like {@link MetricEntityStateOneEnum} */
110+
public static MetricEntityStateBase createAsyncMetric(
111+
MetricEntity metricEntity,
112+
VeniceOpenTelemetryMetricsRepository otelRepository,
113+
Map<VeniceMetricsDimensions, String> baseDimensionsMap,
114+
Attributes baseAttributes,
115+
LongSupplier asyncCallback) {
116+
return new MetricEntityStateBase(metricEntity, otelRepository, baseDimensionsMap, baseAttributes, asyncCallback);
117+
}
118+
66119
/** Overloaded Factory method for constructor with Tehuti parameters */
67120
public static MetricEntityStateBase create(
68121
MetricEntity metricEntity,
@@ -82,6 +135,27 @@ public static MetricEntityStateBase create(
82135
baseAttributes);
83136
}
84137

138+
/** Overloaded Factory method for constructor with Tehuti parameters and async callback */
139+
public static MetricEntityStateBase createAsyncMetric(
140+
MetricEntity metricEntity,
141+
VeniceOpenTelemetryMetricsRepository otelRepository,
142+
TehutiSensorRegistrationFunction registerTehutiSensorFn,
143+
TehutiMetricNameEnum tehutiMetricNameEnum,
144+
List<MeasurableStat> tehutiMetricStats,
145+
Map<VeniceMetricsDimensions, String> baseDimensionsMap,
146+
Attributes baseAttributes,
147+
@Nonnull LongSupplier asyncCallback) {
148+
return new MetricEntityStateBase(
149+
metricEntity,
150+
otelRepository,
151+
registerTehutiSensorFn,
152+
tehutiMetricNameEnum,
153+
tehutiMetricStats,
154+
baseDimensionsMap,
155+
baseAttributes,
156+
asyncCallback);
157+
}
158+
85159
public void record(long value) {
86160
super.record(value, attributes);
87161
}

0 commit comments

Comments
 (0)