Skip to content

Commit

Permalink
Adding gfe_latencies metric to built-in metrics
Browse files Browse the repository at this point in the history
  • Loading branch information
surbhigarg92 committed Nov 20, 2024
1 parent ea1ebad commit 7383362
Show file tree
Hide file tree
Showing 8 changed files with 205 additions and 36 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,9 @@
public class BuiltInMetricsConstant {

public static final String METER_NAME = "spanner.googleapis.com/internal/client";

public static final String GAX_METER_NAME = OpenTelemetryMetricsRecorder.GAX_METER_NAME;

static final String SPANNER_METER_NAME = "spanner-java";
static final String GFE_LATENCIES_NAME = "gfe_latencies";
static final String OPERATION_LATENCIES_NAME = "operation_latencies";
static final String ATTEMPT_LATENCIES_NAME = "attempt_latencies";
static final String OPERATION_LATENCY_NAME = "operation_latency";
Expand Down Expand Up @@ -114,27 +114,39 @@ static Map<InstrumentSelector, View> getAllViews() {
ImmutableMap.Builder<InstrumentSelector, View> views = ImmutableMap.builder();
defineView(
views,
BuiltInMetricsConstant.GAX_METER_NAME,
BuiltInMetricsConstant.OPERATION_LATENCY_NAME,
BuiltInMetricsConstant.OPERATION_LATENCIES_NAME,
BuiltInMetricsConstant.AGGREGATION_WITH_MILLIS_HISTOGRAM,
InstrumentType.HISTOGRAM,
"ms");
defineView(
views,
BuiltInMetricsConstant.GAX_METER_NAME,
BuiltInMetricsConstant.ATTEMPT_LATENCY_NAME,
BuiltInMetricsConstant.ATTEMPT_LATENCIES_NAME,
BuiltInMetricsConstant.AGGREGATION_WITH_MILLIS_HISTOGRAM,
InstrumentType.HISTOGRAM,
"ms");
defineView(
views,
BuiltInMetricsConstant.SPANNER_METER_NAME,
BuiltInMetricsConstant.GFE_LATENCIES_NAME,
BuiltInMetricsConstant.GFE_LATENCIES_NAME,
BuiltInMetricsConstant.AGGREGATION_WITH_MILLIS_HISTOGRAM,
InstrumentType.HISTOGRAM,
"ms");
defineView(
views,
BuiltInMetricsConstant.GAX_METER_NAME,
BuiltInMetricsConstant.OPERATION_COUNT_NAME,
BuiltInMetricsConstant.OPERATION_COUNT_NAME,
Aggregation.sum(),
InstrumentType.COUNTER,
"1");
defineView(
views,
BuiltInMetricsConstant.GAX_METER_NAME,
BuiltInMetricsConstant.ATTEMPT_COUNT_NAME,
BuiltInMetricsConstant.ATTEMPT_COUNT_NAME,
Aggregation.sum(),
Expand All @@ -145,6 +157,7 @@ static Map<InstrumentSelector, View> getAllViews() {

private static void defineView(
ImmutableMap.Builder<InstrumentSelector, View> viewMap,
String meterName,
String metricName,
String metricViewName,
Aggregation aggregation,
Expand All @@ -153,7 +166,7 @@ private static void defineView(
InstrumentSelector selector =
InstrumentSelector.builder()
.setName(BuiltInMetricsConstant.METER_NAME + '/' + metricName)
.setMeterName(BuiltInMetricsConstant.GAX_METER_NAME)
.setMeterName(meterName)
.setType(type)
.setUnit(unit)
.build();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@
import com.google.cloud.opentelemetry.detection.AttributeKeys;
import com.google.cloud.opentelemetry.detection.DetectedPlatform;
import com.google.cloud.opentelemetry.detection.GCPPlatformDetector;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import io.opentelemetry.api.OpenTelemetry;
Expand All @@ -42,6 +44,7 @@
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;
Expand All @@ -57,6 +60,9 @@ final class BuiltInOpenTelemetryMetricsProvider {

private OpenTelemetry openTelemetry;

private final Cache<String, Map<String, String>> clientAttributesCache =
CacheBuilder.newBuilder().maximumSize(1000).build();

private BuiltInOpenTelemetryMetricsProvider() {}

OpenTelemetry getOrCreateOpenTelemetry(String projectId, @Nullable Credentials credentials) {
Expand All @@ -78,16 +84,29 @@ OpenTelemetry getOrCreateOpenTelemetry(String projectId, @Nullable Credentials c
}
}

Map<String, String> createClientAttributes(String projectId, String client_name) {
Map<String, String> clientAttributes = new HashMap<>();
clientAttributes.put(LOCATION_ID_KEY.getKey(), detectClientLocation());
clientAttributes.put(PROJECT_ID_KEY.getKey(), projectId);
clientAttributes.put(INSTANCE_CONFIG_ID_KEY.getKey(), "unknown");
clientAttributes.put(CLIENT_NAME_KEY.getKey(), client_name);
String clientUid = getDefaultTaskValue();
clientAttributes.put(CLIENT_UID_KEY.getKey(), clientUid);
clientAttributes.put(CLIENT_HASH_KEY.getKey(), generateClientHash(clientUid));
return clientAttributes;
Map<String, String> createOrGetClientAttributes(String projectId, String client_name) {
try {
String key = projectId + client_name;
return clientAttributesCache.get(
key,
() -> {
Map<String, String> clientAttributes = new HashMap<>();
clientAttributes.put(LOCATION_ID_KEY.getKey(), detectClientLocation());
clientAttributes.put(PROJECT_ID_KEY.getKey(), projectId);
clientAttributes.put(INSTANCE_CONFIG_ID_KEY.getKey(), "unknown");
clientAttributes.put(CLIENT_NAME_KEY.getKey(), client_name);
String clientUid = getDefaultTaskValue();
clientAttributes.put(CLIENT_UID_KEY.getKey(), clientUid);
clientAttributes.put(CLIENT_HASH_KEY.getKey(), generateClientHash(clientUid));
return clientAttributes;
});
} catch (ExecutionException executionException) {
logger.log(
Level.WARNING,
"Unable to get Client Attributes for client side metrics, will skip exporting client side metrics",
executionException);
return null;
}
}

/**
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
/*
* Copyright 2024 Google LLC
*
* Licensed 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 com.google.cloud.spanner;

import com.google.api.gax.core.GaxProperties;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.common.AttributesBuilder;
import io.opentelemetry.api.metrics.DoubleHistogram;
import io.opentelemetry.api.metrics.Meter;
import java.util.HashMap;
import java.util.Map;

/** OpenTelemetry implementation of recording built in metrics. */
public class BuiltInOpenTelemetryMetricsRecorder {

private final DoubleHistogram gfeLatencyRecorder;
private final Map<String, String> attributes = new HashMap<>();

/**
* Creates the following instruments for the following metrics:
*
* <ul>
* <li>GFE Latency: Histogram
* </ul>
*
* @param openTelemetry OpenTelemetry instance
*/
public BuiltInOpenTelemetryMetricsRecorder(
OpenTelemetry openTelemetry, Map<String, String> clientAttributes) {
if (openTelemetry != null && clientAttributes != null) {
gfeLatencyRecorder = null;
return;
}
Meter meter =
openTelemetry
.meterBuilder(BuiltInMetricsConstant.SPANNER_METER_NAME)
.setInstrumentationVersion(GaxProperties.getLibraryVersion(getClass()))
.build();
this.gfeLatencyRecorder =
meter
.histogramBuilder(
BuiltInMetricsConstant.METER_NAME + '/' + BuiltInMetricsConstant.GFE_LATENCIES_NAME)
.setDescription(
"Latency between Google's network receiving an RPC and reading back the first byte of the response")
.setUnit("ms")
.build();
this.attributes.putAll(clientAttributes);
}

/**
* Record the latency between Google's network receiving an RPC and reading back the first byte of
* the response. Data is stored in a Histogram.
*
* @param gfeLatency Attempt Latency in ms
* @param attributes Map of the attributes to store
*/
public void recordGFELatency(double gfeLatency, Map<String, String> attributes) {
if (gfeLatencyRecorder != null) {
this.attributes.putAll(attributes);
gfeLatencyRecorder.record(gfeLatency, toOtelAttributes(this.attributes));
}
}

@VisibleForTesting
Attributes toOtelAttributes(Map<String, String> attributes) {
Preconditions.checkNotNull(attributes, "Attributes map cannot be null");
AttributesBuilder attributesBuilder = Attributes.builder();
attributes.forEach(attributesBuilder::put);
return attributesBuilder.build();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -1680,6 +1680,18 @@ public OpenTelemetry getOpenTelemetry() {
}
}

/** Returns an instance of OpenTelemetry object for Built-in Client metrics. */
public OpenTelemetry getBuiltInMetricsOpenTelemetry() {
return this.builtInOpenTelemetryMetricsProvider.getOrCreateOpenTelemetry(
this.getProjectId(), getCredentials());
}

/** Returns attributes for an instance of Built-in Client metrics. */
public Map<String, String> getBuiltInMetricsClientAttributes() {
return builtInOpenTelemetryMetricsProvider.createOrGetClientAttributes(
this.getProjectId(), "spanner-java/" + GaxProperties.getLibraryVersion(getClass()));
}

@Override
public ApiTracerFactory getApiTracerFactory() {
return createApiTracerFactory(false, false);
Expand Down Expand Up @@ -1729,11 +1741,13 @@ private ApiTracerFactory createMetricsApiTracerFactory() {
this.builtInOpenTelemetryMetricsProvider.getOrCreateOpenTelemetry(
this.getProjectId(), getCredentials());

return openTelemetry != null
Map<String, String> clientAttributes =
builtInOpenTelemetryMetricsProvider.createOrGetClientAttributes(
this.getProjectId(), "spanner-java/" + GaxProperties.getLibraryVersion(getClass()));
return openTelemetry != null && clientAttributes != null
? new MetricsTracerFactory(
new OpenTelemetryMetricsRecorder(openTelemetry, BuiltInMetricsConstant.METER_NAME),
builtInOpenTelemetryMetricsProvider.createClientAttributes(
this.getProjectId(), "spanner-java/" + GaxProperties.getLibraryVersion(getClass())))
clientAttributes)
: null;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -357,6 +357,8 @@ public GapicSpannerRpc(final SpannerOptions options) {
options.getInterceptorProvider(),
SpannerInterceptorProvider.createDefault(
options.getOpenTelemetry(),
options.getBuiltInMetricsOpenTelemetry(),
options.getBuiltInMetricsClientAttributes(),
(() -> directPathEnabledSupplier.get()))))
// This sets the trace context headers.
.withTraceContext(endToEndTracingEnabled, options.getOpenTelemetry())
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@

import com.google.api.gax.tracing.ApiTracer;
import com.google.cloud.spanner.BuiltInMetricsConstant;
import com.google.cloud.spanner.BuiltInOpenTelemetryMetricsRecorder;
import com.google.cloud.spanner.CompositeTracer;
import com.google.cloud.spanner.SpannerExceptionFactory;
import com.google.cloud.spanner.SpannerRpcMetrics;
Expand Down Expand Up @@ -94,12 +95,17 @@ class HeaderInterceptor implements ClientInterceptor {
private static final Level LEVEL = Level.INFO;
private final SpannerRpcMetrics spannerRpcMetrics;

private final BuiltInOpenTelemetryMetricsRecorder builtInOpenTelemetryMetricsRecorder;

private final Supplier<Boolean> directPathEnabledSupplier;

HeaderInterceptor(
SpannerRpcMetrics spannerRpcMetrics, Supplier<Boolean> directPathEnabledSupplier) {
SpannerRpcMetrics spannerRpcMetrics,
BuiltInOpenTelemetryMetricsRecorder builtInOpenTelemetryMetricsRecorder,
Supplier<Boolean> directPathEnabledSupplier) {
this.spannerRpcMetrics = spannerRpcMetrics;
this.directPathEnabledSupplier = directPathEnabledSupplier;
this.builtInOpenTelemetryMetricsRecorder = builtInOpenTelemetryMetricsRecorder;
}

@Override
Expand All @@ -118,17 +124,22 @@ public void start(Listener<RespT> responseListener, Metadata headers) {
TagContext tagContext = getTagContext(key, method.getFullMethodName(), databaseName);
Attributes attributes =
getMetricAttributes(key, method.getFullMethodName(), databaseName);
Map<String, String> builtInMetricsAttributes =
getBuiltInMetricAttributes(key, databaseName);
Map<String, String> commonBuiltInMetricAttributes =
getCommonBuiltInMetricAttributes(key, databaseName);
super.start(
new SimpleForwardingClientCallListener<RespT>(responseListener) {
@Override
public void onHeaders(Metadata metadata) {
Boolean isDirectPathUsed =
isDirectPathUsed(getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR));
addBuiltInMetricAttributes(
compositeTracer, builtInMetricsAttributes, isDirectPathUsed);
processHeader(metadata, tagContext, attributes, span);
compositeTracer, commonBuiltInMetricAttributes, isDirectPathUsed);
processHeader(
metadata,
tagContext,
attributes,
span,
getBuiltInMetricAttributes(commonBuiltInMetricAttributes, isDirectPathUsed));
super.onHeaders(metadata);
}
},
Expand All @@ -142,7 +153,11 @@ public void onHeaders(Metadata metadata) {
}

private void processHeader(
Metadata metadata, TagContext tagContext, Attributes attributes, Span span) {
Metadata metadata,
TagContext tagContext,
Attributes attributes,
Span span,
Map<String, String> builtInMetricsAttributes) {
MeasureMap measureMap = STATS_RECORDER.newMeasureMap();
String serverTiming = metadata.get(SERVER_TIMING_HEADER_KEY);
if (serverTiming != null && serverTiming.startsWith(SERVER_TIMING_HEADER_PREFIX)) {
Expand All @@ -154,6 +169,7 @@ private void processHeader(

spannerRpcMetrics.recordGfeLatency(latency, attributes);
spannerRpcMetrics.recordGfeHeaderMissingCount(0L, attributes);
builtInOpenTelemetryMetricsRecorder.recordGFELatency(latency, builtInMetricsAttributes);

if (span != null) {
span.setAttribute("gfe_latency", String.valueOf(latency));
Expand Down Expand Up @@ -224,8 +240,8 @@ private Attributes getMetricAttributes(String key, String method, DatabaseName d
});
}

private Map<String, String> getBuiltInMetricAttributes(String key, DatabaseName databaseName)
throws ExecutionException {
private Map<String, String> getCommonBuiltInMetricAttributes(
String key, DatabaseName databaseName) throws ExecutionException {
return builtInAttributesCache.get(
key,
() -> {
Expand All @@ -240,17 +256,21 @@ private Map<String, String> getBuiltInMetricAttributes(String key, DatabaseName
});
}

private Map<String, String> getBuiltInMetricAttributes(
Map<String, String> commonBuiltInMetricsAttributes, Boolean isDirectPathUsed) {
Map<String, String> builtInMetricAttributes = new HashMap<>(commonBuiltInMetricsAttributes);
builtInMetricAttributes.put(
BuiltInMetricsConstant.DIRECT_PATH_USED_KEY.getKey(), Boolean.toString(isDirectPathUsed));
return builtInMetricAttributes;
}

private void addBuiltInMetricAttributes(
CompositeTracer compositeTracer,
Map<String, String> builtInMetricsAttributes,
Map<String, String> commonBuiltInMetricsAttributes,
Boolean isDirectPathUsed) {
if (compositeTracer != null) {
// Direct Path used attribute
Map<String, String> attributes = new HashMap<>(builtInMetricsAttributes);
attributes.put(
BuiltInMetricsConstant.DIRECT_PATH_USED_KEY.getKey(), Boolean.toString(isDirectPathUsed));

compositeTracer.addAttributes(attributes);
compositeTracer.addAttributes(
getBuiltInMetricAttributes(commonBuiltInMetricsAttributes, isDirectPathUsed));
}
}

Expand Down
Loading

0 comments on commit 7383362

Please sign in to comment.