Support Datadog feature flags

docs/tutorials/cli/create-command.md

@@ -134,3 +134,62 @@ $ dda agent-release data
 │                   │ └───┴────────┘ │
 ...
 ```
+
+## Using feature flags
+
+You can guard behavior behind feature flags using the `app.features` manager. It evaluates a remote flag for the current user/machine with a default fallback.
+
+
+Update the command to check a flag before performing the network request:
+
+/// tab | :octicons-file-code-16: src/dda/cli/agent_release/data/__init__.py
+```python hl_lines="15 23-28 31-36"
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from dda.cli.base import dynamic_command, pass_app
+
+if TYPE_CHECKING:
+    from dda.cli.application import Application
+
+
+@dynamic_command(
+    short_help="Show Agent release data",
+    features=["http"],
+)
+@pass_app
+def cmd(app: Application) -> None:
+    """
+    Show Agent release data.
+    """
+    # Check a feature flag to enable this command's behavior
+    # Replace "agent-release-enabled" with your flag key
+    enabled = app.features.enabled(
+        "agent-release-enabled",
+        default=True,
+        scopes={"module": "agent-release"},
+    )
+    if not enabled:
+        app.display_warning("This command is currently disabled by feature flag.")
+        return
+
+    import httpx
+
+    base = "https://raw.githubusercontent.com"
+    repo = "DataDog/datadog-agent"
+    branch = "main"
+    path = "release.json"
+    with app.status("Fetching Agent release data"):
+        response = httpx.get(f"{base}/{repo}/{branch}/{path}")
+
+    response.raise_for_status()
+    app.display_table(response.json())
+```
+///
+
+/// note
+`app.features.enabled(key, default, scopes)` returns the evaluated value for `key`, or `default` if the flag is not found. The client automatically includes base context like platform, CI, environment, and user; you can add `scopes` to refine targeting.
+///
+
+

src/dda/cli/application.py

@@ -16,6 +16,7 @@
 
     from dda.config.file import ConfigFile
     from dda.config.model import RootConfig
+    from dda.feature_flags.manager import FeatureFlagManager
     from dda.github.core import GitHub
     from dda.telemetry.manager import TelemetryManager
     from dda.tools import Tools
@@ -126,6 +127,12 @@ def telemetry(self) -> TelemetryManager:
 
         return TelemetryManager(self)
 
+    @cached_property
+    def features(self) -> FeatureFlagManager:
+        from dda.feature_flags.manager import FeatureFlagManager
+
+        return FeatureFlagManager(self)
+
     @cached_property
     def dynamic_deps_allowed(self) -> bool:
         return os.getenv(AppEnvVars.NO_DYNAMIC_DEPS) not in {"1", "true"}

src/dda/config/constants.py

@@ -22,6 +22,7 @@ class AppEnvVars:
     VERBOSE = "DDA_VERBOSE"
     NO_DYNAMIC_DEPS = "DDA_NO_DYNAMIC_DEPS"
     TELEMETRY_API_KEY = "DDA_TELEMETRY_API_KEY"
+    FEATURE_FLAGS_CLIENT_TOKEN = "DDA_FEATURE_FLAGS_CLIENT_TOKEN"  # noqa: S105i
     TELEMETRY_USER_MACHINE_ID = "DDA_TELEMETRY_USER_MACHINE_ID"
     # https://no-color.org
     NO_COLOR = "NO_COLOR"

src/dda/feature_flags/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <dev@datadoghq.com>
+#
+# SPDX-License-Identifier: MIT

src/dda/feature_flags/client.py

@@ -0,0 +1,104 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <dev@datadoghq.com>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from dda._version import __version__
+
+if TYPE_CHECKING:
+    from dda.cli.application import Application
+
+
+class DatadogFeatureFlag:
+    """
+    Direct HTTP client for Datadog Feature Flag API
+    """
+
+    def __init__(self, client_token: str | None, app: Application):
+        """
+        Initialize the Datadog Feature Flag client
+
+        Parameters:
+            client_token: Your Datadog client token (starts with 'pub_')
+            app: The application instance
+        """
+        self.__client_token = client_token
+        self.__env = "Production"
+        self.__url = f"https://preview.ff-cdn.datadoghq.com/precompute-assignments?dd_env={self.__env}"
+        self.__app_id = "dda"
+        self.__app = app
+
+    def _fetch_flags(
+        self, targeting_key: str = "", targeting_attributes: dict[str, Any] | None = None
+    ) -> dict[str, Any]:
+        """
+        Fetch flag configuration from Datadog API
+
+        Parameters:
+            targeting_key: The targeting key (typically user ID)
+            targeting_attributes: Additional targeting attributes (context)
+
+        Returns:
+            Dictionary containing the flag configuration response
+        """
+        if not self.__client_token:
+            return {}
+
+        from httpx import HTTPError
+
+        # Build headers
+        headers = {
+            "Content-Type": "application/vnd.api+json",
+            "dd-client-token": self.__client_token,
+            "dd-application-id": self.__app_id,
+        }
+
+        # Build request payload (following JSON:API format)
+        payload = {
+            "data": {
+                "type": "precompute-assignments-request",
+                "attributes": {
+                    "env": {
+                        "dd_env": self.__env,
+                    },
+                    "sdk": {
+                        "name": "dda",
+                        "version": __version__,
+                    },
+                    "subject": {
+                        "targeting_key": targeting_key,
+                        "targeting_attributes": targeting_attributes,
+                    },
+                },
+            },
+        }
+
+        try:
+            # Make the request
+            response = self.__app.http.client().post(self.__url, headers=headers, json=payload)
+        except HTTPError as e:
+            self.__app.display_warning(f"Error fetching flags: {e}")
+            return {}
+
+        return response.json()
+
+    def get_flag_value(self, flag: str, targeting_key: str, targeting_attributes: dict[str, Any]) -> bool | None:
+        """
+        Get a flag value by key
+
+        Parameters:
+            flag: The flag key to evaluate
+            targeting_key: The targeting key to use for feature flag evaluation
+            targeting_attributes: The targeting attributes to use for feature flag evaluation
+
+        Returns:
+            The flag value or None if the flag is not found
+        """
+        response = self._fetch_flags(targeting_key, targeting_attributes)
+        flags = response.get("data", {}).get("attributes", {}).get("flags", {})
+        if flag in flags:
+            return flags[flag].get("variationValue", None)
+
+        return None

src/dda/feature_flags/manager.py

@@ -0,0 +1,99 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <dev@datadoghq.com>
+#
+# SPDX-License-Identifier: MIT
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING, Any, Optional
+
+from dda.feature_flags.client import DatadogFeatureFlag
+from dda.user.datadog import User
+from dda.utils.ci import running_in_ci
+
+if TYPE_CHECKING:
+    from dda.cli.application import Application
+    from dda.config.model import RootConfig
+
+
+class FeatureFlagUser(User):
+    def __init__(self, config: RootConfig) -> None:
+        super().__init__(config)
+
+
+class FeatureFlagManager:
+    """
+    A class for querying feature flags. This is available as the
+    [`Application.features`][dda.cli.application.Application.features] property.
+    """
+
+    def __init__(self, app: Application) -> None:
+        self.__app = app
+        # Manually implemented cache to avoid calling several time Feature flag backend on the same flag evaluation.
+        # Cache key is a tuple of the flag, entity and scopes, to make it hashable.
+        # For example after calling `enabled("test-flag", default=False, scopes={"user": "user1"}),
+        # the cache will contain the result for the tuple ("test-flag", "entity", (("user", "user1"),)).
+        self.__cache: dict[tuple[str, str, tuple[tuple[str, str], ...]], Any] = {}
+
+    @cached_property
+    def __client(self) -> DatadogFeatureFlag | None:
+        if running_in_ci():  # We do not support feature flags token retrieval in the CI yet.
+            return None
+
+        from contextlib import suppress
+
+        from dda.secrets.api import fetch_client_token, read_client_token, save_client_token
+
+        client_token: str | None = None
+        with suppress(Exception):
+            client_token = read_client_token()
+            if not client_token:
+                client_token = fetch_client_token()
+                save_client_token(client_token)
+
+        return DatadogFeatureFlag(client_token, self.__app)
+
+    @property
+    def __user(self) -> FeatureFlagUser:
+        return FeatureFlagUser(self.__app.config)
+
+    def __get_entity(self) -> str:
+        if running_in_ci():
+            import os
+
+            return os.getenv("CI_JOB_ID", "default_job_id")
+
+        return self.__user.machine_id
+
+    def enabled(self, flag: str, *, default: bool = False, scopes: Optional[dict[str, str]] = None) -> bool:
+        entity = self.__get_entity()
+        base_scopes = self.__get_base_scopes()
+        if scopes is not None:
+            base_scopes.update(scopes)
+
+        attributes_items = base_scopes.items()
+        tuple_attributes = tuple(((key, value) for key, value in sorted(attributes_items)))
+
+        self.__app.display_debug(f"Checking flag {flag} with entity {entity} and scopes {base_scopes}")
+        flag_value = self.__check_flag(flag, entity, tuple_attributes)
+        if flag_value is None:
+            return default
+        return flag_value
+
+    def __check_flag(self, flag: str, entity: str, scopes: tuple[tuple[str, str], ...]) -> bool | None:
+        if self.__client is None:
+            self.__app.display_debug("Feature flag client not initialized properly")
+            return None
+
+        cache_key = (flag, entity, scopes)
+        if cache_key in self.__cache:
+            return self.__cache[cache_key]
+
+        flag_value = self.__client.get_flag_value(flag, entity, dict(scopes))
+        self.__cache[cache_key] = flag_value
+        return flag_value
+
+    def __get_base_scopes(self) -> dict[str, str]:
+        return {
+            "ci": "true" if running_in_ci() else "false",
+            "user": self.__user.email,
+        }

src/dda/secrets/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025-present Datadog, Inc. <dev@datadoghq.com>
+#
+# SPDX-License-Identifier: MIT

src/dda/telemetry/secrets.py → src/dda/secrets/api.py

@@ -11,6 +11,7 @@
 
 KEYRING_SERVICE = "dda"
 KEYRING_ITEM_API_KEY = "telemetry_api_key"
+KEYRING_ITEM_CLIENT_TOKEN = "feature_flags_client_token"  # noqa: S105
 
 
 def read_api_key() -> str | None:
@@ -20,11 +21,28 @@ def read_api_key() -> str | None:
     return keyring.get_password(KEYRING_SERVICE, KEYRING_ITEM_API_KEY)
 
 
+def read_client_token() -> str | None:
+    if (client_token := os.environ.get(AppEnvVars.FEATURE_FLAGS_CLIENT_TOKEN)) is not None:
+        return client_token
+
+    return keyring.get_password(KEYRING_SERVICE, KEYRING_ITEM_CLIENT_TOKEN)
+
+
 def save_api_key(api_key: str) -> None:
     keyring.set_password(KEYRING_SERVICE, KEYRING_ITEM_API_KEY, api_key)
 
 
+def save_client_token(client_token: str) -> None:
+    keyring.set_password(KEYRING_SERVICE, KEYRING_ITEM_CLIENT_TOKEN, client_token)
+
+
 def fetch_api_key() -> str:
-    from dda.telemetry.vault import fetch_secret
+    from dda.secrets.vault import fetch_secret
 
     return fetch_secret("group/subproduct-agent/deva", "telemetry-api-key")
+
+
+def fetch_client_token() -> str:
+    from dda.secrets.vault import fetch_secret
+
+    return fetch_secret("group/subproduct-agent/deva", "feature-flags-client-token")

src/dda/telemetry/daemon/__main__.py

@@ -14,9 +14,9 @@
 import psutil
 import watchfiles
 
+from dda.secrets.api import fetch_api_key, read_api_key, save_api_key
 from dda.telemetry.constants import DaemonEnvVars
 from dda.telemetry.daemon.handler import finalize_error
-from dda.telemetry.secrets import fetch_api_key, read_api_key, save_api_key
 from dda.utils.fs import Path
 
 if TYPE_CHECKING: