[dynamic control] Add file reader with line-by-line policy provider

dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/LinePerPolicyFileProvider.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.logging.Logger;
+import java.util.stream.Stream;
+
+/**
+ * A {@link PolicyProvider} that reads policies from a local file, where each line represents a
+ * separate policy configuration.
+ *
+ * <p>The file format supports two types of lines:
+ *
+ * <ul>
+ *   <li><b>JSON Objects:</b> Lines starting with <code>{</code> are treated as JSON objects and
+ *       validated against the registered {@link PolicyValidator}s.
+ *   <li><b>Key-Value Pairs:</b> Lines in the format <code>key=value</code> are treated as aliases,
+ *       where the key matches a validator's {@link PolicyValidator#getAlias()} and the value is
+ *       parsed accordingly.
+ * </ul>
+ *
+ * <p>Empty lines and lines starting with <code>#</code> are ignored.
+ */
+final class LinePerPolicyFileProvider implements PolicyProvider {
+  private static final Logger logger = Logger.getLogger(LinePerPolicyFileProvider.class.getName());
+  private final Path file;
+  private final List<PolicyValidator> validators;
+
+  public LinePerPolicyFileProvider(Path file, List<PolicyValidator> validators) {
+    Objects.requireNonNull(file, "file cannot be null");
+    this.file = file;
+    this.validators = new ArrayList<>(validators);
+  }
+
+  @Override
+  public List<TelemetryPolicy> fetchPolicies() throws IOException {
+    List<TelemetryPolicy> policies = new ArrayList<>();
+    if (!Files.exists(file)) {
+      logger.info("Policy file does not exist: " + file);
+      return policies;
+    }
+
+    try (Stream<String> lines = Files.lines(file)) {
+      lines.forEach(
+          line -> {
+            String trimmedLine = line.trim();
+            if (trimmedLine.isEmpty() || trimmedLine.startsWith("#")) {
+              return;
+            }
+
+            TelemetryPolicy policy = null;
+
+            if (trimmedLine.startsWith("{")) {
+              for (PolicyValidator validator : validators) {
+                if (trimmedLine.contains("\"" + validator.getPolicyType() + "\"")) {
+                  policy = validator.validate(trimmedLine);
+                  if (policy != null) {
+                    break;
+                  }
+                }
+              }
+            } else {
+              int idx = trimmedLine.indexOf('=');
+              if (idx > 0) {
+                String key = trimmedLine.substring(0, idx).trim();
+                String valueStr = trimmedLine.substring(idx + 1).trim();
+
+                for (PolicyValidator validator : validators) {
+                  String alias = validator.getAlias();
+                  if (alias != null && alias.equals(key)) {
+                    try {
+                      policy = validator.validateAlias(key, valueStr);
+                    } catch (UnsupportedOperationException e) {
+                      logger.info(
+                          "Validator does not support alias validation: "
+                              + validator.getClass().getName());
+                      continue;
+                    }
+                    if (policy != null) {
+                      break;
+                    }
+                  }
+                }
+              }
+            }
+
+            if (policy == null) {
+              logger.info("Validator not found or rejected for line: " + trimmedLine);
+              return;
+            }
+
+            policies.add(policy);
+          });
+    }
+    return policies;
+  }
+}

dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/PolicyProvider.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy;
+
+import java.io.Closeable;
+import java.util.List;
+import java.util.function.Consumer;
+
+public interface PolicyProvider {
+  /**
+   * Retrieves the current list of telemetry policies provided by this source.
+   *
+   * @return A list of {@link TelemetryPolicy} objects.
+   * @throws Exception if an error occurs while fetching the policies.
+   */
+  List<TelemetryPolicy> fetchPolicies() throws Exception;
+
+  /**
+   * Starts a mechanism to push policy updates to the provided consumer.
+   *
+   * <p>This method is optional and may be a no-op for providers that only support polling via
+   * {@link #fetchPolicies()}.
+   *
+   * @param onUpdate A consumer that accepts the new list of policies when an update occurs.
+   * @return A {@link Closeable} that stops watching when closed.
+   */
+  default Closeable startWatching(Consumer<List<TelemetryPolicy>> onUpdate) {
+    return () -> {};
+  }
+}

dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/LinePerPolicyFileProviderTest.java

@@ -0,0 +1,135 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class LinePerPolicyFileProviderTest {
+
+  private static final String TRACE_SAMPLING_TYPE = "trace-sampling";
+  private static final String TRACE_SAMPLING_ALIAS = "trace-sampling.probability";
+
+  @TempDir Path tempDir;
+
+  @Test
+  void fetchPoliciesReturnsEmptyWhenFileMissing() throws Exception {
+    Path missingFile = tempDir.resolve("missing-policies.txt");
+    LinePerPolicyFileProvider provider =
+        new LinePerPolicyFileProvider(missingFile, Collections.singletonList(acceptingValidator()));
+
+    List<TelemetryPolicy> policies = provider.fetchPolicies();
+
+    assertThat(policies).isEmpty();
+  }
+
+  @Test
+  void fetchPoliciesParsesJsonLines() throws Exception {
+    Path file = writeLines("{\"trace-sampling\": {\"probability\": 0.5}}");
+    LinePerPolicyFileProvider provider =
+        new LinePerPolicyFileProvider(file, Collections.singletonList(acceptingValidator()));
+
+    List<TelemetryPolicy> policies = provider.fetchPolicies();
+
+    assertThat(policies).hasSize(1);
+    assertThat(policies.get(0).getType()).isEqualTo(TRACE_SAMPLING_TYPE);
+  }
+
+  @Test
+  void fetchPoliciesParsesAliasLines() throws Exception {
+    Path file = writeLines("trace-sampling.probability=0.5");
+    LinePerPolicyFileProvider provider =
+        new LinePerPolicyFileProvider(file, Collections.singletonList(acceptingValidator()));
+
+    List<TelemetryPolicy> policies = provider.fetchPolicies();
+
+    assertThat(policies).hasSize(1);
+    assertThat(policies.get(0).getType()).isEqualTo(TRACE_SAMPLING_TYPE);
+  }
+
+  @Test
+  void fetchPoliciesSkipsBlankLinesAndComments() throws Exception {
+    Path file = writeLines("", "   ", "# comment line", "trace-sampling.probability=0.25");
+    LinePerPolicyFileProvider provider =
+        new LinePerPolicyFileProvider(file, Collections.singletonList(acceptingValidator()));
+
+    List<TelemetryPolicy> policies = provider.fetchPolicies();
+
+    assertThat(policies).hasSize(1);
+    assertThat(policies.get(0).getType()).isEqualTo(TRACE_SAMPLING_TYPE);
+  }
+
+  @Test
+  void fetchPoliciesSkipsUnknownOrRejectedPolicies() throws Exception {
+    PolicyValidator rejectingValidator =
+        new TestPolicyValidator(/* acceptJson= */ false, /* acceptAlias= */ false);
+    Path file =
+        writeLines(
+            "{\"trace-sampling\": {\"probability\": 0.5}}",
+            "{\"other-policy\": {\"probability\": 0.5}}",
+            "other.key=1");
+    LinePerPolicyFileProvider provider =
+        new LinePerPolicyFileProvider(file, Collections.singletonList(rejectingValidator));
+
+    List<TelemetryPolicy> policies = provider.fetchPolicies();
+
+    assertThat(policies).isEmpty();
+  }
+
+  private Path writeLines(String... lines) throws IOException {
+    Path file = tempDir.resolve("policies.txt");
+    Files.write(file, Arrays.asList(lines));
+    return file;
+  }
+
+  private static PolicyValidator acceptingValidator() {
+    return new TestPolicyValidator(/* acceptJson= */ true, /* acceptAlias= */ true);
+  }
+
+  private static class TestPolicyValidator implements PolicyValidator {
+    private final boolean acceptJson;
+    private final boolean acceptAlias;
+
+    private TestPolicyValidator(boolean acceptJson, boolean acceptAlias) {
+      this.acceptJson = acceptJson;
+      this.acceptAlias = acceptAlias;
+    }
+
+    @Override
+    public TelemetryPolicy validate(String json) {
+      if (!acceptJson) {
+        return null;
+      }
+      return new TelemetryPolicy(TRACE_SAMPLING_TYPE, null);
+    }
+
+    @Override
+    public String getPolicyType() {
+      return TRACE_SAMPLING_TYPE;
+    }
+
+    @Override
+    public TelemetryPolicy validateAlias(String key, String value) {
+      if (!acceptAlias) {
+        return null;
+      }
+      return new TelemetryPolicy(TRACE_SAMPLING_TYPE, null);
+    }
+
+    @Override
+    public String getAlias() {
+      return TRACE_SAMPLING_ALIAS;
+    }
+  }
+}