Run Calico E2E with the Kubernetes Agent backend

calico/tests/conftest.py

@@ -5,22 +5,49 @@
 
 import pytest
 
-from datadog_checks.dev.conditions import CheckEndpoints, WaitFor
+from datadog_checks.dev.conditions import WaitFor
 from datadog_checks.dev.kind import kind_run
-from datadog_checks.dev.kube_port_forward import port_forward
 from datadog_checks.dev.subprocess import run_command
 
 from .common import EXTRA_METRICS
 
 NAMESPACE = "calico"
 HERE = path.dirname(path.abspath(__file__))
+FELIX_METRICS_ENDPOINT = 'http://felix-metrics-svc.kube-system.svc.cluster.local:9091/metrics'
 
 
 def _felix_config_default_exists():
     result = run_command(["kubectl", "get", "felixconfiguration", "default"], capture='both')
     return result.code == 0
 
 
+def felix_metrics_available() -> bool:
+    result = run_command(
+        [
+            "kubectl",
+            "run",
+            "felix-metrics-readiness",
+            "--namespace",
+            "kube-system",
+            "--image=busybox:1.36.1",
+            "--restart=Never",
+            "--attach",
+            "--rm",
+            "--quiet",
+            "--",
+            "wget",
+            "-q",
+            "-T",
+            "2",
+            "-O",
+            "/dev/null",
+            FELIX_METRICS_ENDPOINT,
+        ],
+        capture='both',
+    )
+    return result.code == 0
+
+
 def setup_calico():
     # Deploy calico
     run_command(["kubectl", "apply", "-f", path.join(HERE, 'kind', 'calico.yaml')])
@@ -60,29 +87,23 @@ def setup_calico():
         check=True,
     )
 
+    # Check from a temporary pod because the host cannot resolve Kubernetes Service DNS.
+    WaitFor(felix_metrics_available, wait=2, attempts=100)()
+
 
 @pytest.fixture(scope='session')
 def dd_environment():
-    with (
-        kind_run(
-            conditions=[setup_calico], kind_config=path.join(HERE, 'kind', 'kind-calico.yaml'), sleep=10
-        ) as kubeconfig,
-        port_forward(kubeconfig, 'kube-system', 9091, 'service', 'felix-metrics-svc') as (
-            calico_host,
-            calico_port,
-        ),
-    ):
-        endpoint = 'http://{}:{}/metrics'.format(calico_host, calico_port)
-
-        # We can't add this to `kind_run` because we don't know the URL at this moment
-        condition = CheckEndpoints(endpoint, wait=2, attempts=100)
-        condition()
-
-        yield {
-            "openmetrics_endpoint": endpoint,
+    with kind_run(
+        conditions=[setup_calico], kind_config=path.join(HERE, 'kind', 'kind-calico.yaml'), sleep=10
+    ) as kubeconfig:
+        instance = {
+            "openmetrics_endpoint": FELIX_METRICS_ENDPOINT,
             "namespace": NAMESPACE,
             "extra_metrics": EXTRA_METRICS,
         }
+        metadata = {'agent_type': 'kubernetes', 'kubernetes': {'kubeconfig': kubeconfig}}
+
+        yield instance, metadata
 
 
 @pytest.fixture

datadog_checks_dev/changelog.d/24639.added

@@ -0,0 +1 @@
+Add backend-neutral Agent log retrieval for E2E test diagnostics.

datadog_checks_dev/datadog_checks/dev/plugin/pytest.py

@@ -219,7 +219,10 @@ def run_check(config=None, **kwargs):
 
         if not matches:
             message_parts = []
-            debug_result = run_command(['docker', 'logs', 'dd_{}_{}'.format(check, env)], capture=True)
+            debug_result = run_command(
+                [python_path, '-m', 'ddev', 'env', 'logs', check, env],
+                capture=True,
+            )
             if not debug_result.code:
                 message_parts.append(debug_result.stdout + debug_result.stderr)
 

ddev/changelog.d/24639.added

@@ -0,0 +1 @@
+Add a Kubernetes Agent interface for running the Agent in Kind-based E2E test clusters, alongside the existing Docker and Vagrant interfaces.

ddev/src/ddev/cli/env/__init__.py

@@ -6,6 +6,7 @@
 from ddev.cli.env.agent import agent
 from ddev.cli.env.check import check
 from ddev.cli.env.config import config
+from ddev.cli.env.logs import logs
 from ddev.cli.env.reload import reload_command
 from ddev.cli.env.shell import shell
 from ddev.cli.env.show import show
@@ -24,6 +25,7 @@ def env():
 env.add_command(agent)
 env.add_command(check)
 env.add_command(config)
+env.add_command(logs)
 env.add_command(reload_command)
 env.add_command(shell)
 env.add_command(show)

ddev/src/ddev/cli/env/agent.py

@@ -49,6 +49,18 @@ def _validate_env_vars(ctx: click.Context, param: click.Parameter, value: tuple[
     return env_vars or None
 
 
+def _sync_restored_config(app: Application, agent: AgentInterface) -> None:
+    import sys
+
+    original_error = sys.exception()
+    try:
+        agent.sync_config()
+    except Exception as e:
+        if original_error is None:
+            raise
+        app.display_warning(f'Unable to restore the Agent configuration: {e}')
+
+
 @click.command(
     short_help='Invoke the Agent', context_settings={'help_option_names': [], 'ignore_unknown_options': True}
 )
@@ -79,9 +91,8 @@ def agent(
     """
     import subprocess
 
-    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.agent import create_agent_interface
     from ddev.e2e.config import EnvDataStorage
-    from ddev.e2e.constants import DEFAULT_AGENT_TYPE, E2EMetadata
     from ddev.utils.fs import Path
 
     integration = app.repo.integrations.get(intg_name)
@@ -91,8 +102,7 @@ def agent(
         app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
 
     metadata = env_data.read_metadata()
-    agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
-    agent = get_agent_interface(agent_type)(app, integration, environment, metadata, env_data.config_file)
+    agent = create_agent_interface(app, integration, environment, metadata, env_data.config_file)
 
     full_args = list(args)
     trigger_run = False
@@ -131,6 +141,7 @@ def agent(
             app.abort(str(e))
         finally:
             env_data.config_file.unlink()
+            _sync_restored_config(app, agent)
     else:
         temp_config_file = env_data.config_file.parent / f'{env_data.config_file.name}.bak.example'
         env_data.config_file.replace(temp_config_file)
@@ -141,3 +152,4 @@ def agent(
             app.abort(str(e))
         finally:
             temp_config_file.replace(env_data.config_file)
+            _sync_restored_config(app, agent)

ddev/src/ddev/cli/env/logs.py

@@ -0,0 +1,35 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from ddev.cli.application import Application
+
+
+@click.command('logs', short_help='Show logs for the Agent')
+@click.argument('intg_name', metavar='INTEGRATION')
+@click.argument('environment')
+@click.pass_obj
+def logs(app: Application, *, intg_name: str, environment: str):
+    """Show backend-specific diagnostics for the Agent."""
+    from ddev.e2e.agent import create_agent_interface
+    from ddev.e2e.config import EnvDataStorage
+
+    integration = app.repo.integrations.get(intg_name)
+    env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
+
+    if not env_data.exists():
+        app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
+
+    metadata = env_data.read_metadata()
+    agent = create_agent_interface(app, integration, environment, metadata, env_data.config_file)
+
+    try:
+        agent.show_logs()
+    except Exception as e:
+        app.abort(str(e))

ddev/src/ddev/cli/env/reload.py

@@ -19,9 +19,8 @@ def reload_command(app: Application, *, intg_name: str, environment: str):
     """
     Restart the Agent to detect environment changes.
     """
-    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.agent import create_agent_interface
     from ddev.e2e.config import EnvDataStorage
-    from ddev.e2e.constants import DEFAULT_AGENT_TYPE, E2EMetadata
 
     integration = app.repo.integrations.get(intg_name)
     env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
@@ -30,8 +29,7 @@ def reload_command(app: Application, *, intg_name: str, environment: str):
         app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
 
     metadata = env_data.read_metadata()
-    agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
-    agent = get_agent_interface(agent_type)(app, integration, environment, metadata, env_data.config_file)
+    agent = create_agent_interface(app, integration, environment, metadata, env_data.config_file)
 
     try:
         agent.restart()

ddev/src/ddev/cli/env/shell.py

@@ -21,9 +21,8 @@ def shell(app: Application, *, intg_name: str, environment: str):
     """
     import subprocess
 
-    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.agent import create_agent_interface
     from ddev.e2e.config import EnvDataStorage
-    from ddev.e2e.constants import DEFAULT_AGENT_TYPE, E2EMetadata
 
     integration = app.repo.integrations.get(intg_name)
     env_data = EnvDataStorage(app.data_dir).get(integration.name, environment)
@@ -32,8 +31,7 @@ def shell(app: Application, *, intg_name: str, environment: str):
         app.abort(f'Environment `{environment}` for integration `{integration.name}` is not running')
 
     metadata = env_data.read_metadata()
-    agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
-    agent = get_agent_interface(agent_type)(app, integration, environment, metadata, env_data.config_file)
+    agent = create_agent_interface(app, integration, environment, metadata, env_data.config_file)
 
     try:
         agent.enter_shell()

ddev/src/ddev/cli/env/show.py

@@ -72,7 +72,7 @@ def show(app: Application, *, intg_name: str | None, environment: str | None, fo
         app.display_table('Available', available_columns, show_lines=True, force_ascii=force_ascii)
     # Display information about a specific environment
     else:
-        from ddev.e2e.agent import get_agent_interface
+        from ddev.e2e.agent import create_agent_interface
 
         integration = app.repo.integrations.get(intg_name)
         env_data = storage.get(integration.name, environment)
@@ -82,7 +82,7 @@ def show(app: Application, *, intg_name: str | None, environment: str | None, fo
 
         metadata = env_data.read_metadata()
         agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
-        agent = get_agent_interface(agent_type)(app, integration, environment, metadata, env_data.config_file)
+        agent = create_agent_interface(app, integration, environment, metadata, env_data.config_file)
 
         app.display_pair('Agent type', agent_type)
         app.display_pair('Agent ID', agent.get_id())

ddev/src/ddev/cli/env/start.py

@@ -132,6 +132,7 @@ def start(
         result = json.loads(result_file.read_text())
 
     metadata = result['metadata']
+    agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
 
     # TODO Remove once we have migrated the `docker_run` function
     if serialized_volumes := metadata.get(E2EMetadata.ENV_VARS, {}).get(E2EEnvVars.DOCKER_VOLUMES):
@@ -144,24 +145,26 @@ def start(
     config = result['config']
     env_data.write_config(config)
 
-    agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
-
-    if agent_type == "vagrant" and running_on_ci():
-        app.abort(text="Vagrant is not supported on CI", code=0)
+    agent_class = get_agent_interface(agent_type)
+    if running_on_ci() and not agent_class.supports_ci:
+        app.abort(text=f'{agent_type.capitalize()} is not supported on CI', code=0)
 
-    agent = get_agent_interface(agent_type)(app, integration, environment, metadata, env_data.config_file)
+    agent = agent_class(app, integration, environment, metadata, env_data.config_file)
 
     if not agent_build:
+        configured_agent_build = agent.get_configured_build(app.config.agent.config)
         agent_build = (
             os.getenv(E2EEnvVars.AGENT_BUILD_PY2 if agent.python_version[0] == 2 else E2EEnvVars.AGENT_BUILD)
-            or app.config.agent.config.get(agent_type)
+            or configured_agent_build
             or ''
         )
 
     agent_env_vars = _get_agent_env_vars(app.config.org.config, metadata, extra_env_vars, dogstatsd)
 
     try:
         agent.start(agent_build=agent_build, local_packages=local_packages, env_vars=agent_env_vars)
+        # Backends may add runtime metadata needed by later ddev processes.
+        env_data.write_metadata(metadata)
     except Exception as e:
         from ddev.cli.env.stop import stop
 

ddev/src/ddev/cli/env/stop.py

@@ -20,9 +20,9 @@ def stop(app: Application, *, intg_name: str, environment: str, ignore_state: bo
     """
     Stop environments. To stop all the running environments, use `all` as the integration name and the environment.
     """
-    from ddev.e2e.agent import get_agent_interface
+    from ddev.e2e.agent import create_agent_interface
     from ddev.e2e.config import EnvDataStorage
-    from ddev.e2e.constants import DEFAULT_AGENT_TYPE, E2EEnvVars, E2EMetadata
+    from ddev.e2e.constants import E2EEnvVars, E2EMetadata
     from ddev.e2e.run import E2EEnvironmentRunner
     from ddev.utils.fs import temp_directory
 
@@ -63,8 +63,7 @@ def stop(app: Application, *, intg_name: str, environment: str, ignore_state: bo
                 metadata = env_data.read_metadata()
                 env_vars.update(metadata.get(E2EMetadata.ENV_VARS, {}))
 
-                agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
-                agent = get_agent_interface(agent_type)(app, integration, env, metadata, env_data.config_file)
+                agent = create_agent_interface(app, integration, env, metadata, env_data.config_file)
 
                 try:
                     agent.stop()

ddev/src/ddev/e2e/agent/__init__.py

@@ -3,10 +3,26 @@
 # Licensed under a 3-clause BSD style license (see LICENSE)
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
+    from ddev.cli.application import Application
     from ddev.e2e.agent.interface import AgentInterface
+    from ddev.integration.core import Integration
+    from ddev.utils.fs import Path
+
+
+def create_agent_interface(
+    app: Application,
+    integration: Integration,
+    environment: str,
+    metadata: dict[str, Any],
+    config_file: Path,
+) -> AgentInterface:
+    from ddev.e2e.constants import DEFAULT_AGENT_TYPE, E2EMetadata
+
+    agent_type = metadata.get(E2EMetadata.AGENT_TYPE, DEFAULT_AGENT_TYPE)
+    return get_agent_interface(agent_type)(app, integration, environment, metadata, config_file)
 
 
 def get_agent_interface(agent_type: str) -> type[AgentInterface]:
@@ -20,4 +36,9 @@ def get_agent_interface(agent_type: str) -> type[AgentInterface]:
 
         return VagrantAgent
 
+    if agent_type == "kubernetes":
+        from ddev.e2e.agent.kubernetes import KubernetesAgent
+
+        return KubernetesAgent
+
     raise NotImplementedError(f"Unsupported Agent type: {agent_type}")