From de795895baa7b0b17920809009aaee17eeaf3ab9 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Tue, 24 Mar 2026 16:58:28 +0100 Subject: [PATCH 1/2] migrate otel auth extension. --- .../beam/gradle/BeamModulePlugin.groovy | 2 + .../build.gradle | 48 + .../gcp/auth/ConfigurableOption.java | 163 +++ ...thAutoConfigurationCustomizerProvider.java | 276 ++++ .../gcp/auth/GoogleAuthException.java | 69 + ...re.spi.AutoConfigurationCustomizerProvider | 1 + ...toConfigurationCustomizerProviderTest.java | 1242 +++++++++++++++++ settings.gradle.kts | 1 + 8 files changed, 1802 insertions(+) create mode 100644 sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle create mode 100644 sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/ConfigurableOption.java create mode 100644 sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java create mode 100644 sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GoogleAuthException.java create mode 100644 sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider create mode 100644 sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index 5ca0de9de846..8f7955d600da 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -869,6 +869,8 @@ class BeamModulePlugin implements Plugin { opentelemetry_sdk : "io.opentelemetry:opentelemetry-sdk", // opentelemetry-bom sets version opentelemetry_exporter_otlp : "io.opentelemetry:opentelemetry-exporter-otlp", // opentelemetry-bom sets version opentelemetry_extension_autoconfigure : "io.opentelemetry:opentelemetry-sdk-extension-autoconfigure", // opentelemetry-bom sets version + opentelemetry_proto : "io.opentelemetry.proto:opentelemetry-proto:$opentelemetry_version-alpha", + opentelemetry_sdk_testing : "io.opentelemetry:opentelemetry-sdk-testing:$opentelemetry_version", postgres : "org.postgresql:postgresql:$postgres_version", protobuf_java : "com.google.protobuf:protobuf-java:$protobuf_version", protobuf_java_util : "com.google.protobuf:protobuf-java-util:$protobuf_version", diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle b/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle new file mode 100644 index 000000000000..7a750597a87e --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle @@ -0,0 +1,48 @@ +/* + * 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. + */ + +plugins { id 'org.apache.beam.module' } +applyJavaNature( + automaticModuleName: 'org.apache.beam.sdk.extensions.opentelemetry.gcp.auth', +) + +description = "Apache Beam :: SDKs :: Java :: Extensions :: OpenTelemetry GCP Auth" +ext.summary = "OpenTelemetry extension that provides GCP authentication support for OTLP exporters." + +dependencies { + implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom) + implementation platform(library.java.opentelemetry_bom) + implementation library.java.google_auth_library_oauth2_http + compileOnly library.java.opentelemetry_api + compileOnly library.java.opentelemetry_extension_autoconfigure + compileOnly library.java.opentelemetry_exporter_otlp + + testImplementation library.java.junit + testImplementation library.java.mockito_core + testImplementation library.java.mockito_inline + testImplementation library.java.jupiter_api + testImplementation library.java.jupiter_params + testRuntimeOnly library.java.jupiter_engine + testImplementation library.java.truth + testImplementation library.java.opentelemetry_api + testImplementation library.java.opentelemetry_sdk + testImplementation library.java.opentelemetry_exporter_otlp + testImplementation library.java.opentelemetry_sdk_testing + testImplementation library.java.opentelemetry_extension_autoconfigure +} +test { useJUnitPlatform() } diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/ConfigurableOption.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/ConfigurableOption.java new file mode 100644 index 000000000000..e18e5693e3a1 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/ConfigurableOption.java @@ -0,0 +1,163 @@ +/* + * 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.beam.sdk.extensions.opentelemetry.gcp.auth; + +import static java.util.Locale.ROOT; + +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException; +import java.util.Optional; +import java.util.function.Supplier; + +/** + * An enum representing configurable options for a GCP Authentication Extension. Each option has a + * user-readable name and can be configured using environment variables or system properties. + * + *

Copied from + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/main/java/io/opentelemetry/contrib/gcp/auth/ConfigurableOption.java + */ +enum ConfigurableOption { + /** + * Represents the Google Cloud Project ID option. Can be configured using the environment variable + * `GOOGLE_CLOUD_PROJECT` or the system property `google.cloud.project`. + */ + GOOGLE_CLOUD_PROJECT("Google Cloud Project ID"), + + /** + * Represents the Google Cloud Quota Project ID option. Can be configured using the environment + * variable `GOOGLE_CLOUD_QUOTA_PROJECT` or the system property `google.cloud.quota.project`. The + * quota project is the project that is used for quota management and billing for the API usage. + * + *

The environment variable name is selected to be consistent with the official GCP client + * libraries. + */ + GOOGLE_CLOUD_QUOTA_PROJECT("Google Cloud Quota Project ID"), + + /** + * Specifies a comma-separated list of OpenTelemetry signals for which this authentication + * extension should be active. The authentication mechanisms provided by this extension will only + * be applied to the listed signals. If not set, {@code all} is assumed to be set which means + * authentication is enabled for all supported signals. + * + *

Valid signal values are: + * + *

+ * + *

The values are case-sensitive. Whitespace around commas and values is ignored. Can be + * configured using the environment variable `GOOGLE_OTEL_AUTH_TARGET_SIGNALS` or the system + * property `google.otel.auth.target.signals`. + */ + GOOGLE_OTEL_AUTH_TARGET_SIGNALS("Target Signals for Google Authentication Extension"); + + private final String userReadableName; + private final String environmentVariableName; + private final String systemPropertyName; + + ConfigurableOption(String userReadableName) { + this.userReadableName = userReadableName; + this.environmentVariableName = this.name(); + this.systemPropertyName = this.environmentVariableName.toLowerCase(ROOT).replace('_', '.'); + } + + /** + * Returns the environment variable name associated with this option. + * + * @return the environment variable name (e.g., GOOGLE_CLOUD_PROJECT) + */ + String getEnvironmentVariable() { + return this.environmentVariableName; + } + + /** + * Returns the system property name associated with this option. + * + * @return the system property name (e.g., google.cloud.project) + */ + String getSystemProperty() { + return this.systemPropertyName; + } + + /** + * Returns the user readable name associated with this option. + * + * @return the user readable name (e.g., "Google Cloud Quota Project ID") + */ + String getUserReadableName() { + return this.userReadableName; + } + + /** + * Retrieves the configured value for this option. This method checks the environment variable + * first and then the system property. + * + * @return The configured value as a string, or throws an exception if not configured. + * @throws ConfigurationException if neither the environment variable nor the system property is + * set. + */ + String getConfiguredValue(ConfigProperties configProperties) { + String configuredValue = configProperties.getString(this.getSystemProperty()); + if (configuredValue != null && !configuredValue.isEmpty()) { + return configuredValue; + } else { + throw new ConfigurationException( + String.format( + "GCP Authentication Extension not configured properly: %s not configured. Configure it by exporting environment variable %s or system property %s", + this.userReadableName, this.getEnvironmentVariable(), this.getSystemProperty())); + } + } + + /** + * Retrieves the value for this option, prioritizing environment variables and system properties. + * If neither an environment variable nor a system property is set for this option, the provided + * fallback function is used to determine the value. + * + * @param fallback A {@link Supplier} that provides the default value for the option when it is + * not explicitly configured via an environment variable or system property. + * @return The configured value for the option, obtained from the environment variable, system + * property, or the fallback function, in that order of precedence. + */ + String getConfiguredValueWithFallback( + ConfigProperties configProperties, Supplier fallback) { + try { + return this.getConfiguredValue(configProperties); + } catch (ConfigurationException e) { + return fallback.get(); + } + } + + /** + * Retrieves the value for this option, prioritizing environment variables before system + * properties. If neither an environment variable nor a system property is set for this option, + * then an empty {@link Optional} is returned. + * + * @return The configured value for the option, if set, obtained from the environment variable, + * system property, or empty {@link Optional}, in that order of precedence. + */ + Optional getConfiguredValueAsOptional(ConfigProperties configProperties) { + try { + return Optional.of(this.getConfiguredValue(configProperties)); + } catch (ConfigurationException e) { + return Optional.empty(); + } + } +} diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java new file mode 100644 index 000000000000..42af9d4fcb58 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java @@ -0,0 +1,276 @@ +/* + * 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.beam.sdk.extensions.opentelemetry.gcp.auth; + +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static java.util.Arrays.stream; +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toMap; + +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auto.service.AutoService; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter; +import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException; +import io.opentelemetry.sdk.metrics.export.MetricExporter; +import io.opentelemetry.sdk.resources.Resource; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.logging.Level; +import java.util.logging.Logger; +import javax.annotation.Nonnull; +import org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GoogleAuthException.Reason; + +/** + * An AutoConfigurationCustomizerProvider for Google Cloud Platform (GCP) OpenTelemetry (OTLP) + * integration. + * + *

This class is registered as a service provider using {@link AutoService} and is responsible + * for customizing the OpenTelemetry configuration for GCP specific behavior. It retrieves Google + * Application Default Credentials (ADC) and adds them as authorization headers to the configured + * {@link SpanExporter}. It also sets default properties and resource attributes for GCP + * integration. + * + *

Copied from + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/main/java/io/opentelemetry/contrib/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java + * + * @see AutoConfigurationCustomizerProvider + * @see GoogleCredentials + */ +@AutoService(AutoConfigurationCustomizerProvider.class) +public class GcpAuthAutoConfigurationCustomizerProvider + implements AutoConfigurationCustomizerProvider { + + private static final Logger logger = + Logger.getLogger(GcpAuthAutoConfigurationCustomizerProvider.class.getName()); + private static final String SIGNAL_TARGET_WARNING_FIX_SUGGESTION = + String.format( + "You may safely ignore this warning if it is intentional, otherwise please configure the '%s' by exporting valid values to environment variable: %s or by setting valid values in system property: %s.", + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getUserReadableName(), + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getEnvironmentVariable(), + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty()); + + static final String QUOTA_USER_PROJECT_HEADER = "x-goog-user-project"; + static final String GCP_USER_PROJECT_ID_KEY = "gcp.project_id"; + + static final String SIGNAL_TYPE_TRACES = "traces"; + static final String SIGNAL_TYPE_METRICS = "metrics"; + static final String SIGNAL_TYPE_ALL = "all"; + + /** + * Customizes the provided {@link AutoConfigurationCustomizer} such that authenticated exports to + * GCP Telemetry API are possible from the configured OTLP exporter. + * + *

This method attempts to retrieve Google Application Default Credentials (ADC) and performs + * the following: + * + *

+ * + * The 'customization' performed includes customizing the exporters by adding required headers to + * the export calls made and customizing the resource by adding required resource attributes to + * enable GCP integration. + * + * @param autoConfiguration the AutoConfigurationCustomizer to customize. + * @throws GoogleAuthException if there's an error retrieving Google Application Default + * Credentials. + * @throws io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException if required options are + * not configured through environment variables or system properties. + */ + @Override + public void customize(@Nonnull AutoConfigurationCustomizer autoConfiguration) { + GoogleCredentials credentials; + try { + credentials = GoogleCredentials.getApplicationDefault(); + } catch (IOException e) { + throw new GoogleAuthException(Reason.FAILED_ADC_RETRIEVAL, e); + } + autoConfiguration + .addSpanExporterCustomizer( + (spanExporter, configProperties) -> + customizeSpanExporter(spanExporter, credentials, configProperties)) + .addMetricExporterCustomizer( + (metricExporter, configProperties) -> + customizeMetricExporter(metricExporter, credentials, configProperties)) + .addResourceCustomizer( + (resource, configProperties) -> + customizeResource(resource, credentials, configProperties)); + } + + @Override + public int order() { + return Integer.MAX_VALUE - 1; + } + + private static SpanExporter customizeSpanExporter( + SpanExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + if (isSignalTargeted(SIGNAL_TYPE_TRACES, configProperties)) { + return addAuthorizationHeaders(exporter, credentials, configProperties); + } else { + String[] params = {SIGNAL_TYPE_TRACES, SIGNAL_TARGET_WARNING_FIX_SUGGESTION}; + logger.log( + Level.WARNING, + "GCP Authentication Extension is not configured for signal type: {0}. {1}", + params); + } + return exporter; + } + + private static MetricExporter customizeMetricExporter( + MetricExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + if (isSignalTargeted(SIGNAL_TYPE_METRICS, configProperties)) { + return addAuthorizationHeaders(exporter, credentials, configProperties); + } else { + String[] params = {SIGNAL_TYPE_METRICS, SIGNAL_TARGET_WARNING_FIX_SUGGESTION}; + logger.log( + Level.WARNING, + "GCP Authentication Extension is not configured for signal type: {0}. {1}", + params); + } + return exporter; + } + + // Checks if the auth extension is configured to target the passed signal for authentication. + private static boolean isSignalTargeted(String checkSignal, ConfigProperties configProperties) { + String endpoint = configProperties.getString("otel.exporter.otlp." + checkSignal + ".endpoint"); + if (endpoint == null) { + endpoint = configProperties.getString("otel.exporter.otlp.endpoint"); + } + if (endpoint == null + || (!endpoint.startsWith("https://telemetry.googleapis.com") + && !endpoint.startsWith("https://telemetry.mtls.googleapis.com"))) { + return false; + } + String userSpecifiedTargetedSignals = + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getConfiguredValueWithFallback( + configProperties, () -> SIGNAL_TYPE_ALL); + return stream(userSpecifiedTargetedSignals.split(",")) + .map(String::trim) + .anyMatch( + targetedSignal -> + targetedSignal.equals(checkSignal) || targetedSignal.equals(SIGNAL_TYPE_ALL)); + } + + // Adds authorization headers to the calls made by the OtlpGrpcSpanExporter and + // OtlpHttpSpanExporter. + private static SpanExporter addAuthorizationHeaders( + SpanExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + if (exporter instanceof OtlpHttpSpanExporter) { + OtlpHttpSpanExporterBuilder builder = + ((OtlpHttpSpanExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); + return builder.build(); + } else if (exporter instanceof OtlpGrpcSpanExporter) { + OtlpGrpcSpanExporterBuilder builder = + ((OtlpGrpcSpanExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); + return builder.build(); + } + return exporter; + } + + // Adds authorization headers to the calls made by the OtlpGrpcMetricExporter and + // OtlpHttpMetricExporter. + private static MetricExporter addAuthorizationHeaders( + MetricExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + if (exporter instanceof OtlpHttpMetricExporter) { + OtlpHttpMetricExporterBuilder builder = + ((OtlpHttpMetricExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); + return builder.build(); + } else if (exporter instanceof OtlpGrpcMetricExporter) { + OtlpGrpcMetricExporterBuilder builder = + ((OtlpGrpcMetricExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); + return builder.build(); + } + return exporter; + } + + private static Map getRequiredHeaderMap( + GoogleCredentials credentials, ConfigProperties configProperties) { + Map> gcpHeaders; + try { + // this also refreshes the credentials, if required + gcpHeaders = credentials.getRequestMetadata(); + } catch (IOException e) { + throw new GoogleAuthException(Reason.FAILED_ADC_REFRESH, e); + } + Map flattenedHeaders = + gcpHeaders.entrySet().stream() + .collect( + toMap( + Map.Entry::getKey, + entry -> + entry.getValue().stream() + .filter(Objects::nonNull) // Filter nulls + .filter(s -> !s.isEmpty()) // Filter empty strings + .collect(joining(",")))); + // Add quota user project header if not detected by the auth library and user provided it via + // system properties. + if (!flattenedHeaders.containsKey(QUOTA_USER_PROJECT_HEADER)) { + Optional maybeConfiguredQuotaProjectId = + ConfigurableOption.GOOGLE_CLOUD_QUOTA_PROJECT.getConfiguredValueAsOptional( + configProperties); + maybeConfiguredQuotaProjectId.ifPresent( + configuredQuotaProjectId -> + flattenedHeaders.put(QUOTA_USER_PROJECT_HEADER, configuredQuotaProjectId)); + } + return flattenedHeaders; + } + + // Updates the current resource with the attributes required for ingesting OTLP data on GCP. + private static Resource customizeResource( + Resource resource, GoogleCredentials credentials, ConfigProperties configProperties) { + String gcpProjectId; + try { + gcpProjectId = ConfigurableOption.GOOGLE_CLOUD_PROJECT.getConfiguredValue(configProperties); + } catch (ConfigurationException e) { + gcpProjectId = credentials.getProjectId(); + if (gcpProjectId == null || gcpProjectId.isEmpty()) { + throw e; + } + } + Resource res = Resource.create(Attributes.of(stringKey(GCP_USER_PROJECT_ID_KEY), gcpProjectId)); + return resource.merge(res); + } +} diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GoogleAuthException.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GoogleAuthException.java new file mode 100644 index 000000000000..978b48934067 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GoogleAuthException.java @@ -0,0 +1,69 @@ +/* + * 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.beam.sdk.extensions.opentelemetry.gcp.auth; + +/** + * An unchecked exception indicating a failure during Google authentication. This exception is + * thrown when there are issues with retrieving or refreshing Google Application Default Credentials + * (ADC). + * + *

Copied from + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/main/java/io/opentelemetry/contrib/gcp/auth/GoogleAuthException.java + */ +public class GoogleAuthException extends RuntimeException { + + private static final long serialVersionUID = 149908685226796448L; + + /** + * Constructs a new {@code GoogleAuthException} with the specified reason and cause. + * + * @param reason the reason for the authentication failure. + * @param cause the underlying cause of the exception (e.g., an IOException). + */ + GoogleAuthException(Reason reason, Throwable cause) { + super(reason.message, cause); + } + + /** Enumerates the possible reasons for a Google authentication failure. */ + enum Reason { + /** Indicates a failure to retrieve Google Application Default Credentials. */ + FAILED_ADC_RETRIEVAL("Unable to retrieve Google Application Default Credentials."), + /** Indicates a failure to retrieve Google Application Default Credentials. */ + FAILED_ADC_REFRESH("Unable to refresh Google Application Default Credentials."); + + private final String message; + + /** + * Constructs a new {@code Reason} with the specified message. + * + * @param message the message describing the reason. + */ + Reason(String message) { + this.message = message; + } + + /** + * Returns the message associated with this reason. + * + * @return the message describing the reason. + */ + public String getMessage() { + return message; + } + } +} diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider new file mode 100644 index 000000000000..105921f0b4d3 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider @@ -0,0 +1 @@ +org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java new file mode 100644 index 000000000000..c825a48cc4b3 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java @@ -0,0 +1,1242 @@ +/* + * 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.beam.sdk.extensions.opentelemetry.gcp.auth; + +import static com.google.common.truth.Truth.assertThat; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.GCP_USER_PROJECT_ID_KEY; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.QUOTA_USER_PROJECT_HEADER; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.SIGNAL_TYPE_ALL; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.SIGNAL_TYPE_METRICS; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.SIGNAL_TYPE_TRACES; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auto.value.AutoValue; +import com.google.common.collect.ImmutableMap; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.common.ComponentLoader; +import io.opentelemetry.context.Scope; +import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter; +import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder; +import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException; +import io.opentelemetry.sdk.autoconfigure.spi.metrics.ConfigurableMetricExporterProvider; +import io.opentelemetry.sdk.autoconfigure.spi.traces.ConfigurableSpanExporterProvider; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.common.export.MemoryMode; +import io.opentelemetry.sdk.metrics.Aggregation; +import io.opentelemetry.sdk.metrics.InstrumentType; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.export.MetricExporter; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.io.IOException; +import java.time.Duration; +import java.time.Instant; +import java.util.AbstractMap.SimpleEntry; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import java.util.stream.Stream; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.MockitoAnnotations; +import org.mockito.stubbing.Answer; + +/** + * Copied from + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/test/java/io/opentelemetry/contrib/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java + */ +class GcpAuthAutoConfigurationCustomizerProviderTest { + + private static final String DUMMY_GCP_RESOURCE_PROJECT_ID = "my-gcp-resource-project-id"; + private static final String DUMMY_GCP_QUOTA_PROJECT_ID = "my-gcp-quota-project-id"; + private static final Random TEST_RANDOM = new Random(); + + @Mock private GoogleCredentials mockedGoogleCredentials; + + @Captor private ArgumentCaptor>> traceHeaderSupplierCaptor; + @Captor private ArgumentCaptor>> metricHeaderSupplierCaptor; + + private static final ImmutableMap defaultOtelPropertiesSpanExporter = + ImmutableMap.of( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "otlp", + "otel.metrics.exporter", + "none", + "otel.logs.exporter", + "none", + "otel.resource.attributes", + "foo=bar"); + + private static final ImmutableMap defaultOtelPropertiesMetricExporter = + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.traces.exporter", + "none", + "otel.metrics.exporter", + "otlp", + "otel.logs.exporter", + "none", + "otel.resource.attributes", + "foo=bar"); + + @BeforeEach + public void setup() { + MockitoAnnotations.openMocks(this); + } + + @AfterEach + public void teardown() { + System.clearProperty(ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty()); + System.clearProperty(ConfigurableOption.GOOGLE_CLOUD_QUOTA_PROJECT.getSystemProperty()); + System.clearProperty(ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty()); + } + + // TODO: Use parameterized test for testing traces customizer for http & grpc. + @Test + void testTraceCustomizerOtlpHttp() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_TRACES); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpHttpSpanExporter mockOtlpHttpSpanExporter = mock(OtlpHttpSpanExporter.class); + OtlpHttpSpanExporterBuilder otlpSpanExporterBuilder = OtlpHttpSpanExporter.builder(); + OtlpHttpSpanExporterBuilder spyOtlpHttpSpanExporterBuilder = + Mockito.spy(otlpSpanExporterBuilder); + when(spyOtlpHttpSpanExporterBuilder.build()).thenReturn(mockOtlpHttpSpanExporter); + + when(mockOtlpHttpSpanExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + List exportedSpans = new ArrayList<>(); + when(mockOtlpHttpSpanExporter.export(any())) + .thenAnswer( + invocationOnMock -> { + exportedSpans.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + Mockito.when(mockOtlpHttpSpanExporter.toBuilder()).thenReturn(spyOtlpHttpSpanExporterBuilder); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpHttpSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpHttpSpanExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpHttpSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + assertThat(traceHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(traceHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpHttpSpanExporter, Mockito.atLeast(1)).export(Mockito.anyCollection()); + + assertThat(exportedSpans).isNotEmpty(); + for (SpanData spanData : exportedSpans) { + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(spanData.getAttributes().asMap()).containsKey(AttributeKey.longKey("work_loop")); + } + } + } + + @Test + void testTraceCustomizerOtlpGrpc() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_TRACES); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpGrpcSpanExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpGrpcSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + assertThat(traceHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(traceHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpGrpcSpanExporter, Mockito.atLeast(1)).export(Mockito.anyCollection()); + + assertThat(exportedSpans).isNotEmpty(); + for (SpanData spanData : exportedSpans) { + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(spanData.getAttributes().asMap()).containsKey(AttributeKey.longKey("work_loop")); + } + } + } + + @Test + void testMetricCustomizerOtlpHttp() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), + SIGNAL_TYPE_METRICS); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpHttpMetricExporter mockOtlpHttpMetricExporter = Mockito.mock(OtlpHttpMetricExporter.class); + OtlpHttpMetricExporterBuilder otlpMetricExporterBuilder = OtlpHttpMetricExporter.builder(); + OtlpHttpMetricExporterBuilder spyOtlpHttpMetricExporterBuilder = + Mockito.spy(otlpMetricExporterBuilder); + List exportedMetrics = new ArrayList<>(); + configureHttpMockMetricExporter( + mockOtlpHttpMetricExporter, spyOtlpHttpMetricExporterBuilder, exportedMetrics); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpHttpMetricExporter); + generateTestMetric(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpHttpMetricExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpHttpMetricExporterBuilder, Mockito.times(1)) + .setHeaders(metricHeaderSupplierCaptor.capture()); + assertThat(metricHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(metricHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpHttpMetricExporter, Mockito.atLeast(1)) + .export(Mockito.anyCollection()); + + assertThat(exportedMetrics).isNotEmpty(); + for (MetricData metricData : exportedMetrics) { + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(metricData.getLongSumData().getPoints()).isNotEmpty(); + for (io.opentelemetry.sdk.metrics.data.PointData pointData : + metricData.getLongSumData().getPoints()) { + assertThat(pointData.getAttributes().asMap()) + .containsKey(AttributeKey.longKey("work_loop")); + } + } + } + } + + @Test + void testMetricCustomizerOtlpGrpc() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), + SIGNAL_TYPE_METRICS); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpGrpcMetricExporter mockOtlpGrpcMetricExporter = Mockito.mock(OtlpGrpcMetricExporter.class); + OtlpGrpcMetricExporterBuilder otlpMetricExporterBuilder = OtlpGrpcMetricExporter.builder(); + OtlpGrpcMetricExporterBuilder spyOtlpGrpcMetricExporterBuilder = + Mockito.spy(otlpMetricExporterBuilder); + List exportedMetrics = new ArrayList<>(); + configureGrpcMockMetricExporter( + mockOtlpGrpcMetricExporter, spyOtlpGrpcMetricExporterBuilder, exportedMetrics); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcMetricExporter); + generateTestMetric(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpGrpcMetricExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpGrpcMetricExporterBuilder, Mockito.times(1)) + .setHeaders(metricHeaderSupplierCaptor.capture()); + assertThat(metricHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(metricHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpGrpcMetricExporter, Mockito.atLeast(1)) + .export(Mockito.anyCollection()); + + assertThat(exportedMetrics).isNotEmpty(); + for (MetricData metricData : exportedMetrics) { + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(metricData.getLongSumData().getPoints()).isNotEmpty(); + for (io.opentelemetry.sdk.metrics.data.PointData pointData : + metricData.getLongSumData().getPoints()) { + assertThat(pointData.getAttributes().asMap()) + .containsKey(AttributeKey.longKey("work_loop")); + } + } + } + } + + @Test + void testCustomizerFailWithMissingResourceProject() { + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_ALL); + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + assertThrows( + ConfigurationException.class, + () -> buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter)); + } + } + + @ParameterizedTest + @MethodSource("provideQuotaBehaviorTestCases") + @SuppressWarnings("CannotMockMethod") + void testQuotaProjectBehavior(QuotaProjectIdTestBehavior testCase) throws IOException { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_ALL); + + // Prepare request metadata + AccessToken fakeAccessToken = new AccessToken("fake", Date.from(Instant.now())); + ImmutableMap> mockedRequestMetadata; + if (testCase.getIsQuotaProjectPresentInMetadata()) { + mockedRequestMetadata = + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue()), + QUOTA_USER_PROJECT_HEADER, + Collections.singletonList(DUMMY_GCP_QUOTA_PROJECT_ID)); + } else { + mockedRequestMetadata = + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue())); + } + // mock credentials to return the prepared request metadata + Mockito.when(mockedGoogleCredentials.getRequestMetadata()).thenReturn(mockedRequestMetadata); + + // configure environment according to test case + String quotaProjectId = testCase.getUserSpecifiedQuotaProjectId(); // maybe empty string + if (quotaProjectId != null) { + // user specified a quota project id + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_QUOTA_PROJECT.getSystemProperty(), quotaProjectId); + } + + // prepare mock exporter + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + // Export telemetry to capture headers in the export calls + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + Mockito.verify(spyOtlpGrpcSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + + // assert that the Authorization bearer token header is present + Map exportHeaders = traceHeaderSupplierCaptor.getValue().get(); + assertThat(exportHeaders).containsEntry("Authorization", "Bearer fake"); + + if (testCase.getExpectedQuotaProjectInHeader() == null) { + // there should be no user quota project header + assertThat(exportHeaders).doesNotContainKey(QUOTA_USER_PROJECT_HEADER); + } else { + // there should be user quota project header with expected value + assertThat(exportHeaders) + .containsEntry(QUOTA_USER_PROJECT_HEADER, testCase.getExpectedQuotaProjectInHeader()); + } + } + } + + @ParameterizedTest + @MethodSource("provideProjectIdBehaviorTestCases") + @SuppressWarnings("CannotMockMethod") + void testProjectIdBehavior(ProjectIdTestBehavior testCase) throws IOException { + System.clearProperty(ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty()); + System.clearProperty(ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty()); + + // configure environment according to test case + String userSpecifiedProjectId = testCase.getUserSpecifiedProjectId(); + if (userSpecifiedProjectId != null) { + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), userSpecifiedProjectId); + } + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_TRACES); + + // prepare request metadata (may or may not be called depending on test scenario) + AccessToken fakeAccessToken = new AccessToken("fake", Date.from(Instant.now())); + ImmutableMap> mockedRequestMetadata = + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue())); + Mockito.lenient() + .when(mockedGoogleCredentials.getRequestMetadata()) + .thenReturn(mockedRequestMetadata); + + // only mock getProjectId() if it will be called (i.e., user didn't specify project ID) + boolean shouldFallbackToCredentials = + userSpecifiedProjectId == null || userSpecifiedProjectId.isEmpty(); + if (shouldFallbackToCredentials) { + Mockito.when(mockedGoogleCredentials.getProjectId()) + .thenReturn(testCase.getCredentialsProjectId()); + } + + // prepare mock exporter + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + if (testCase.getExpectedToThrow()) { + // expect exception to be thrown when project ID is not available + assertThrows( + ConfigurationException.class, + () -> buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter)); + // verify getProjectId() was called to attempt fallback + Mockito.verify(mockedGoogleCredentials, Mockito.times(1)).getProjectId(); + } else { + // export telemetry and verify resource attributes contain expected project ID + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + assertThat(exportedSpans).isNotEmpty(); + for (SpanData spanData : exportedSpans) { + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), + testCase.getExpectedProjectIdInResource()); + } + + // verify whether getProjectId() was called based on whether fallback was needed + if (shouldFallbackToCredentials) { + Mockito.verify(mockedGoogleCredentials, Mockito.times(1)).getProjectId(); + } else { + Mockito.verify(mockedGoogleCredentials, Mockito.never()).getProjectId(); + } + } + } + } + + @ParameterizedTest + @MethodSource("provideTargetSignalBehaviorTestCases") + void testTargetSignalsBehavior(TargetSignalBehavior testCase) { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + // Prepare mocks + // Prepare mocked credential + prepareMockBehaviorForGoogleCredentials(); + + // Prepare mocked span exporter + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + // Prepare mocked metrics exporter + OtlpGrpcMetricExporter mockOtlpGrpcMetricExporter = Mockito.mock(OtlpGrpcMetricExporter.class); + OtlpGrpcMetricExporterBuilder otlpMetricExporterBuilder = OtlpGrpcMetricExporter.builder(); + OtlpGrpcMetricExporterBuilder spyOtlpGrpcMetricExporterBuilder = + Mockito.spy(otlpMetricExporterBuilder); + List exportedMetrics = new ArrayList<>(); + configureGrpcMockMetricExporter( + mockOtlpGrpcMetricExporter, spyOtlpGrpcMetricExporterBuilder, exportedMetrics); + + // configure environment according to test case + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), + testCase.getConfiguredTargetSignals()); + + // Build Autoconfigured OpenTelemetry SDK using the mocks and send signals + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = + buildOpenTelemetrySdkWithExporter( + mockOtlpGrpcSpanExporter, + mockOtlpGrpcMetricExporter, + testCase.getUserSpecifiedOtelProperties()); + generateTestMetric(sdk); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + // Check Traces modification conditions + if (testCase.getExpectedIsTraceSignalModified()) { + // If traces signal is expected to be modified, auth headers must be present + Mockito.verify(spyOtlpGrpcSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + assertThat(traceHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(traceHeaderSupplierCaptor.getValue().get())) + .isTrue(); + } else { + // If traces signals is not expected to be modified then no interaction with the builder + // should be made + Mockito.verifyNoInteractions(spyOtlpGrpcSpanExporterBuilder); + } + + // Check Metric modification conditions + if (testCase.getExpectedIsMetricsSignalModified()) { + // If metrics signal is expected to be modified, auth headers must be present + Mockito.verify(spyOtlpGrpcMetricExporterBuilder, Mockito.times(1)) + .setHeaders(metricHeaderSupplierCaptor.capture()); + assertThat(metricHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(metricHeaderSupplierCaptor.getValue().get())) + .isTrue(); + } else { + // If metrics signals is not expected to be modified then no interaction with the builder + // should be made + Mockito.verifyNoInteractions(spyOtlpGrpcMetricExporterBuilder); + } + } + } + + /** Test cases specifying expected behavior for GOOGLE_OTEL_AUTH_TARGET_SIGNALS */ + private static Stream provideTargetSignalBehaviorTestCases() { + return Stream.of( + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("traces") + .setUserSpecifiedOtelProperties(defaultOtelPropertiesSpanExporter) + .setExpectedIsMetricsSignalModified(false) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metrics") + .setUserSpecifiedOtelProperties(defaultOtelPropertiesMetricExporter) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(false) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("all") + .setUserSpecifiedOtelProperties( + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "otlp", + "otel.metrics.exporter", + "otlp", + "otel.logs.exporter", + "none")) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metrics, traces") + .setUserSpecifiedOtelProperties( + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "otlp", + "otel.metrics.exporter", + "otlp", + "otel.logs.exporter", + "none")) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("") + .setUserSpecifiedOtelProperties( + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "otlp", + "otel.metrics.exporter", + "otlp", + "otel.logs.exporter", + "none")) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("all") + .setUserSpecifiedOtelProperties( + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "none", + "otel.metrics.exporter", + "none", + "otel.logs.exporter", + "none")) + .setExpectedIsMetricsSignalModified(false) + .setExpectedIsTraceSignalModified(false) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metric, trace") + .setUserSpecifiedOtelProperties( + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "otlp", + "otel.metrics.exporter", + "otlp", + "otel.logs.exporter", + "none")) + .setExpectedIsMetricsSignalModified(false) + .setExpectedIsTraceSignalModified(false) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metrics, trace") + .setUserSpecifiedOtelProperties( + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "otlp", + "otel.metrics.exporter", + "otlp", + "otel.logs.exporter", + "none")) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(false) + .build())); + } + + /** + * Test cases specifying expected value for the project ID in the resource given the user input + * and the current credentials state. + * + *

{@code null} for {@link ProjectIdTestBehavior#getUserSpecifiedProjectId()} indicates the + * case of user not specifying the project ID. + * + *

{@code null} value for {@link ProjectIdTestBehavior#getCredentialsProjectId()} indicates + * that the mocked credentials are not providing a project ID. + * + *

{@code true} for {@link ProjectIdTestBehavior#getExpectedToThrow()} indicates the + * expectation that an exception should be thrown. + */ + private static Stream provideProjectIdBehaviorTestCases() { + return Stream.of( + // User specified project ID takes precedence + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId(DUMMY_GCP_RESOURCE_PROJECT_ID) + .setCredentialsProjectId("credentials-project-id") + .setExpectedProjectIdInResource(DUMMY_GCP_RESOURCE_PROJECT_ID) + .setExpectedToThrow(false) + .build()), + // If user specified project ID is empty, fallback to credentials.getProjectId() + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId("") + .setCredentialsProjectId("credentials-project-id") + .setExpectedProjectIdInResource("credentials-project-id") + .setExpectedToThrow(false) + .build()), + // If user doesn't specify project ID, fallback to credentials.getProjectId() + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId(null) + .setCredentialsProjectId("credentials-project-id") + .setExpectedProjectIdInResource("credentials-project-id") + .setExpectedToThrow(false) + .build()), + // If user doesn't specify and credentials.getProjectId() returns null, throw exception + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId(null) + .setCredentialsProjectId(null) + .setExpectedProjectIdInResource(null) + .setExpectedToThrow(true) + .build()), + // If user specified project ID is empty and credentials.getProjectId() returns null, throw + // exception + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId("") + .setCredentialsProjectId(null) + .setExpectedProjectIdInResource(null) + .setExpectedToThrow(true) + .build()), + // If user specifies empty and credentials returns empty (edge case), throw exception + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId("") + .setCredentialsProjectId("") + .setExpectedProjectIdInResource(null) + .setExpectedToThrow(true) + .build())); + } + + /** + * Test cases specifying expected value for the user quota project header given the user input and + * the current credentials state. + * + *

{@code null} for {@link QuotaProjectIdTestBehavior#getUserSpecifiedQuotaProjectId()} + * indicates the case of user not specifying the quota project ID. + * + *

{@code null} value for {@link QuotaProjectIdTestBehavior#getExpectedQuotaProjectInHeader()} + * indicates the expectation that the QUOTA_USER_PROJECT_HEADER should not be present in the + * export headers. + * + *

{@code true} for {@link QuotaProjectIdTestBehavior#getIsQuotaProjectPresentInMetadata()} + * indicates that the mocked credentials are configured to provide DUMMY_GCP_QUOTA_PROJECT_ID as + * the quota project ID. + */ + private static Stream provideQuotaBehaviorTestCases() { + return Stream.of( + // If quota project present in metadata, it will be used + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(DUMMY_GCP_QUOTA_PROJECT_ID) + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("my-custom-quota-project-id") + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + // If quota project not present in request metadata, then user specified project is used + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(DUMMY_GCP_QUOTA_PROJECT_ID) + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("my-custom-quota-project-id") + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader("my-custom-quota-project-id") + .build()), + // Testing for special edge case inputs + // user-specified quota project is empty + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("") // user explicitly specifies empty + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("") + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader(null) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(null) // user omits specifying quota project + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(null) + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader(null) + .build())); + } + + // Configure necessary behavior on the gRPC mock span exporters to work. + // Mockito.lenient is used here because this method is used with parameterized tests where based + // on certain inputs, certain stubbings may not be required. + private static void configureGrpcMockSpanExporter( + OtlpGrpcSpanExporter mockGrpcExporter, + OtlpGrpcSpanExporterBuilder spyGrpcExporterBuilder, + List exportedSpanContainer) { + Mockito.lenient().when(spyGrpcExporterBuilder.build()).thenReturn(mockGrpcExporter); + Mockito.lenient() + .when(mockGrpcExporter.shutdown()) + .thenReturn(CompletableResultCode.ofSuccess()); + Mockito.lenient().when(mockGrpcExporter.toBuilder()).thenReturn(spyGrpcExporterBuilder); + Mockito.lenient() + .when(mockGrpcExporter.export(Mockito.anyCollection())) + .thenAnswer( + invocationOnMock -> { + exportedSpanContainer.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + } + + // Configure necessary behavior on the http mock metric exporters to work. + private static void configureHttpMockMetricExporter( + OtlpHttpMetricExporter mockOtlpHttpMetricExporter, + OtlpHttpMetricExporterBuilder spyOtlpHttpMetricExporterBuilder, + List exportedMetricContainer) { + Mockito.when(spyOtlpHttpMetricExporterBuilder.build()).thenReturn(mockOtlpHttpMetricExporter); + Mockito.when(mockOtlpHttpMetricExporter.shutdown()) + .thenReturn(CompletableResultCode.ofSuccess()); + Mockito.when(mockOtlpHttpMetricExporter.toBuilder()) + .thenReturn(spyOtlpHttpMetricExporterBuilder); + Mockito.when(mockOtlpHttpMetricExporter.export(Mockito.anyCollection())) + .thenAnswer( + invocationOnMock -> { + exportedMetricContainer.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + // mock the get default aggregation and aggregation temporality - they're required for valid + // metric collection. + Mockito.when(mockOtlpHttpMetricExporter.getDefaultAggregation(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpHttpMetricExporter.getDefault().getDefaultAggregation(instrumentType); + }); + Mockito.when(mockOtlpHttpMetricExporter.getAggregationTemporality(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpHttpMetricExporter.getDefault() + .getAggregationTemporality(instrumentType); + }); + } + + // Configure necessary behavior on the gRPC mock metrics exporters to work. + // Mockito.lenient is used here because this method is used with parameterized tests where based + // on certain inputs, certain stubbings may not be required. + private static void configureGrpcMockMetricExporter( + OtlpGrpcMetricExporter mockOtlpGrpcMetricExporter, + OtlpGrpcMetricExporterBuilder spyOtlpGrpcMetricExporterBuilder, + List exportedMetricContainer) { + Mockito.lenient() + .when(spyOtlpGrpcMetricExporterBuilder.build()) + .thenReturn(mockOtlpGrpcMetricExporter); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.shutdown()) + .thenReturn(CompletableResultCode.ofSuccess()); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.toBuilder()) + .thenReturn(spyOtlpGrpcMetricExporterBuilder); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.export(Mockito.anyCollection())) + .thenAnswer( + invocationOnMock -> { + exportedMetricContainer.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + // mock the get default aggregation and aggregation temporality - they're required for valid + // metric collection. + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.getDefaultAggregation(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpGrpcMetricExporter.getDefault().getDefaultAggregation(instrumentType); + }); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.getAggregationTemporality(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpGrpcMetricExporter.getDefault() + .getAggregationTemporality(instrumentType); + }); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.getMemoryMode()) + .thenReturn(MemoryMode.IMMUTABLE_DATA); + } + + @AutoValue + abstract static class ProjectIdTestBehavior { + // A null user specified project ID represents the use case where user omits specifying it + @Nullable + abstract String getUserSpecifiedProjectId(); + + // The project ID that credentials.getProjectId() returns (can be null) + @Nullable + abstract String getCredentialsProjectId(); + + // The expected project ID in the resource attributes (null if exception expected) + @Nullable + abstract String getExpectedProjectIdInResource(); + + // Whether an exception is expected to be thrown + abstract boolean getExpectedToThrow(); + + static Builder builder() { + return new AutoValue_GcpAuthAutoConfigurationCustomizerProviderTest_ProjectIdTestBehavior + .Builder(); + } + + @AutoValue.Builder + abstract static class Builder { + abstract Builder setUserSpecifiedProjectId(String projectId); + + abstract Builder setCredentialsProjectId(String projectId); + + abstract Builder setExpectedProjectIdInResource(String projectId); + + abstract Builder setExpectedToThrow(boolean expectedToThrow); + + abstract ProjectIdTestBehavior build(); + } + } + + @AutoValue + abstract static class QuotaProjectIdTestBehavior { + // A null user specified quota represents the use case where user omits specifying quota + @Nullable + abstract String getUserSpecifiedQuotaProjectId(); + + abstract boolean getIsQuotaProjectPresentInMetadata(); + + // If expected quota project in header is null, the header entry should not be present in export + @Nullable + abstract String getExpectedQuotaProjectInHeader(); + + static Builder builder() { + return new AutoValue_GcpAuthAutoConfigurationCustomizerProviderTest_QuotaProjectIdTestBehavior + .Builder(); + } + + @AutoValue.Builder + abstract static class Builder { + abstract Builder setUserSpecifiedQuotaProjectId(String quotaProjectId); + + abstract Builder setIsQuotaProjectPresentInMetadata(boolean quotaProjectPresentInMetadata); + + /** + * Sets the expected quota project header value for the test case. A null value is allowed, + * and it indicates that the header should not be present in the export request. + * + * @param expectedQuotaProjectInHeader the expected header value to match in the export + * headers. + */ + abstract Builder setExpectedQuotaProjectInHeader(String expectedQuotaProjectInHeader); + + abstract QuotaProjectIdTestBehavior build(); + } + } + + @AutoValue + abstract static class TargetSignalBehavior { + @Nonnull + abstract String getConfiguredTargetSignals(); + + @Nonnull + abstract ImmutableMap getUserSpecifiedOtelProperties(); + + abstract boolean getExpectedIsTraceSignalModified(); + + abstract boolean getExpectedIsMetricsSignalModified(); + + static Builder builder() { + return new AutoValue_GcpAuthAutoConfigurationCustomizerProviderTest_TargetSignalBehavior + .Builder(); + } + + @AutoValue.Builder + abstract static class Builder { + abstract Builder setConfiguredTargetSignals(String targetSignals); + + abstract Builder setUserSpecifiedOtelProperties(Map oTelProperties); + + // Set whether the combination of specified OTel properties and configured target signals + // should lead to modification of the OTLP trace exporters. + abstract Builder setExpectedIsTraceSignalModified(boolean expectedModified); + + // Set whether the combination of specified OTel properties and configured target signals + // should lead to modification of the OTLP metrics exporters. + abstract Builder setExpectedIsMetricsSignalModified(boolean expectedModified); + + abstract TargetSignalBehavior build(); + } + } + + // Mockito.lenient is used here because this method is used with parameterized tests where based + @SuppressWarnings("CannotMockMethod") + private void prepareMockBehaviorForGoogleCredentials() { + AccessToken fakeAccessToken = new AccessToken("fake", Date.from(Instant.now())); + try { + Mockito.lenient() + .when(mockedGoogleCredentials.getRequestMetadata()) + .thenReturn( + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue()), + QUOTA_USER_PROJECT_HEADER, + Collections.singletonList(DUMMY_GCP_QUOTA_PROJECT_ID))); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter(SpanExporter spanExporter) { + return buildOpenTelemetrySdkWithExporter( + spanExporter, OtlpHttpMetricExporter.getDefault(), defaultOtelPropertiesSpanExporter); + } + + @SuppressWarnings("UnusedMethod") + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter( + SpanExporter spanExporter, ImmutableMap customOTelProperties) { + return buildOpenTelemetrySdkWithExporter( + spanExporter, OtlpHttpMetricExporter.getDefault(), customOTelProperties); + } + + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter(MetricExporter metricExporter) { + return buildOpenTelemetrySdkWithExporter( + OtlpHttpSpanExporter.getDefault(), metricExporter, defaultOtelPropertiesMetricExporter); + } + + @SuppressWarnings("UnusedMethod") + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter( + MetricExporter metricExporter, ImmutableMap customOtelProperties) { + return buildOpenTelemetrySdkWithExporter( + OtlpHttpSpanExporter.getDefault(), metricExporter, customOtelProperties); + } + + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter( + SpanExporter spanExporter, + MetricExporter metricExporter, + ImmutableMap customOtelProperties) { + SpiHelper spiHelper = + SpiHelper.create(GcpAuthAutoConfigurationCustomizerProviderTest.class.getClassLoader()); + AutoConfiguredOpenTelemetrySdkBuilder builder = + AutoConfiguredOpenTelemetrySdk.builder() + .addPropertiesSupplier(() -> customOtelProperties) + .setComponentLoader( + new ComponentLoader() { + @Override + public List load(Class spiClass) { + if (spiClass == ConfigurableSpanExporterProvider.class) { + return Collections.singletonList( + spiClass.cast( + new ConfigurableSpanExporterProvider() { + @Override + public SpanExporter createExporter( + ConfigProperties configProperties) { + return spanExporter; + } + + @Override + public String getName() { + return "otlp"; + } + })); + } + if (spiClass == ConfigurableMetricExporterProvider.class) { + return Collections.singletonList( + spiClass.cast( + new ConfigurableMetricExporterProvider() { + @Override + public MetricExporter createExporter( + ConfigProperties configProperties) { + return metricExporter; + } + + @Override + public String getName() { + return "otlp"; + } + })); + } + return spiHelper.load(spiClass); + } + }); + + return builder.build().getOpenTelemetrySdk(); + } + + private static boolean authHeadersQuotaProjectIsPresent(Map headers) { + Set> headerEntrySet = headers.entrySet(); + return headerEntrySet.contains( + new SimpleEntry<>( + QUOTA_USER_PROJECT_HEADER, + GcpAuthAutoConfigurationCustomizerProviderTest.DUMMY_GCP_QUOTA_PROJECT_ID)) + && headerEntrySet.contains(new SimpleEntry<>("Authorization", "Bearer fake")); + } + + private static void generateTestSpan(OpenTelemetrySdk openTelemetrySdk) { + Span span = openTelemetrySdk.getTracer("test").spanBuilder("sample").startSpan(); + try (Scope ignored = span.makeCurrent()) { + long workOutput = busyloop(); + span.setAttribute("work_loop", workOutput); + } finally { + span.end(); + } + } + + private static void generateTestMetric(OpenTelemetrySdk openTelemetrySdk) { + LongCounter longCounter = + openTelemetrySdk + .getMeter("test") + .counterBuilder("sample") + .setDescription("sample counter") + .setUnit("1") + .build(); + long workOutput = busyloop(); + long randomValue = TEST_RANDOM.nextInt(1000); + longCounter.add(randomValue, Attributes.of(AttributeKey.longKey("work_loop"), workOutput)); + } + + // loop to simulate work done + private static long busyloop() { + Instant start = Instant.now(); + Instant end; + long counter = 0; + do { + counter++; + end = Instant.now(); + } while (Duration.between(start, end).toMillis() < 1000); + return counter; + } +} diff --git a/settings.gradle.kts b/settings.gradle.kts index 603832045f3e..443d9c567752 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -192,6 +192,7 @@ include(":sdks:java:extensions:avro") include(":sdks:java:extensions:euphoria") include(":sdks:java:extensions:kryo") include(":sdks:java:extensions:google-cloud-platform-core") +include(":sdks:java:extensions:opentelemetry-gcp-auth-extension") include(":sdks:java:extensions:jackson") include(":sdks:java:extensions:join-library") include(":sdks:java:extensions:kafka-factories") From bc2ff6207c501f004bd99805507b8ab1b330dbda Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Tue, 2 Jun 2026 16:40:57 +0200 Subject: [PATCH 2/2] cleanup --- .../build.gradle | 1 + ...thAutoConfigurationCustomizerProvider.java | 162 ++++++++--------- .../opentelemetry/gcp/auth/package-info.java | 20 +++ ...re.spi.AutoConfigurationCustomizerProvider | 15 ++ ...toConfigurationCustomizerProviderTest.java | 166 +++++++++--------- 5 files changed, 198 insertions(+), 166 deletions(-) create mode 100644 sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/package-info.java diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle b/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle index 7a750597a87e..d69d67e1268a 100644 --- a/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle @@ -28,6 +28,7 @@ dependencies { implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom) implementation platform(library.java.opentelemetry_bom) implementation library.java.google_auth_library_oauth2_http + implementation library.java.vendored_guava_32_1_2_jre compileOnly library.java.opentelemetry_api compileOnly library.java.opentelemetry_extension_autoconfigure compileOnly library.java.opentelemetry_exporter_otlp diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java index 42af9d4fcb58..a37f83d93c71 100644 --- a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java @@ -26,13 +26,9 @@ import com.google.auto.service.AutoService; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter; -import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder; import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter; -import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder; import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter; -import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder; import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; -import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder; import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider; import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; @@ -45,9 +41,8 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; -import java.util.logging.Level; -import java.util.logging.Logger; import javax.annotation.Nonnull; +import javax.annotation.Nullable; import org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GoogleAuthException.Reason; /** @@ -70,15 +65,6 @@ public class GcpAuthAutoConfigurationCustomizerProvider implements AutoConfigurationCustomizerProvider { - private static final Logger logger = - Logger.getLogger(GcpAuthAutoConfigurationCustomizerProvider.class.getName()); - private static final String SIGNAL_TARGET_WARNING_FIX_SUGGESTION = - String.format( - "You may safely ignore this warning if it is intentional, otherwise please configure the '%s' by exporting valid values to environment variable: %s or by setting valid values in system property: %s.", - ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getUserReadableName(), - ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getEnvironmentVariable(), - ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty()); - static final String QUOTA_USER_PROJECT_HEADER = "x-goog-user-project"; static final String GCP_USER_PROJECT_ID_KEY = "gcp.project_id"; @@ -86,12 +72,13 @@ public class GcpAuthAutoConfigurationCustomizerProvider static final String SIGNAL_TYPE_METRICS = "metrics"; static final String SIGNAL_TYPE_ALL = "all"; + private @Nullable GoogleCredentials credentials; + /** * Customizes the provided {@link AutoConfigurationCustomizer} such that authenticated exports to * GCP Telemetry API are possible from the configured OTLP exporter. * - *

This method attempts to retrieve Google Application Default Credentials (ADC) and performs - * the following: + *

This method performs the following: * *

    *
  • Verifies whether the configured OTLP endpoint (base or signal specific) is a known GCP @@ -107,29 +94,13 @@ public class GcpAuthAutoConfigurationCustomizerProvider * enable GCP integration. * * @param autoConfiguration the AutoConfigurationCustomizer to customize. - * @throws GoogleAuthException if there's an error retrieving Google Application Default - * Credentials. - * @throws io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException if required options are - * not configured through environment variables or system properties. */ @Override public void customize(@Nonnull AutoConfigurationCustomizer autoConfiguration) { - GoogleCredentials credentials; - try { - credentials = GoogleCredentials.getApplicationDefault(); - } catch (IOException e) { - throw new GoogleAuthException(Reason.FAILED_ADC_RETRIEVAL, e); - } autoConfiguration - .addSpanExporterCustomizer( - (spanExporter, configProperties) -> - customizeSpanExporter(spanExporter, credentials, configProperties)) - .addMetricExporterCustomizer( - (metricExporter, configProperties) -> - customizeMetricExporter(metricExporter, credentials, configProperties)) - .addResourceCustomizer( - (resource, configProperties) -> - customizeResource(resource, credentials, configProperties)); + .addSpanExporterCustomizer(this::customizeSpanExporter) + .addMetricExporterCustomizer(this::customizeMetricExporter) + .addResourceCustomizer(this::customizeResource); } @Override @@ -137,30 +108,29 @@ public int order() { return Integer.MAX_VALUE - 1; } - private static SpanExporter customizeSpanExporter( - SpanExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + private synchronized GoogleCredentials getCredentials() { + if (credentials == null) { + try { + credentials = GoogleCredentials.getApplicationDefault(); + } catch (IOException e) { + throw new GoogleAuthException(Reason.FAILED_ADC_RETRIEVAL, e); + } + } + return credentials; + } + + private SpanExporter customizeSpanExporter( + SpanExporter exporter, ConfigProperties configProperties) { if (isSignalTargeted(SIGNAL_TYPE_TRACES, configProperties)) { - return addAuthorizationHeaders(exporter, credentials, configProperties); - } else { - String[] params = {SIGNAL_TYPE_TRACES, SIGNAL_TARGET_WARNING_FIX_SUGGESTION}; - logger.log( - Level.WARNING, - "GCP Authentication Extension is not configured for signal type: {0}. {1}", - params); + return addAuthorizationHeaders(exporter, configProperties); } return exporter; } - private static MetricExporter customizeMetricExporter( - MetricExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + private MetricExporter customizeMetricExporter( + MetricExporter exporter, ConfigProperties configProperties) { if (isSignalTargeted(SIGNAL_TYPE_METRICS, configProperties)) { - return addAuthorizationHeaders(exporter, credentials, configProperties); - } else { - String[] params = {SIGNAL_TYPE_METRICS, SIGNAL_TARGET_WARNING_FIX_SUGGESTION}; - logger.log( - Level.WARNING, - "GCP Authentication Extension is not configured for signal type: {0}. {1}", - params); + return addAuthorizationHeaders(exporter, configProperties); } return exporter; } @@ -171,11 +141,25 @@ private static boolean isSignalTargeted(String checkSignal, ConfigProperties con if (endpoint == null) { endpoint = configProperties.getString("otel.exporter.otlp.endpoint"); } - if (endpoint == null - || (!endpoint.startsWith("https://telemetry.googleapis.com") - && !endpoint.startsWith("https://telemetry.mtls.googleapis.com"))) { + if (endpoint == null) { return false; } + + try { + java.net.URI uri = new java.net.URI(endpoint); + String host = uri.getHost(); + String scheme = uri.getScheme(); + if (host == null + || scheme == null + || !scheme.equalsIgnoreCase("https") + || (!host.equalsIgnoreCase("telemetry.googleapis.com") + && !host.equalsIgnoreCase("telemetry.mtls.googleapis.com"))) { + return false; + } + } catch (java.net.URISyntaxException e) { + return false; + } + String userSpecifiedTargetedSignals = ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getConfiguredValueWithFallback( configProperties, () -> SIGNAL_TYPE_ALL); @@ -186,57 +170,74 @@ private static boolean isSignalTargeted(String checkSignal, ConfigProperties con targetedSignal.equals(checkSignal) || targetedSignal.equals(SIGNAL_TYPE_ALL)); } + private boolean isAnySignalTargeted(ConfigProperties configProperties) { + return isSignalTargeted(SIGNAL_TYPE_TRACES, configProperties) + || isSignalTargeted(SIGNAL_TYPE_METRICS, configProperties); + } + // Adds authorization headers to the calls made by the OtlpGrpcSpanExporter and // OtlpHttpSpanExporter. - private static SpanExporter addAuthorizationHeaders( - SpanExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + private SpanExporter addAuthorizationHeaders( + SpanExporter exporter, ConfigProperties configProperties) { if (exporter instanceof OtlpHttpSpanExporter) { - OtlpHttpSpanExporterBuilder builder = + SpanExporter result = ((OtlpHttpSpanExporter) exporter) .toBuilder() - .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); - return builder.build(); + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; } else if (exporter instanceof OtlpGrpcSpanExporter) { - OtlpGrpcSpanExporterBuilder builder = + SpanExporter result = ((OtlpGrpcSpanExporter) exporter) .toBuilder() - .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); - return builder.build(); + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; } return exporter; } // Adds authorization headers to the calls made by the OtlpGrpcMetricExporter and // OtlpHttpMetricExporter. - private static MetricExporter addAuthorizationHeaders( - MetricExporter exporter, GoogleCredentials credentials, ConfigProperties configProperties) { + private MetricExporter addAuthorizationHeaders( + MetricExporter exporter, ConfigProperties configProperties) { if (exporter instanceof OtlpHttpMetricExporter) { - OtlpHttpMetricExporterBuilder builder = + MetricExporter result = ((OtlpHttpMetricExporter) exporter) .toBuilder() - .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); - return builder.build(); + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; } else if (exporter instanceof OtlpGrpcMetricExporter) { - OtlpGrpcMetricExporterBuilder builder = + MetricExporter result = ((OtlpGrpcMetricExporter) exporter) .toBuilder() - .setHeaders(() -> getRequiredHeaderMap(credentials, configProperties)); - return builder.build(); + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; } return exporter; } - private static Map getRequiredHeaderMap( - GoogleCredentials credentials, ConfigProperties configProperties) { + private Map getRequiredHeaderMap(ConfigProperties configProperties) { + GoogleCredentials creds = getCredentials(); Map> gcpHeaders; try { // this also refreshes the credentials, if required - gcpHeaders = credentials.getRequestMetadata(); + gcpHeaders = creds.getRequestMetadata(); } catch (IOException e) { throw new GoogleAuthException(Reason.FAILED_ADC_REFRESH, e); } + if (gcpHeaders == null) { + return Map.of(); + } Map flattenedHeaders = gcpHeaders.entrySet().stream() + .filter(entry -> entry.getKey() != null && entry.getValue() != null) .collect( toMap( Map.Entry::getKey, @@ -259,13 +260,16 @@ private static Map getRequiredHeaderMap( } // Updates the current resource with the attributes required for ingesting OTLP data on GCP. - private static Resource customizeResource( - Resource resource, GoogleCredentials credentials, ConfigProperties configProperties) { + private Resource customizeResource(Resource resource, ConfigProperties configProperties) { + if (!isAnySignalTargeted(configProperties)) { + return resource; + } + String gcpProjectId; try { gcpProjectId = ConfigurableOption.GOOGLE_CLOUD_PROJECT.getConfiguredValue(configProperties); } catch (ConfigurationException e) { - gcpProjectId = credentials.getProjectId(); + gcpProjectId = getCredentials().getProjectId(); if (gcpProjectId == null || gcpProjectId.isEmpty()) { throw e; } diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/package-info.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/package-info.java new file mode 100644 index 000000000000..e389a1a7dc9e --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/package-info.java @@ -0,0 +1,20 @@ +/* + * 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. + */ + +/** Google Cloud Platform (GCP) OpenTelemetry (OTLP) authentication extension. */ +package org.apache.beam.sdk.extensions.opentelemetry.gcp.auth; diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider index 105921f0b4d3..bf1ba2cad985 100644 --- a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider @@ -1 +1,16 @@ +# 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. + org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java index c825a48cc4b3..e1db55c3635b 100644 --- a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java @@ -31,7 +31,6 @@ import com.google.auth.oauth2.AccessToken; import com.google.auth.oauth2.GoogleCredentials; import com.google.auto.value.AutoValue; -import com.google.common.collect.ImmutableMap; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.metrics.LongCounter; @@ -80,6 +79,7 @@ import java.util.stream.Stream; import javax.annotation.Nonnull; import javax.annotation.Nullable; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -95,7 +95,7 @@ import org.mockito.stubbing.Answer; /** - * Copied from + * Copied from open-telemetry. Link: * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/test/java/io/opentelemetry/contrib/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java */ class GcpAuthAutoConfigurationCustomizerProviderTest { @@ -109,31 +109,23 @@ class GcpAuthAutoConfigurationCustomizerProviderTest { @Captor private ArgumentCaptor>> traceHeaderSupplierCaptor; @Captor private ArgumentCaptor>> metricHeaderSupplierCaptor; - private static final ImmutableMap defaultOtelPropertiesSpanExporter = - ImmutableMap.of( - "otel.exporter.otlp.traces.endpoint", - "https://telemetry.googleapis.com/v1/traces", - "otel.traces.exporter", - "otlp", - "otel.metrics.exporter", - "none", - "otel.logs.exporter", - "none", - "otel.resource.attributes", - "foo=bar"); - - private static final ImmutableMap defaultOtelPropertiesMetricExporter = - ImmutableMap.of( - "otel.exporter.otlp.metrics.endpoint", - "https://telemetry.googleapis.com/v1/metrics", - "otel.traces.exporter", - "none", - "otel.metrics.exporter", - "otlp", - "otel.logs.exporter", - "none", - "otel.resource.attributes", - "foo=bar"); + private static final ImmutableMap DEFAULT_OTEL_PROPERTIES_SPAN_EXPORTER = + ImmutableMap.builder() + .put("otel.exporter.otlp.traces.endpoint", "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "none") + .put("otel.logs.exporter", "none") + .put("otel.resource.attributes", "foo=bar") + .build(); + + private static final ImmutableMap DEFAULT_OTEL_PROPERTIES_METRIC_EXPORTER = + ImmutableMap.builder() + .put("otel.exporter.otlp.metrics.endpoint", "https://telemetry.googleapis.com/v1/metrics") + .put("otel.traces.exporter", "none") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .put("otel.resource.attributes", "foo=bar") + .build(); @BeforeEach public void setup() { @@ -628,20 +620,20 @@ void testTargetSignalsBehavior(TargetSignalBehavior testCase) { } } - /** Test cases specifying expected behavior for GOOGLE_OTEL_AUTH_TARGET_SIGNALS */ + /** Test cases specifying expected behavior for GOOGLE_OTEL_AUTH_TARGET_SIGNALS. */ private static Stream provideTargetSignalBehaviorTestCases() { return Stream.of( Arguments.of( TargetSignalBehavior.builder() .setConfiguredTargetSignals("traces") - .setUserSpecifiedOtelProperties(defaultOtelPropertiesSpanExporter) + .setUserSpecifiedOtelProperties(DEFAULT_OTEL_PROPERTIES_SPAN_EXPORTER) .setExpectedIsMetricsSignalModified(false) .setExpectedIsTraceSignalModified(true) .build()), Arguments.of( TargetSignalBehavior.builder() .setConfiguredTargetSignals("metrics") - .setUserSpecifiedOtelProperties(defaultOtelPropertiesMetricExporter) + .setUserSpecifiedOtelProperties(DEFAULT_OTEL_PROPERTIES_METRIC_EXPORTER) .setExpectedIsMetricsSignalModified(true) .setExpectedIsTraceSignalModified(false) .build()), @@ -649,17 +641,17 @@ private static Stream provideTargetSignalBehaviorTestCases() { TargetSignalBehavior.builder() .setConfiguredTargetSignals("all") .setUserSpecifiedOtelProperties( - ImmutableMap.of( - "otel.exporter.otlp.metrics.endpoint", - "https://telemetry.googleapis.com/v1/metrics", - "otel.exporter.otlp.traces.endpoint", - "https://telemetry.googleapis.com/v1/traces", - "otel.traces.exporter", - "otlp", - "otel.metrics.exporter", - "otlp", - "otel.logs.exporter", - "none")) + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) .setExpectedIsMetricsSignalModified(true) .setExpectedIsTraceSignalModified(true) .build()), @@ -667,17 +659,17 @@ private static Stream provideTargetSignalBehaviorTestCases() { TargetSignalBehavior.builder() .setConfiguredTargetSignals("metrics, traces") .setUserSpecifiedOtelProperties( - ImmutableMap.of( - "otel.exporter.otlp.metrics.endpoint", - "https://telemetry.googleapis.com/v1/metrics", - "otel.exporter.otlp.traces.endpoint", - "https://telemetry.googleapis.com/v1/traces", - "otel.traces.exporter", - "otlp", - "otel.metrics.exporter", - "otlp", - "otel.logs.exporter", - "none")) + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) .setExpectedIsMetricsSignalModified(true) .setExpectedIsTraceSignalModified(true) .build()), @@ -685,17 +677,17 @@ private static Stream provideTargetSignalBehaviorTestCases() { TargetSignalBehavior.builder() .setConfiguredTargetSignals("") .setUserSpecifiedOtelProperties( - ImmutableMap.of( - "otel.exporter.otlp.metrics.endpoint", - "https://telemetry.googleapis.com/v1/metrics", - "otel.exporter.otlp.traces.endpoint", - "https://telemetry.googleapis.com/v1/traces", - "otel.traces.exporter", - "otlp", - "otel.metrics.exporter", - "otlp", - "otel.logs.exporter", - "none")) + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) .setExpectedIsMetricsSignalModified(true) .setExpectedIsTraceSignalModified(true) .build()), @@ -721,17 +713,17 @@ private static Stream provideTargetSignalBehaviorTestCases() { TargetSignalBehavior.builder() .setConfiguredTargetSignals("metric, trace") .setUserSpecifiedOtelProperties( - ImmutableMap.of( - "otel.exporter.otlp.metrics.endpoint", - "https://telemetry.googleapis.com/v1/metrics", - "otel.exporter.otlp.traces.endpoint", - "https://telemetry.googleapis.com/v1/traces", - "otel.traces.exporter", - "otlp", - "otel.metrics.exporter", - "otlp", - "otel.logs.exporter", - "none")) + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) .setExpectedIsMetricsSignalModified(false) .setExpectedIsTraceSignalModified(false) .build()), @@ -739,17 +731,17 @@ private static Stream provideTargetSignalBehaviorTestCases() { TargetSignalBehavior.builder() .setConfiguredTargetSignals("metrics, trace") .setUserSpecifiedOtelProperties( - ImmutableMap.of( - "otel.exporter.otlp.metrics.endpoint", - "https://telemetry.googleapis.com/v1/metrics", - "otel.exporter.otlp.traces.endpoint", - "https://telemetry.googleapis.com/v1/traces", - "otel.traces.exporter", - "otlp", - "otel.metrics.exporter", - "otlp", - "otel.logs.exporter", - "none")) + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) .setExpectedIsMetricsSignalModified(true) .setExpectedIsTraceSignalModified(false) .build())); @@ -1122,7 +1114,7 @@ private void prepareMockBehaviorForGoogleCredentials() { private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter(SpanExporter spanExporter) { return buildOpenTelemetrySdkWithExporter( - spanExporter, OtlpHttpMetricExporter.getDefault(), defaultOtelPropertiesSpanExporter); + spanExporter, OtlpHttpMetricExporter.getDefault(), DEFAULT_OTEL_PROPERTIES_SPAN_EXPORTER); } @SuppressWarnings("UnusedMethod") @@ -1134,7 +1126,7 @@ private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter( private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter(MetricExporter metricExporter) { return buildOpenTelemetrySdkWithExporter( - OtlpHttpSpanExporter.getDefault(), metricExporter, defaultOtelPropertiesMetricExporter); + OtlpHttpSpanExporter.getDefault(), metricExporter, DEFAULT_OTEL_PROPERTIES_METRIC_EXPORTER); } @SuppressWarnings("UnusedMethod")