Skip to content

perf(spanner): create Cloud Monitoring MetricServiceClient lazily #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fornwall wants to merge 1 commit into
base: main
Choose a base branch
from
Open

perf(spanner): create Cloud Monitoring MetricServiceClient lazily #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 86 additions & 19 deletions ...-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@
import java.util.stream.Collectors;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;

/**
* Spanner Cloud Monitoring OpenTelemetry Exporter.
Expand All @@ -72,7 +73,27 @@ class SpannerCloudMonitoringExporter implements MetricExporter {
// https://cloud.google.com/monitoring/quotas#custom_metrics_quotas.
private static final int EXPORT_BATCH_SIZE_LIMIT = 200;
private final Set<String> spannerExportFailureLoggedProjects = ConcurrentHashMap.newKeySet();
private final MetricServiceClient client;

/**
* Creates the underlying {@link MetricServiceClient}. Invoked lazily so that the (relatively
* expensive) Cloud Monitoring client — with its own channel, executor and watchdog — is only
* built the first time metrics are actually exported, rather than eagerly while the Spanner
* client is still being constructed.
*/
@FunctionalInterface
interface MetricServiceClientFactory {
MetricServiceClient create() throws IOException;
}

private final MetricServiceClientFactory clientFactory;
private final Object clientLock = new Object();

@GuardedBy("clientLock")
private MetricServiceClient client;

@GuardedBy("clientLock")
private boolean shutdown;

private final Supplier<String> fallbackProjectIdSupplier;

static SpannerCloudMonitoringExporter create(
Expand Down Expand Up @@ -114,8 +135,12 @@ static SpannerCloudMonitoringExporter create(
// it as not retried for now.
settingsBuilder.createServiceTimeSeriesSettings().setSimpleTimeoutNoRetriesDuration(timeout);

// Building the settings is cheap, but creating the MetricServiceClient allocates a channel,
// executor and watchdog. Defer that to the first export() so it does not slow down Spanner
// client construction (the first metric export only happens after the first reader interval).
MetricServiceSettings settings = settingsBuilder.build();
return new SpannerCloudMonitoringExporter(
fallbackProjectIdSupplier, MetricServiceClient.create(settingsBuilder.build()));
fallbackProjectIdSupplier, () -> MetricServiceClient.create(settings));
}

@VisibleForTesting
Expand All @@ -126,32 +151,64 @@ static SpannerCloudMonitoringExporter create(
@VisibleForTesting
SpannerCloudMonitoringExporter(
Supplier<String> fallbackProjectIdSupplier, MetricServiceClient client) {
this.client = client;
this(fallbackProjectIdSupplier, () -> client);
}

@VisibleForTesting
SpannerCloudMonitoringExporter(
Supplier<String> fallbackProjectIdSupplier, MetricServiceClientFactory clientFactory) {
this.clientFactory = clientFactory;
this.fallbackProjectIdSupplier = fallbackProjectIdSupplier;
}

/**
* Returns the lazily-created {@link MetricServiceClient}, or {@code null} if the exporter has
* already been shut down.
*/
@Nullable
private MetricServiceClient getOrCreateClient() throws IOException {
synchronized (clientLock) {
if (shutdown) {
return null;
}
if (client == null) {
client = clientFactory.create();
}
return client;
}
}

@Override
public CompletableResultCode export(@Nonnull Collection<MetricData> collection) {
if (client.isShutdown()) {
// Skip exporting (and avoid creating the Cloud Monitoring client) when there's nothing to send.
if (collection.isEmpty()) {
return CompletableResultCode.ofSuccess();
}

MetricServiceClient client;
try {
client = getOrCreateClient();
} catch (Throwable e) {
logger.log(
Level.WARNING, "Failed to create the Cloud Monitoring client for exporting metrics", e);
return CompletableResultCode.ofFailure();
}
if (client == null || client.isShutdown()) {
logger.log(Level.WARNING, "Exporter is shut down");
return CompletableResultCode.ofFailure();
}

return exportSpannerClientMetrics(collection);
return exportSpannerClientMetrics(collection, client);
}

@VisibleForTesting
MetricServiceClient getMetricServiceClient() {
return client;
MetricServiceClient getMetricServiceClient() throws IOException {
return getOrCreateClient();
}

/** Export client built in metrics */
private CompletableResultCode exportSpannerClientMetrics(Collection<MetricData> collection) {
// Skips exporting if there's none
if (collection.isEmpty()) {
return CompletableResultCode.ofSuccess();
}

private CompletableResultCode exportSpannerClientMetrics(
Collection<MetricData> collection, MetricServiceClient client) {
List<TimeSeries> spannerTimeSeries;
try {
spannerTimeSeries =
Expand Down Expand Up @@ -182,7 +239,8 @@ private CompletableResultCode exportSpannerClientMetrics(Collection<MetricData>
List<ApiFuture<List<Empty>>> futures = new ArrayList<>();
for (Map.Entry<String, List<TimeSeries>> entry : timeSeriesByProject.entrySet()) {
ProjectName projectName = ProjectName.of(entry.getKey());
ApiFuture<List<Empty>> future = exportTimeSeriesInBatch(projectName, entry.getValue());
ApiFuture<List<Empty>> future =
exportTimeSeriesInBatch(projectName, entry.getValue(), client);
ApiFutures.addCallback(
future,
new ApiFutureCallback<List<Empty>>() {
Expand Down Expand Up @@ -246,7 +304,7 @@ private void logExportFailure(Throwable throwable, ProjectName projectName) {
}

private ApiFuture<List<Empty>> exportTimeSeriesInBatch(
ProjectName projectName, List<TimeSeries> timeSeries) {
ProjectName projectName, List<TimeSeries> timeSeries, MetricServiceClient client) {
List<ApiFuture<Empty>> batchResults = new ArrayList<>();

for (List<TimeSeries> batch : Iterables.partition(timeSeries, EXPORT_BATCH_SIZE_LIMIT)) {
Expand All @@ -255,7 +313,7 @@ private ApiFuture<List<Empty>> exportTimeSeriesInBatch(
.setName(projectName.toString())
.addAllTimeSeries(batch)
.build();
batchResults.add(this.client.createServiceTimeSeriesCallable().futureCall(req));
batchResults.add(client.createServiceTimeSeriesCallable().futureCall(req));
}

return ApiFutures.allAsList(batchResults);
Expand All @@ -268,13 +326,22 @@ public CompletableResultCode flush() {

@Override
public CompletableResultCode shutdown() {
if (client.isShutdown()) {
logger.log(Level.WARNING, "shutdown is called multiple times");
MetricServiceClient clientToShutdown;
synchronized (clientLock) {
if (shutdown) {
logger.log(Level.WARNING, "shutdown is called multiple times");
return CompletableResultCode.ofSuccess();
}
shutdown = true;
clientToShutdown = client;
}
// The client is created lazily on the first export, so it may never have been created.
if (clientToShutdown == null) {
return CompletableResultCode.ofSuccess();
}
CompletableResultCode shutdownResult = new CompletableResultCode();
try {
client.shutdown();
clientToShutdown.shutdown();
shutdownResult.succeed();
} catch (Throwable e) {
logger.log(Level.WARNING, "failed to shutdown the monitoring client", e);
Expand Down
72 changes: 72 additions & 0 deletions ...ud-spanner/src/test/java/com/google/cloud/spanner/SpannerCloudMonitoringExporterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import io.opentelemetry.api.trace.SpanContext;
import io.opentelemetry.api.trace.TraceFlags;
import io.opentelemetry.api.trace.TraceState;
import io.opentelemetry.sdk.common.CompletableResultCode;
import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
import io.opentelemetry.sdk.metrics.InstrumentType;
import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
Expand All @@ -70,6 +71,7 @@
import io.opentelemetry.sdk.resources.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.junit.After;
import org.junit.Before;
Expand Down Expand Up @@ -682,6 +684,76 @@ public void testUniverseDomain() throws IOException {
assertEquals("monitoringa.abc.goog:443", metricServiceSettings.getEndpoint());
}

@Test
public void testClientCreatedLazilyOnFirstNonEmptyExport() {
AtomicInteger creationCount = new AtomicInteger();
SpannerCloudMonitoringExporter lazyExporter =
new SpannerCloudMonitoringExporter(
() -> null,
() -> {
creationCount.incrementAndGet();
return fakeMetricServiceClient;
});

// The client must not be created at construction time.
assertThat(creationCount.get()).isEqualTo(0);

// An empty export must not trigger client creation either.
assertThat(lazyExporter.export(Collections.emptyList()).isSuccess()).isTrue();
assertThat(creationCount.get()).isEqualTo(0);

UnaryCallable<CreateTimeSeriesRequest, Empty> mockCallable =
mock(UnaryCallable.class, withSettings().withoutAnnotations());
when(mockMetricServiceStub.createServiceTimeSeriesCallable()).thenReturn(mockCallable);
when(mockCallable.futureCall(Mockito.any()))
.thenReturn(ApiFutures.immediateFuture(Empty.getDefaultInstance()));

// The first non-empty export creates the client exactly once...
lazyExporter.export(Collections.singletonList(buildLongData()));
assertThat(creationCount.get()).isEqualTo(1);

// ...and subsequent exports reuse it.
lazyExporter.export(Collections.singletonList(buildLongData()));
assertThat(creationCount.get()).isEqualTo(1);
}

@Test
public void testShutdownBeforeExportDoesNotCreateClient() {
AtomicInteger creationCount = new AtomicInteger();
SpannerCloudMonitoringExporter lazyExporter =
new SpannerCloudMonitoringExporter(
() -> null,
() -> {
creationCount.incrementAndGet();
return fakeMetricServiceClient;
});

// Shutting down before any export succeeds without ever creating the client.
assertThat(lazyExporter.shutdown().isSuccess()).isTrue();
assertThat(creationCount.get()).isEqualTo(0);

// A second shutdown is a no-op.
assertThat(lazyExporter.shutdown().isSuccess()).isTrue();

// Exporting after shutdown fails and still never creates a client.
CompletableResultCode exportResult =
lazyExporter.export(Collections.singletonList(buildLongData()));
assertThat(exportResult.isSuccess()).isFalse();
assertThat(creationCount.get()).isEqualTo(0);
}

private MetricData buildLongData() {
LongPointData longPointData = ImmutableLongPointData.create(10, 15, attributes, 11L);
return ImmutableMetricData.createLongSum(
resource,
InstrumentationScopeInfo.create(GRPC_METER_NAME),
"spanner.googleapis.com/internal/client/" + OPERATION_COUNT_NAME,
"description",
"1",
ImmutableSumData.create(
true, AggregationTemporality.CUMULATIVE, ImmutableList.of(longPointData)));
}

private static class FakeMetricServiceClient extends MetricServiceClient {

protected FakeMetricServiceClient(MetricServiceStub stub) {
Expand Down