diff --git a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporter.java b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporter.java index be1b04f1df07..e3def1845111 100644 --- a/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporter.java +++ b/java-spanner/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerCloudMonitoringExporter.java @@ -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. @@ -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 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 fallbackProjectIdSupplier; static SpannerCloudMonitoringExporter create( @@ -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 @@ -126,32 +151,64 @@ static SpannerCloudMonitoringExporter create( @VisibleForTesting SpannerCloudMonitoringExporter( Supplier fallbackProjectIdSupplier, MetricServiceClient client) { - this.client = client; + this(fallbackProjectIdSupplier, () -> client); + } + + @VisibleForTesting + SpannerCloudMonitoringExporter( + Supplier 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 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 collection) { - // Skips exporting if there's none - if (collection.isEmpty()) { - return CompletableResultCode.ofSuccess(); - } - + private CompletableResultCode exportSpannerClientMetrics( + Collection collection, MetricServiceClient client) { List spannerTimeSeries; try { spannerTimeSeries = @@ -182,7 +239,8 @@ private CompletableResultCode exportSpannerClientMetrics(Collection List>> futures = new ArrayList<>(); for (Map.Entry> entry : timeSeriesByProject.entrySet()) { ProjectName projectName = ProjectName.of(entry.getKey()); - ApiFuture> future = exportTimeSeriesInBatch(projectName, entry.getValue()); + ApiFuture> future = + exportTimeSeriesInBatch(projectName, entry.getValue(), client); ApiFutures.addCallback( future, new ApiFutureCallback>() { @@ -246,7 +304,7 @@ private void logExportFailure(Throwable throwable, ProjectName projectName) { } private ApiFuture> exportTimeSeriesInBatch( - ProjectName projectName, List timeSeries) { + ProjectName projectName, List timeSeries, MetricServiceClient client) { List> batchResults = new ArrayList<>(); for (List batch : Iterables.partition(timeSeries, EXPORT_BATCH_SIZE_LIMIT)) { @@ -255,7 +313,7 @@ private ApiFuture> exportTimeSeriesInBatch( .setName(projectName.toString()) .addAllTimeSeries(batch) .build(); - batchResults.add(this.client.createServiceTimeSeriesCallable().futureCall(req)); + batchResults.add(client.createServiceTimeSeriesCallable().futureCall(req)); } return ApiFutures.allAsList(batchResults); @@ -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); diff --git a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpannerCloudMonitoringExporterTest.java b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpannerCloudMonitoringExporterTest.java index 15bf91fa9205..ab9f4e318c28 100644 --- a/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpannerCloudMonitoringExporterTest.java +++ b/java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/SpannerCloudMonitoringExporterTest.java @@ -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; @@ -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; @@ -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 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) {