[FLINK-36456] feat(datadog): prefer DD_API_KEY env var, update tests and docs

Author: barryib

docs/content.zh/docs/deployment/metric_reporters.md

@@ -270,7 +270,7 @@ metrics.reporter.stsd.interval: 60 SECONDS
 
 参数:
 
-- `apikey` - Datadog 的 API KEY。
+- `apikey` - Datadog 的 API KEY。（或者设置环境变量 `DD_API_KEY` 来覆盖配置）
 - `proxyHost` - （可选的）发送到 Datadog 时使用的代理主机。
 - `proxyPort` - （可选的）发送到 Datadog 时使用的代理端口，默认为 8080。
 - `dataCenter` - （可选的）要连接的数据中心（`EU`/`US`），默认为 `US`。

docs/content/docs/deployment/metric_reporters.md

@@ -254,7 +254,7 @@ In contrast to Datadog-provided Histograms the reported aggregations are not com
 
 Parameters:
 
-- `apikey` - the Datadog API key
+- `apikey` - the Datadog API key (or set environment variable `DD_API_KEY` to override)
 - `proxyHost` - (optional) The proxy host to use when sending to Datadog.
 - `proxyPort` - (optional) The proxy port to use when sending to Datadog, defaults to 8080.
 - `dataCenter` - (optional) The data center (`EU`/`US`) to connect to, defaults to `US`.

flink-metrics/flink-metrics-datadog/src/main/java/org/apache/flink/metrics/datadog/DatadogHttpReporterFactory.java

@@ -32,7 +32,7 @@ public class DatadogHttpReporterFactory implements MetricReporterFactory {
     private static final Logger LOG = LoggerFactory.getLogger(DatadogHttpReporterFactory.class);
 
     private static final String API_KEY = "apikey";
-    private static final String API_KEY_ENV_VAR = "DATADOG_API_KEY";
+    private static final String API_KEY_ENV_VAR = "DD_API_KEY";
     private static final String PROXY_HOST = "proxyHost";
     private static final String PROXY_PORT = "proxyPort";
     private static final String DATA_CENTER = "dataCenter";
@@ -68,23 +68,27 @@ public MetricReporter createMetricReporter(Properties config) {
     }
 
     /**
-     * Retrieves the Datadog API key from configuration or environment variable.
-     *
-     * <p>The API key is resolved in the following order:
-     *
-     * <ol>
-     *   <li>Configuration property "apikey"
-     *   <li>Environment variable "DATADOG_API_KEY"
-     * </ol>
+    * Retrieves the Datadog API key from environment variable or configuration.
+    *
+    * <p>The API key is resolved in the following order:
+    *
+    * <ol>
+    *   <li>Environment variable "DD_API_KEY"
+    *   <li>Configuration property "apikey"
+    * </ol>
      *
      * @param config the configuration properties
      * @return the Datadog API key, or null if not found in either location
      */
     private String getApiKey(Properties config) {
+        final String envApiKey = System.getenv(API_KEY_ENV_VAR);
+        if (envApiKey != null && !envApiKey.isEmpty()) {
+            return envApiKey;
+        }
         String apiKey = config.getProperty(API_KEY, null);
         if (apiKey != null && !apiKey.isEmpty()) {
             return apiKey;
         }
-        return System.getenv(API_KEY_ENV_VAR);
+        return null;
     }
 }

flink-metrics/flink-metrics-datadog/src/test/java/org/apache/flink/metrics/datadog/DatadogHttpReporterFactoryTest.java

@@ -55,6 +55,31 @@ void testNoConfigurationPropertyReturnsNull() throws Exception {
         assertThat(resolvedApiKey).isNull();
     }
 
+    @Test
+    void testApiKeyFromEnvironmentVariable() throws Exception {
+        final String envKey = "env-api-key";
+        setEnv("DD_API_KEY", envKey);
+        try {
+            final String resolvedApiKey = getApiKeyViaReflection(createConfig());
+            assertThat(resolvedApiKey).isEqualTo(envKey);
+        } finally {
+            setEnv("DD_API_KEY", null);
+        }
+    }
+
+    @Test
+    void testEnvironmentVariableTakesPrecedenceOverConfiguration() throws Exception {
+        final String envKey = "env-api-key";
+        final String configKey = "config-api-key";
+        setEnv("DD_API_KEY", envKey);
+        try {
+            final String resolvedApiKey = getApiKeyViaReflection(createConfig("apikey", configKey));
+            assertThat(resolvedApiKey).isEqualTo(envKey);
+        } finally {
+            setEnv("DD_API_KEY", null);
+        }
+    }
+
     private Properties createConfig(String key, String value) {
         Properties config = new Properties();
         config.setProperty(key, value);
@@ -73,4 +98,43 @@ private String getApiKeyViaReflection(Properties config) throws Exception {
         getApiKeyMethod.setAccessible(true);
         return (String) getApiKeyMethod.invoke(factory, config);
     }
+
+    @SuppressWarnings("unchecked")
+    private void setEnv(String key, String value) throws Exception {
+        try {
+            Map<String, String> env = System.getenv();
+            Class<?> cl = env.getClass();
+            java.lang.reflect.Field field = cl.getDeclaredField("m");
+            field.setAccessible(true);
+            Map<String, String> writableEnv = (Map<String, String>) field.get(env);
+            if (value == null) {
+                writableEnv.remove(key);
+            } else {
+                writableEnv.put(key, value);
+            }
+        } catch (NoSuchFieldException e) {
+            // Fallback for other JVM implementations
+            try {
+                Class<?> pe = Class.forName("java.lang.ProcessEnvironment");
+                java.lang.reflect.Field theEnvironmentField = pe.getDeclaredField("theEnvironment");
+                theEnvironmentField.setAccessible(true);
+                Map<String, String> env = (Map<String, String>) theEnvironmentField.get(null);
+                if (value == null) {
+                    env.remove(key);
+                } else {
+                    env.put(key, value);
+                }
+                java.lang.reflect.Field theCaseInsensitiveEnvironmentField = pe.getDeclaredField("theCaseInsensitiveEnvironment");
+                theCaseInsensitiveEnvironmentField.setAccessible(true);
+                Map<String, String> cienv = (Map<String, String>) theCaseInsensitiveEnvironmentField.get(null);
+                if (value == null) {
+                    cienv.remove(key);
+                } else {
+                    cienv.put(key, value);
+                }
+            } catch (ClassNotFoundException | NoSuchFieldException ex) {
+                // ignore; best-effort environment mutation for tests
+            }
+        }
+    }
 }