Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,10 @@
import java.util.logging.Level;
import java.util.logging.LogManager;
import java.util.logging.Logger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.beam.sdk.util.InstanceBuilder;
import org.apache.beam.sdk.util.Secret;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.checkerframework.checker.index.qual.NonNegative;
Expand Down Expand Up @@ -467,11 +470,19 @@ public Duration create(PipelineOptions options) {

/**
* The OpenTelemetry properties that will be appended to the set of system properties for SDK
* harness instances. Property names must be specified without the 'otel.' prefix.
* harness instances.
*
* <p>Property values may contain inline secret placeholders formatted as {@code
* ${type:<secret_type>;<secret_param>:<value>}} (see {@link Secret#parseSecretOption(String)}),
* which are resolved on the worker before setting system properties. For example: {@code
* otel.exporter.otlp.metrics.headers=api-key=${type:GcpSecret;version_name:projects/my-project/secrets/my-secret/versions/latest}}.
* Secret parameter values may contain {@code ':'} (such as ARNs or URLs), but cannot contain
* {@code ';'} or {@code '}'}.
*/
@Description(
"The OpenTelemetry properties that will be appended to the set of system properties for SDK "
+ "harness instances. Property names must be specified without the 'otel.' prefix.")
+ "harness instances. Property values may reference secrets using "
+ "${type:<secret_type>;<secret_param>:<value>} (parameter values cannot contain ';' or '}').")
Map<String, String> getOpenTelemetryProperties();

void setOpenTelemetryProperties(Map<String, String> value);
Expand All @@ -483,7 +494,7 @@ static void configureOpenTelemetry(PipelineOptions options) {
openTelemetryProperties.forEach(
(k, v) -> {
if (k != null && v != null) {
System.setProperty(k, v);
System.setProperty(k, resolveOpenTelemetryPropertyValue(v));
}
});
if (System.getProperty("otel.service.instance.id") == null) {
Expand All @@ -499,6 +510,33 @@ static void configureOpenTelemetry(PipelineOptions options) {
}
}

/**
* Resolves any inline {@code ${...}} secret placeholders in an OpenTelemetry property value using
* {@link Secret#parseSecretOption(String)}.
*/
@VisibleForTesting
static String resolveOpenTelemetryPropertyValue(String value) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if this would be better as a generic function packaged in util alongside Secret? Right now, it is just used by OpenTelemetry, but I could imagine it being broadly useful, and it would be nice to make sure we do this consistently.

WDYT?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, just missed mentioning this before the merge. It is fine to leave or move, I will defer to you

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will move it in follow up pr, thanks for feedback

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I agree. Moving this logic to a generic utility alongside Secret would be a great way to ensure we handle this consistently across IOs that accept a map of configuration, like icebergIO and kafkaIO.

This PR also highlights that using an inline secret string format is much more user-friendly than attempting to parse nested JSON. However, I am afraid this will confuse users since in some places we support JSON secret spec and others support inline secret string.

Do you consider it as a problem @damccorm ?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the UX here is not particularly problematic (it mostly puts some work on us to handle it correctly, which I think is fine)

Matcher matcher = GlobalOpenTelemetryFactory.SECRET_PLACEHOLDER_PATTERN.matcher(value);
if (!matcher.find()) {
return value;
}
// Use StringBuffer for Java 8 compatibility with Matcher#appendReplacement.
StringBuffer resolved = new StringBuffer();
do {
String secretOption = checkStateNotNull(matcher.group(1));
String secretValue = Secret.parseSecretOption(secretOption).getString(true);
if (secretValue == null) {
throw new IllegalStateException(
"Resolved secret value cannot be null for secret option: " + secretOption);
}
// Trim surrounding whitespace/newlines commonly stored in Secret Manager payloads and quote
// regex replacement metacharacters ('$' and '\') so they are not interpreted as group refs.
matcher.appendReplacement(resolved, Matcher.quoteReplacement(secretValue.trim()));
} while (matcher.find());
matcher.appendTail(resolved);
return resolved.toString();
}

@JsonIgnore
@Hidden
@Default.InstanceFactory(GlobalOpenTelemetryFactory.class)
Expand All @@ -507,6 +545,8 @@ static void configureOpenTelemetry(PipelineOptions options) {
void setOpenTelemetry(OpenTelemetry value);

class GlobalOpenTelemetryFactory implements DefaultValueFactory<OpenTelemetry> {
private static final Pattern SECRET_PLACEHOLDER_PATTERN = Pattern.compile("\\$\\{([^}]+)\\}");

@Override
public OpenTelemetry create(PipelineOptions options) {
return GlobalOpenTelemetry.get();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,20 +19,26 @@

import static org.apache.beam.sdk.options.SdkHarnessOptions.LogLevel.WARN;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThrows;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.beam.sdk.options.SdkHarnessOptions.DefaultMaxCacheMemoryUsageMb;
import org.apache.beam.sdk.options.SdkHarnessOptions.SdkHarnessLogLevelOverrides;
import org.apache.beam.sdk.testing.RestoreSystemProperties;
import org.apache.beam.sdk.util.common.ReflectHelpers;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestRule;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/** Tests for {@link SdkHarnessOptions}. */
@RunWith(JUnit4.class)
public class SdkHarnessOptionsTest {
@Rule public TestRule restoreSystemProperties = new RestoreSystemProperties();

private static final ObjectMapper MAPPER =
new ObjectMapper()
.registerModules(ObjectMapper.findModules(ReflectHelpers.findClassLoader()));
Expand Down Expand Up @@ -112,4 +118,77 @@ public void testSerializationAndDeserializationOf() throws Exception {
testValue,
MAPPER.writeValueAsString(MAPPER.readValue(testValue, SdkHarnessLogLevelOverrides.class)));
}

@Test
public void testResolveOpenTelemetryPropertyValuePlain() {
assertEquals(
"https://otlp.example.com/v1/metrics",
SdkHarnessOptions.resolveOpenTelemetryPropertyValue("https://otlp.example.com/v1/metrics"));
}

@Test
public void testResolveOpenTelemetryPropertyValueWithSecretPlaceholder() {
assertEquals(
"api-key=my-secret-api-key,env=prod",
SdkHarnessOptions.resolveOpenTelemetryPropertyValue(
"api-key=${type:TestSecret;secret:my-secret-api-key},env=prod"));
assertEquals(
"api-key=my-secret-api-key,env=prod",
SdkHarnessOptions.resolveOpenTelemetryPropertyValue(
"api-key=${secret:my-secret-api-key;type:TestSecret},env=prod"));
assertEquals(
"endpoint=https://otlp.example.com:4318",
SdkHarnessOptions.resolveOpenTelemetryPropertyValue(
"endpoint=${type:TestSecret;secret:https://otlp.example.com:4318}"));
}

@Test
public void testResolveOpenTelemetryPropertyValueTrimsAndQuotesSpecialChars() {
assertEquals(
"api-key=sec$1\\path,token=tok2",
SdkHarnessOptions.resolveOpenTelemetryPropertyValue(
"api-key=${type:TestSecret;secret: sec$1\\path\n},token=${type:testsecret;secret:tok2\r\n}"));
}

@Test
public void testResolveOpenTelemetryPropertyValueWithInvalidSecretThrows() {
assertThrows(
IllegalArgumentException.class,
() ->
SdkHarnessOptions.resolveOpenTelemetryPropertyValue(
"api-key=${type:UnknownSecretType;secret:foo}"));
assertThrows(
IllegalArgumentException.class,
() -> SdkHarnessOptions.resolveOpenTelemetryPropertyValue("api-key=${secret:foo}"));
}

@Test
public void testConfigureOpenTelemetrySetsResolvedPropertiesAndDoesNotMutateOptions() {
System.clearProperty("otel.exporter.otlp.metrics.headers");
System.clearProperty("otel.service.instance.id");

SdkHarnessOptions options = PipelineOptionsFactory.as(SdkHarnessOptions.class);
String rawHeaderValue = "api-key=${type:TestSecret;secret:secret-api-key}";
options.setOpenTelemetryProperties(
ImmutableMap.of("otel.exporter.otlp.metrics.headers", rawHeaderValue));

SdkHarnessOptions.configureOpenTelemetry(options);

assertEquals(
"api-key=secret-api-key", System.getProperty("otel.exporter.otlp.metrics.headers"));
assertNotNull(System.getProperty("otel.service.instance.id"));
assertEquals(
rawHeaderValue,
options.getOpenTelemetryProperties().get("otel.exporter.otlp.metrics.headers"));
}

@Test
public void testConfigureOpenTelemetryWhenEmptyDisablesAuthTargetSignals() {
System.clearProperty("google.otel.auth.target.signals");
SdkHarnessOptions options = PipelineOptionsFactory.as(SdkHarnessOptions.class);

SdkHarnessOptions.configureOpenTelemetry(options);

assertEquals("none", System.getProperty("google.otel.auth.target.signals"));
}
}
Loading