From 668f3fd0404ca9fa61c63a724a41e42869f49a2d Mon Sep 17 00:00:00 2001 From: Enrico Donnici Date: Wed, 22 Jul 2026 12:29:04 +0000 Subject: [PATCH 1/2] Deploy Kubernetes Agent with Helm --- ddev/src/ddev/e2e/agent/kubernetes.py | 469 ++++++------- ddev/src/ddev/e2e/agent/kubernetes_helm.py | 655 +++++++++++++++++++ ddev/tests/e2e/agent/test_kubernetes.py | 560 +++++++++++++--- ddev/tests/e2e/agent/test_kubernetes_helm.py | 384 +++++++++++ docs/developer/ddev/plugins.md | 2 +- docs/developer/ddev/test.md | 14 +- 6 files changed, 1707 insertions(+), 377 deletions(-) create mode 100644 ddev/src/ddev/e2e/agent/kubernetes_helm.py create mode 100644 ddev/tests/e2e/agent/test_kubernetes_helm.py diff --git a/ddev/src/ddev/e2e/agent/kubernetes.py b/ddev/src/ddev/e2e/agent/kubernetes.py index ea171e1433e45..e204d56b22dcb 100644 --- a/ddev/src/ddev/e2e/agent/kubernetes.py +++ b/ddev/src/ddev/e2e/agent/kubernetes.py @@ -12,23 +12,26 @@ from ddev.e2e.agent.docker import _normalize_agent_image_name from ddev.e2e.agent.interface import AgentInterface +from ddev.e2e.agent.kubernetes_helm import CONTAINER_NAME, AgentPodSelectionError, HelmDaemonSetDeployment if TYPE_CHECKING: import subprocess from ddev.utils.fs import Path -_POD_NAME = 'ddev-agent' -_CONTAINER_NAME = 'agent' _DEFAULT_NAMESPACE_PREFIX = 'ddev-agent' -_DEFAULT_WAIT_TIMEOUT = 120 +_DEFAULT_WAIT_TIMEOUT = 300 _LOCAL_PACKAGES_METADATA = 'local_packages' _OWNER_ID_METADATA = '_kubernetes_owner_id' -_OWNER_LABEL = 'ddev.datadoghq.com/environment' +_PREPARED_MARKER = '/home/.ddev-agent-prepared' + + +class AgentPodReplacedError(RuntimeError): + pass class KubernetesAgent(AgentInterface): - """Run the E2E Agent inside a Kubernetes test cluster. + """Run the E2E Agent in a Helm-managed Kubernetes DaemonSet. The initial implementation intentionally supports one schedulable node and one Agent pod. Cluster creation and image loading belong to the environment @@ -53,7 +56,7 @@ def _kubeconfig(self) -> str: def _namespace(self) -> str: namespace = self._kubernetes_metadata.get('namespace') if namespace is not None: - if not isinstance(namespace, str) or not re.fullmatch(r'[a-z0-9]([-a-z0-9]*[a-z0-9])?', namespace): + if not isinstance(namespace, str) or not re.fullmatch(r'[a-z]([-a-z0-9]*[a-z0-9])?', namespace): raise ValueError(f'Invalid Kubernetes Agent namespace: {namespace!r}') if len(namespace) > 63: raise ValueError('Kubernetes Agent namespace must contain at most 63 characters') @@ -66,10 +69,6 @@ def _namespace(self) -> str: normalized_id = normalized_id[: 63 - len(_DEFAULT_NAMESPACE_PREFIX) - len(digest) - 2].rstrip('-') return f'{_DEFAULT_NAMESPACE_PREFIX}-{normalized_id}-{digest}' - @cached_property - def _cluster_resource_name(self) -> str: - return self._namespace - @cached_property def _owner_id(self) -> str: owner_id = self.metadata.get(_OWNER_ID_METADATA) @@ -79,13 +78,6 @@ def _owner_id(self) -> str: raise ValueError('Kubernetes Agent internal owner ID must contain at most 63 characters') return owner_id - @cached_property - def _resource_labels(self) -> dict[str, str]: - return { - 'app.kubernetes.io/managed-by': 'ddev', - _OWNER_LABEL: self._owner_id, - } - @cached_property def _config_dir(self) -> str: return f'/etc/datadog-agent/conf.d/{self.integration.name}.d' @@ -105,6 +97,18 @@ def _wait_timeout(self) -> int: raise ValueError('Kubernetes Agent `wait_timeout` must be a positive integer') return timeout + @cached_property + def _deployment(self) -> HelmDaemonSetDeployment: + return HelmDaemonSetDeployment( + platform=self.platform, + kubeconfig=self._kubeconfig, + namespace=self._namespace, + owner_id=self._owner_id, + kubernetes_metadata=self._kubernetes_metadata, + state_dir=self.config_file.parent.parent, + wait_timeout=self._wait_timeout, + ) + def _kubectl(self, args: list[str], **kwargs) -> subprocess.CompletedProcess: return self.platform.run_command([*self._kubectl_prefix, *args], **kwargs) @@ -118,10 +122,13 @@ def _captured_kubectl(self, args: list[str], **kwargs) -> subprocess.CompletedPr @staticmethod def _process_output(process: subprocess.CompletedProcess) -> str: - output = process.stdout or b'' - if isinstance(output, bytes): - return output.decode('utf-8', errors='replace') - return output + return HelmDaemonSetDeployment.process_output(process) + + def _pod_name(self, *, ready: bool = True) -> str: + pod = self._deployment.agent_pod(ready=ready) + if pod is None: # pragma: no cover - required=True guarantees a result + raise RuntimeError('Kubernetes Agent pod is unavailable') + return pod.name def _exec( self, @@ -130,8 +137,11 @@ def _exec( env_vars: dict[str, str] | None = None, check: bool = True, capture: bool = False, + pod_ready: bool = True, + pod_name: str | None = None, ) -> subprocess.CompletedProcess: - args = ['exec', '--namespace', self._namespace, f'pod/{_POD_NAME}', '--container', _CONTAINER_NAME, '--'] + pod_name = pod_name or self._pod_name(ready=pod_ready) + args = ['exec', '--namespace', self._namespace, f'pod/{pod_name}', '--container', CONTAINER_NAME, '--'] if env_vars: args.append('env') args.extend(f'{key}={value}' for key, value in sorted(env_vars.items())) @@ -140,29 +150,7 @@ def _exec( return self._captured_kubectl(args, check=check) return self._kubectl(args, check=check) - def _validate_resource_ownership(self) -> None: - resources = ( - ('namespace', self._namespace), - ('clusterrole', self._cluster_resource_name), - ('clusterrolebinding', self._cluster_resource_name), - ) - existing_resources = [] - for resource, name in resources: - process = self._captured_kubectl(['get', resource, name, '--ignore-not-found=true', '-o', 'name']) - if process.returncode: - raise RuntimeError( - f'Unable to inspect Kubernetes Agent {resource} `{name}`: {self._process_output(process)}' - ) - if self._process_output(process).strip(): - existing_resources.append(f'{resource}/{name}') - - if existing_resources: - raise RuntimeError( - 'Refusing to overwrite Kubernetes resources not owned by this environment: ' - + ', '.join(existing_resources) - ) - - def _validate_topology(self) -> None: + def _validate_topology(self) -> str: process = self._captured_kubectl(['get', 'nodes', '-o', 'json']) if process.returncode: raise RuntimeError(f'Unable to inspect Kubernetes nodes: {self._process_output(process)}') @@ -170,190 +158,70 @@ def _validate_topology(self) -> None: try: node_data = json.loads(self._process_output(process)) schedulable_nodes = [node for node in node_data['items'] if not node.get('spec', {}).get('unschedulable')] + node_names = [node['metadata']['name'] for node in schedulable_nodes] except (KeyError, TypeError, json.JSONDecodeError) as e: raise RuntimeError(f'Unable to parse Kubernetes node data: {e}') from e - if len(schedulable_nodes) != 1: + if len(node_names) != 1: raise NotImplementedError( 'KubernetesAgent currently requires exactly one schedulable node; ' - f'found {len(schedulable_nodes)}. Multi-node execution needs an explicit Agent targeting policy.' + f'found {len(node_names)}. Multi-node execution needs an explicit Agent targeting policy.' ) + return node_names[0] - def _manifest(self, agent_build: str, env_vars: dict[str, str]) -> dict[str, Any]: - env_vars = env_vars.copy() - env_vars.setdefault('DD_API_KEY', 'a' * 32) - env_vars.setdefault('DD_APM_ENABLED', 'false') - env_vars.setdefault('DD_AUTOCONFIG_FROM_ENVIRONMENT', 'true') - env_vars.setdefault('DD_HOSTNAME', self._namespace) - env_vars.setdefault('DD_KUBELET_TLS_VERIFY', 'false') - - container_env: list[dict[str, Any]] = [{'name': key, 'value': value} for key, value in sorted(env_vars.items())] - container_env.extend( - [ - { - 'name': 'DD_KUBERNETES_KUBELET_HOST', - 'valueFrom': {'fieldRef': {'fieldPath': 'status.hostIP'}}, - }, - { - 'name': 'DD_KUBERNETES_KUBELET_NODENAME', - 'valueFrom': {'fieldRef': {'fieldPath': 'spec.nodeName'}}, - }, - ] - ) - - extra_labels = self._kubernetes_metadata.get('pod_labels', {}) - if not isinstance(extra_labels, dict) or not all( - isinstance(key, str) and isinstance(value, str) for key, value in extra_labels.items() - ): - raise ValueError('Kubernetes Agent `pod_labels` must be a mapping of strings to strings') - labels = {**extra_labels, **self._resource_labels, 'app.kubernetes.io/name': _POD_NAME} - - image_pull_policy = self._kubernetes_metadata.get('image_pull_policy', 'Always') - if image_pull_policy not in {'Always', 'IfNotPresent', 'Never'}: - raise ValueError('Kubernetes Agent `image_pull_policy` must be Always, IfNotPresent, or Never') - - role_rules = [ - {'apiGroups': [''], 'resources': ['nodes'], 'verbs': ['get', 'list', 'watch']}, - { - 'apiGroups': [''], - 'resources': ['nodes/metrics', 'nodes/spec', 'nodes/stats', 'nodes/proxy'], - 'verbs': ['get'], - }, - { - 'apiGroups': [''], - 'resources': ['pods', 'endpoints', 'services'], - 'verbs': ['get', 'list', 'watch'], - }, - ] - - return { - 'apiVersion': 'v1', - 'kind': 'List', - 'items': [ - { - 'apiVersion': 'v1', - 'kind': 'Namespace', - 'metadata': {'name': self._namespace, 'labels': self._resource_labels}, - }, - { - 'apiVersion': 'v1', - 'kind': 'ServiceAccount', - 'metadata': {'name': _POD_NAME, 'namespace': self._namespace, 'labels': self._resource_labels}, - }, - { - 'apiVersion': 'rbac.authorization.k8s.io/v1', - 'kind': 'ClusterRole', - 'metadata': {'name': self._cluster_resource_name, 'labels': self._resource_labels}, - 'rules': role_rules, - }, - { - 'apiVersion': 'rbac.authorization.k8s.io/v1', - 'kind': 'ClusterRoleBinding', - 'metadata': {'name': self._cluster_resource_name, 'labels': self._resource_labels}, - 'roleRef': { - 'apiGroup': 'rbac.authorization.k8s.io', - 'kind': 'ClusterRole', - 'name': self._cluster_resource_name, - }, - 'subjects': [ - {'kind': 'ServiceAccount', 'name': _POD_NAME, 'namespace': self._namespace}, - ], - }, - { - 'apiVersion': 'v1', - 'kind': 'Pod', - 'metadata': {'name': _POD_NAME, 'namespace': self._namespace, 'labels': labels}, - 'spec': { - 'serviceAccountName': _POD_NAME, - 'restartPolicy': 'Always', - 'terminationGracePeriodSeconds': 0, - 'tolerations': [{'operator': 'Exists'}], - 'containers': [ - { - 'name': _CONTAINER_NAME, - 'image': agent_build, - 'imagePullPolicy': image_pull_policy, - 'env': container_env, - } - ], - }, - }, - ], - } - - def _create_payload(self, payload: dict[str, Any]) -> None: - process = self._captured_kubectl(['create', '-f', '-'], input=json.dumps(payload).encode()) - if process.returncode: - raise RuntimeError(f'Unable to create Kubernetes Agent resources: {self._process_output(process)}') - - def _create_manifest(self, manifest: dict[str, Any]) -> None: - # Acquire the namespace as an atomic ownership lock before creating any - # other resource. `create`, unlike `apply`, fails rather than overwriting - # a namespace that appears after the ownership preflight. - namespace, *resources = manifest['items'] - self._create_payload(namespace) - self._create_payload({'apiVersion': 'v1', 'kind': 'List', 'items': resources}) - - def _wait_for_pod(self) -> None: - process = self._captured_kubectl( - [ - 'wait', - '--namespace', - self._namespace, - '--for=condition=Ready', - f'pod/{_POD_NAME}', - f'--timeout={self._wait_timeout}s', - ] - ) - if process.returncode: - self._show_logs() - raise RuntimeError(f'Kubernetes Agent pod did not become ready: {self._process_output(process)}') - - def _wait_for_agent(self) -> None: + def _wait_for_agent(self, *, pod_name: str | None = None) -> None: deadline = time.monotonic() + self._wait_timeout last_output = '' while time.monotonic() < deadline: - process = self._exec(['agent', 'status'], check=False, capture=True) + process = self._exec(['agent', 'status'], check=False, capture=True, pod_ready=False, pod_name=pod_name) if process.returncode == 0: + self._deployment.wait_for_daemonset() return last_output = self._process_output(process) + if pod_name: + current_pod = self._deployment.agent_pod(ready=False, required=False) + if current_pod is not None and current_pod.name != pod_name: + raise AgentPodReplacedError( + f'Kubernetes Agent pod `{pod_name}` was replaced by `{current_pod.name}`' + ) time.sleep(1) self._show_logs() raise RuntimeError(f'Kubernetes Agent did not become ready: {last_output}') - def _copy_file(self, source: str, destination: str) -> None: + def _copy_file(self, source: str, destination: str, *, pod_name: str | None = None) -> None: from ddev.utils.fs import Path source_path = Path(source).resolve() + pod_name = pod_name or self._pod_name() self._kubectl( [ 'cp', '--container', - _CONTAINER_NAME, + CONTAINER_NAME, source_path.name, - f'{self._namespace}/{_POD_NAME}:{destination}', + f'{self._namespace}/{pod_name}:{destination}', ], check=True, cwd=source_path.parent, ) - def _sync_config(self) -> None: - self._exec(['mkdir', '-p', self._config_dir]) + def _sync_config(self, *, pod_name: str | None = None) -> None: + self._exec(['mkdir', '-p', self._config_dir], pod_name=pod_name) destination = f'{self._config_dir}/conf.yaml' if self.config_file.is_file(): - self._copy_file(str(self.config_file), destination) + self._copy_file(str(self.config_file), destination, pod_name=pod_name) else: - self._exec(['rm', '-f', destination]) + self._exec(['rm', '-f', destination], pod_name=pod_name) - def _sync_auto_conf(self) -> None: + def _sync_auto_conf(self, *, pod_name: str | None = None) -> None: auto_conf = self._kubernetes_metadata.get('auto_conf') if auto_conf is None: return if not isinstance(auto_conf, str) or not auto_conf: raise ValueError('Kubernetes Agent `auto_conf` must be a non-empty path') - self._exec(['mkdir', '-p', self._config_dir]) - self._copy_file(auto_conf, f'{self._config_dir}/auto_conf.yaml') + self._exec(['mkdir', '-p', self._config_dir], pod_name=pod_name) + self._copy_file(auto_conf, f'{self._config_dir}/auto_conf.yaml', pod_name=pod_name) def _remember_local_packages(self, local_packages: dict[Path, str]) -> None: self._kubernetes_metadata[_LOCAL_PACKAGES_METADATA] = [ @@ -375,11 +243,11 @@ def _local_package_specs(self) -> list[dict[str, str]]: raise ValueError(f'Invalid Kubernetes Agent local package name: {spec["name"]!r}') return specs - def _sync_local_packages(self, *, install: bool = False) -> None: + def _sync_local_packages(self, *, install: bool = False, pod_name: str | None = None) -> None: for spec in self._local_package_specs(): destination = f'/home/{spec["name"]}' - self._exec(['rm', '-rf', destination]) - self._copy_file(spec['path'], destination) + self._exec(['rm', '-rf', destination], pod_name=pod_name) + self._copy_file(spec['path'], destination, pod_name=pod_name) if install: self._exec( [ @@ -390,15 +258,40 @@ def _sync_local_packages(self, *, install: bool = False) -> None: '--disable-pip-version-check', '-e', f'{destination}{spec["features"]}', - ] + ], + capture=True, + pod_name=pod_name, ) - def _run_metadata_commands(self, key: str) -> None: + def _run_metadata_commands(self, key: str, *, capture: bool = False, pod_name: str | None = None) -> None: commands = self.metadata.get(key, []) if not isinstance(commands, list) or not all(isinstance(command, str) for command in commands): raise ValueError(f'Kubernetes Agent `{key}` must be a list of commands') for command in commands: - self._exec(self.platform.modules.shlex.split(command)) + self._exec(self.platform.modules.shlex.split(command), capture=capture, pod_name=pod_name) + + def _mark_prepared(self, *, pod_name: str | None = None) -> None: + self._exec(['touch', _PREPARED_MARKER], pod_name=pod_name) + + def _is_prepared(self, *, pod_name: str | None = None) -> bool: + return ( + self._exec(['test', '-f', _PREPARED_MARKER], check=False, capture=True, pod_name=pod_name).returncode == 0 + ) + + def _prepare_container(self, *, capture_commands: bool = False, pod_name: str | None = None) -> None: + pod_name = pod_name or self._pod_name() + self._run_metadata_commands('start_commands', capture=capture_commands, pod_name=pod_name) + self._sync_local_packages(install=True, pod_name=pod_name) + self._sync_config(pod_name=pod_name) + self._sync_auto_conf(pod_name=pod_name) + self._run_metadata_commands('post_install_commands', capture=capture_commands, pod_name=pod_name) + self._restart_agent_process(pod_name=pod_name) + self._mark_prepared(pod_name=pod_name) + + def _ensure_prepared(self) -> None: + pod_name = self._pod_name() + if not self._is_prepared(pod_name=pod_name): + self._prepare_container(pod_name=pod_name) def start(self, *, agent_build: str | None, local_packages: dict[Path, str], env_vars: dict[str, str]) -> None: agent_build = _normalize_agent_image_name( @@ -406,83 +299,60 @@ def start(self, *, agent_build: str | None, local_packages: dict[Path, str], env ) # Validate values needed by teardown before creating any resources. _ = self._owner_id, self._wait_timeout - self._validate_topology() - self._validate_resource_ownership() - self._create_manifest(self._manifest(agent_build, env_vars)) - self._wait_for_pod() - self._run_metadata_commands('start_commands') - self._remember_local_packages(local_packages) - self._sync_local_packages(install=True) - self._sync_config() - self._sync_auto_conf() - self._run_metadata_commands('post_install_commands') - self._restart_agent_process() - - def _resource_is_owned(self, resource: str, name: str) -> bool: - process = self._captured_kubectl(['get', resource, name, '--ignore-not-found=true', '-o', 'json']) - if process.returncode: - output = self._process_output(process) - raise RuntimeError(f'Unable to inspect Kubernetes Agent {resource} `{name}`: {output}') - if not self._process_output(process).strip(): - return False + self._deployment.check_helm() + node_name = self._validate_topology() + values = self._deployment.values(agent_build, env_vars, node_name=node_name) + self._deployment.validate_namespace_absent() + self._deployment.create_namespace() + self._deployment.install(values) try: - data = json.loads(self._process_output(process)) - return data.get('metadata', {}).get('labels', {}).get(_OWNER_LABEL) == self._owner_id - except (AttributeError, json.JSONDecodeError) as e: - raise RuntimeError(f'Unable to parse Kubernetes Agent {resource} `{name}`: {e}') from e + self._deployment.wait_for_daemonset() + pod = self._deployment.agent_pod() + except Exception: + self._show_logs() + raise + if pod is None or pod.node_name != node_name: + actual_node = '' if pod is None else pod.node_name + raise RuntimeError( + f'Kubernetes Agent pod was scheduled on unexpected node {actual_node!r}; expected {node_name!r}' + ) + self._remember_local_packages(local_packages) + self._prepare_container(pod_name=pod.name) def stop(self) -> None: - errors: list[Exception] = [] + if not self._deployment.namespace_is_owned(): + if self._deployment.owned_cluster_resources_exist(): + raise RuntimeError( + 'Kubernetes Agent namespace ownership was lost while Helm-managed cluster-scoped resources remain; ' + 'preserving environment state rather than deleting chart resources outside Helm' + ) + return - # Do not execute commands in a namespace that merely happens to have - # the configured name. Startup may have failed because it was already - # owned by the cluster's caller. + errors: list[Exception] = [] try: - namespace_owned = self._resource_is_owned('namespace', self._namespace) + if pod := self._deployment.agent_pod(ready=False, required=False): + self._run_metadata_commands('stop_commands', pod_name=pod.name) except Exception as e: errors.append(e) - namespace_owned = False - if namespace_owned: - pod = self._captured_kubectl( - ['get', 'pod', _POD_NAME, '--namespace', self._namespace, '--ignore-not-found=true', '-o', 'name'] - ) - if pod.returncode == 0 and self._process_output(pod).strip(): - try: - self._run_metadata_commands('stop_commands') - except Exception as e: - errors.append(e) - - selector = f'{_OWNER_LABEL}={self._owner_id}' - for args in ( - [ - 'delete', - 'namespace', - '--selector', - selector, - '--ignore-not-found=true', - '--wait=true', - f'--timeout={self._wait_timeout}s', - ], - [ - 'delete', - 'clusterrole,clusterrolebinding', - '--selector', - selector, - '--ignore-not-found=true', - ], - ): - process = self._captured_kubectl(args) - if process.returncode: - errors.append( - RuntimeError(f'Unable to remove Kubernetes Agent resources: {self._process_output(process)}') - ) + try: + self._deployment.uninstall() + except Exception as e: + details = '; '.join(str(error) for error in [*errors, e]) + raise RuntimeError( + f'Errors while stopping Kubernetes Agent; preserving the Helm release namespace for retry: {details}' + ) from e + + try: + self._deployment.delete_namespace() + except Exception as e: + errors.append(e) if errors: details = '; '.join(str(error) for error in errors) raise RuntimeError(f'Errors while stopping Kubernetes Agent: {details}') from errors[0] - def _restart_agent_process(self) -> None: + def _restart_agent_process(self, *, pod_name: str | None = None) -> None: # The Agent image's s6 finish handler normally shuts down the whole # service tree when the main Agent exits. Remove it before killing the # process so s6 starts a fresh Agent in the same container, preserving @@ -498,34 +368,75 @@ def _restart_agent_process(self) -> None: 'sleep 1; elapsed=$((elapsed + 1)); ' 'done' ) - self._exec(['sh', '-c', restart_command]) - self._wait_for_agent() + self._exec(['sh', '-c', restart_command], pod_name=pod_name) + self._wait_for_agent(pod_name=pod_name) def restart(self) -> None: - self._sync_local_packages() - self._sync_config() - self._sync_auto_conf() - self._restart_agent_process() + pod_name = self._pod_name() + if not self._is_prepared(pod_name=pod_name): + self._prepare_container(pod_name=pod_name) + return + self._sync_local_packages(pod_name=pod_name) + self._sync_config(pod_name=pod_name) + self._sync_auto_conf(pod_name=pod_name) + self._restart_agent_process(pod_name=pod_name) + self._mark_prepared(pod_name=pod_name) def sync_config(self) -> None: self._sync_config() def invoke(self, args: list[str], *, env_vars: dict[str, str] | None = None) -> None: - self._sync_local_packages() - self._sync_config() - self._sync_auto_conf() - self._exec(['agent', *args], env_vars=env_vars) + prepared_pod = None + for attempt in range(2): + try: + pod = self._deployment.wait_for_agent_pod() + if self._is_prepared(pod_name=pod.name): + self._sync_local_packages(pod_name=pod.name) + self._sync_config(pod_name=pod.name) + self._sync_auto_conf(pod_name=pod.name) + else: + # Container recovery happens inline with the Agent command. Do not + # mix lifecycle command output into machine-readable Agent output. + self._prepare_container(capture_commands=True, pod_name=pod.name) + + current_pod = self._deployment.agent_pod() + except AgentPodSelectionError as e: + if attempt or e.candidate_count: + raise + self._deployment.wait_for_agent_pod() + continue + except (AgentPodReplacedError, self.platform.modules.subprocess.CalledProcessError): + if attempt: + raise + self._deployment.wait_for_agent_pod() + continue + + if current_pod.uid == pod.uid and self._is_prepared(pod_name=current_pod.name): + prepared_pod = current_pod + break + self._deployment.wait_for_agent_pod() + + if prepared_pod is None: + raise RuntimeError('Kubernetes Agent pod changed repeatedly while preparing the Agent command') + + # The chart-generated Kubernetes settings produce startup log messages + # before `agent check --json`. Keep stdout machine-readable for the + # existing dd_agent_check replay path while preserving explicit overrides. + invocation_env = {'DD_LOG_LEVEL': 'off', **(env_vars or {})} + self._exec(['agent', *args], env_vars=invocation_env, pod_name=prepared_pod.name) def enter_shell(self) -> None: + self._ensure_prepared() + pod_name = self._pod_name() self._kubectl( [ 'exec', '-it', '--namespace', self._namespace, - f'pod/{_POD_NAME}', + f'pod/{pod_name}', '--container', - _CONTAINER_NAME, + CONTAINER_NAME, '--', 'bash', ], @@ -533,13 +444,19 @@ def enter_shell(self) -> None: ) def _show_logs(self) -> None: - self._kubectl( - ['logs', '--namespace', self._namespace, f'pod/{_POD_NAME}', '--container', _CONTAINER_NAME], - check=False, - ) + try: + pod = self._deployment.agent_pod(ready=False, required=False) + except Exception: + return + if pod is not None: + self._kubectl( + ['logs', '--namespace', self._namespace, f'pod/{pod.name}', '--container', CONTAINER_NAME], + check=False, + ) def show_logs(self) -> None: + pod_name = self._pod_name(ready=False) self._kubectl( - ['logs', '--namespace', self._namespace, f'pod/{_POD_NAME}', '--container', _CONTAINER_NAME], + ['logs', '--namespace', self._namespace, f'pod/{pod_name}', '--container', CONTAINER_NAME], check=True, ) diff --git a/ddev/src/ddev/e2e/agent/kubernetes_helm.py b/ddev/src/ddev/e2e/agent/kubernetes_helm.py new file mode 100644 index 0000000000000..f7471cf6c4a96 --- /dev/null +++ b/ddev/src/ddev/e2e/agent/kubernetes_helm.py @@ -0,0 +1,655 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +from __future__ import annotations + +import json +import os +import re +import time +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Literal, overload + +if TYPE_CHECKING: + import subprocess + + from ddev.utils.fs import Path + from ddev.utils.platform import Platform + +CHART_NAME = 'datadog' +CHART_REPOSITORY = 'https://helm.datadoghq.com' +CHART_VERSION = '3.231.6' +RELEASE_NAME = 'ddev-agent' +CONTAINER_NAME = 'agent' +OWNER_LABEL = 'ddev.datadoghq.com/environment' +_AGENT_COMPONENT_LABEL = 'app.kubernetes.io/component' +_AGENT_COMPONENT = 'agent' +_LEGACY_APP_LABEL = 'app' +_NAMESPACE_UID_METADATA = '_namespace_uid' + + +class AgentPodSelectionError(RuntimeError): + def __init__(self, message: str, *, candidate_count: int) -> None: + super().__init__(message) + self.candidate_count = candidate_count + + +@dataclass(frozen=True) +class PodIdentity: + name: str + uid: str + node_name: str + + +@dataclass(frozen=True) +class AgentImage: + repository: str + tag: str | None = None + digest: str | None = None + + +def parse_agent_image(image: str) -> AgentImage: + """Split a complete OCI image reference into Datadog Helm chart fields.""" + if not image or any(character.isspace() for character in image): + raise ValueError(f'Invalid Kubernetes Agent image reference: {image!r}') + + if '@' in image: + if image.count('@') != 1: + raise ValueError(f'Invalid Kubernetes Agent image reference: {image!r}') + repository, digest = image.rsplit('@', 1) + if not repository or not re.fullmatch(r'sha256:[A-Fa-f0-9]{64}', digest): + raise ValueError(f'Invalid Kubernetes Agent image digest: {image!r}') + return AgentImage(repository=repository, digest=digest) + + tag_separator = image.rfind(':') + if tag_separator <= image.rfind('/'): + raise ValueError(f'Kubernetes Agent image reference must include a tag or digest: {image!r}') + + repository = image[:tag_separator] + tag = image[tag_separator + 1 :] + if not repository or not re.fullmatch(r'[\w][\w.-]{0,127}', tag): + raise ValueError(f'Invalid Kubernetes Agent image tag: {image!r}') + return AgentImage(repository=repository, tag=tag) + + +class HelmDaemonSetDeployment: + """Own the Helm release used by the Kubernetes Agent backend.""" + + def __init__( + self, + *, + platform: Platform, + kubeconfig: str, + namespace: str, + owner_id: str, + kubernetes_metadata: dict[str, Any], + state_dir: Path, + wait_timeout: int, + ) -> None: + self.platform = platform + self.kubeconfig = kubeconfig + self.namespace = namespace + self.owner_id = owner_id + self.metadata = kubernetes_metadata + self.state_dir = state_dir + self.wait_timeout = wait_timeout + + @property + def namespace_labels(self) -> dict[str, str]: + return { + 'app.kubernetes.io/managed-by': 'ddev', + OWNER_LABEL: self.owner_id, + } + + @property + def pod_labels(self) -> dict[str, str]: + labels = self.metadata.get('pod_labels', {}) + if not isinstance(labels, dict) or not all( + isinstance(key, str) and isinstance(value, str) for key, value in labels.items() + ): + raise ValueError('Kubernetes Agent `pod_labels` must be a mapping of strings to strings') + return { + **labels, + OWNER_LABEL: self.owner_id, + _AGENT_COMPONENT_LABEL: _AGENT_COMPONENT, + _LEGACY_APP_LABEL: self.namespace, + } + + @property + def image_pull_policy(self) -> str: + pull_policy = self.metadata.get('image_pull_policy', 'Always') + if pull_policy not in {'Always', 'IfNotPresent', 'Never'}: + raise ValueError('Kubernetes Agent `image_pull_policy` must be Always, IfNotPresent, or Never') + return pull_policy + + @property + def helm_environment(self) -> dict[str, str]: + helm_dir = self.state_dir / 'helm' + directories = { + 'HELM_CACHE_HOME': helm_dir / 'cache', + 'HELM_CONFIG_HOME': helm_dir / 'config', + 'HELM_DATA_HOME': helm_dir / 'data', + } + environment = os.environ.copy() + for name, directory in directories.items(): + directory.ensure_dir_exists() + environment[name] = str(directory) + return environment + + @property + def kubectl_prefix(self) -> list[str]: + return ['kubectl', '--kubeconfig', self.kubeconfig] + + def _kubectl(self, args: list[str], **kwargs) -> subprocess.CompletedProcess: + return self.platform.run_command([*self.kubectl_prefix, *args], **kwargs) + + def _captured_kubectl(self, args: list[str], **kwargs) -> subprocess.CompletedProcess: + return self._kubectl( + args, + stdout=self.platform.modules.subprocess.PIPE, + stderr=self.platform.modules.subprocess.STDOUT, + **kwargs, + ) + + def _helm(self, args: list[str], **kwargs) -> subprocess.CompletedProcess: + kwargs.setdefault('env', self.helm_environment) + return self.platform.run_command(['helm', *args], **kwargs) + + def _captured_helm(self, args: list[str], **kwargs) -> subprocess.CompletedProcess: + return self._helm( + args, + stdout=self.platform.modules.subprocess.PIPE, + stderr=self.platform.modules.subprocess.STDOUT, + **kwargs, + ) + + @staticmethod + def process_output(process: subprocess.CompletedProcess) -> str: + output = process.stdout or b'' + if isinstance(output, bytes): + return output.decode('utf-8', errors='replace') + return output + + def check_helm(self) -> None: + try: + process = self._captured_helm(['version', '--short']) + except OSError as e: + raise RuntimeError('The Kubernetes Agent backend requires the `helm` executable') from e + if process.returncode: + raise RuntimeError(f'Unable to run Helm: {self.process_output(process)}') + + def values(self, agent_build: str, env_vars: dict[str, str], *, node_name: str) -> dict[str, Any]: + image = parse_agent_image(agent_build) + env_vars = env_vars.copy() + + def pop_bool(name: str, default: bool) -> bool: + value = env_vars.pop(name, str(default).lower()).lower() + if value not in {'true', 'false'}: + raise ValueError(f'Kubernetes Agent `{name}` must be true or false') + return value == 'true' + + def pop_port(name: str, default: int) -> int: + value = env_vars.pop(name, str(default)) + try: + port = int(value) + except ValueError as e: + raise ValueError(f'Kubernetes Agent `{name}` must be a valid port') from e + if not 1 <= port <= 65535: + raise ValueError(f'Kubernetes Agent `{name}` must be a valid port') + return port + + api_key = env_vars.pop('DD_API_KEY', 'a' * 32) + site = env_vars.pop('DD_SITE', None) + cluster_name = env_vars.pop('DD_CLUSTER_NAME', self.namespace) + log_level = env_vars.pop('DD_LOG_LEVEL', None) + dogstatsd_tag_cardinality = env_vars.pop('DD_DOGSTATSD_TAG_CARDINALITY', 'low') + tls_verify = pop_bool('DD_KUBELET_TLS_VERIFY', False) + apm_enabled = pop_bool('DD_APM_ENABLED', False) + logs_enabled = pop_bool('DD_LOGS_ENABLED', False) + logs_collect_all = pop_bool('DD_LOGS_CONFIG_CONTAINER_COLLECT_ALL', False) + logs_auto_multi_line = pop_bool('DD_LOGS_CONFIG_AUTO_MULTI_LINE_DETECTION', False) + if apm_enabled: + raise ValueError('Kubernetes Agent `DD_APM_ENABLED` must remain false for this E2E deployment') + if logs_enabled or logs_collect_all or logs_auto_multi_line: + raise ValueError('Kubernetes Agent log collection must remain disabled for this E2E deployment') + remote_configuration_enabled = pop_bool('DD_REMOTE_CONFIGURATION_ENABLED', False) + container_lifecycle_enabled = pop_bool('DD_CONTAINER_LIFECYCLE_ENABLED', False) + dogstatsd_non_local_traffic = pop_bool('DD_DOGSTATSD_NON_LOCAL_TRAFFIC', True) + dogstatsd_origin_detection = pop_bool('DD_DOGSTATSD_ORIGIN_DETECTION', False) + dogstatsd_port = pop_port('DD_DOGSTATSD_PORT', 8125) + expvar_port = pop_port('DD_EXPVAR_PORT', 5000) + + # These variables are generated by the pinned chart. Adding them again + # through agents.containers.agent.env produces an invalid Kubernetes env + # list (Helm 4's server-side apply rejects the duplicate merge keys). + chart_generated_variables = { + 'DD_APM_NON_LOCAL_TRAFFIC', + 'DD_APM_RECEIVER_PORT', + 'DD_APM_RECEIVER_SOCKET', + 'DD_AUTH_TOKEN_FILE_PATH', + 'DD_COMPLIANCE_CONFIG_ENABLED', + 'DD_COMPLIANCE_CONFIG_RUN_IN_SYSTEM_PROBE', + 'DD_CONTAINER_IMAGE_ENABLED', + 'DD_CSI_ENABLED', + 'DD_DOGSTATSD_ORIGIN_DETECTION_CLIENT', + 'DD_DOGSTATSD_SOCKET', + 'DD_HEALTH_PORT', + 'DD_INSTRUMENTATION_INSTALL_ID', + 'DD_INSTRUMENTATION_INSTALL_TIME', + 'DD_INSTRUMENTATION_INSTALL_TYPE', + 'DD_KUBELET_CORE_CHECK_ENABLED', + 'DD_KUBELET_USE_API_SERVER', + 'DD_KUBERNETES_KUBELET_HOST', + 'DD_KUBERNETES_KUBELET_NODENAME', + 'DD_KUBERNETES_KUBELET_PODRESOURCES_SOCKET', + 'DD_KUBERNETES_KUBE_SERVICE_IGNORE_READINESS', + 'DD_KUBERNETES_USE_ENDPOINT_SLICES', + 'DD_LANGUAGE_DETECTION_ENABLED', + 'DD_LANGUAGE_DETECTION_REPORTING_ENABLED', + 'DD_LOGS_CONFIG_K8S_CONTAINER_USE_FILE', + 'DD_ORCHESTRATOR_EXPLORER_ENABLED', + 'DD_ORIGIN_DETECTION_UNIFIED', + 'DD_OTLP_CONFIG_LOGS_ENABLED', + 'DD_PROCESS_AGENT_DISCOVERY_ENABLED', + 'DD_PROCESS_CONFIG_CONTAINER_COLLECTION_ENABLED', + 'DD_PROCESS_CONFIG_PROCESS_COLLECTION_ENABLED', + 'DD_PROCESS_CONFIG_RUN_IN_CORE_AGENT_ENABLED', + 'DD_STRIP_PROCESS_ARGS', + 'KUBERNETES', + } + if duplicates := sorted(chart_generated_variables.intersection(env_vars)): + raise ValueError( + 'Kubernetes Agent environment variables are managed by the Helm chart and cannot be overridden: ' + + ', '.join(duplicates) + ) + + env_vars.setdefault('DD_AUTOCONFIG_FROM_ENVIRONMENT', 'true') + + datadog_values: dict[str, Any] = { + 'apiKey': api_key, + 'clusterName': cluster_name, + 'collectEvents': False, + 'leaderElection': False, + 'useHostPID': False, + 'clusterChecks': {'enabled': False}, + 'kubeStateMetricsCore': {'enabled': False}, + 'kubelet': {'tlsVerify': tls_verify}, + 'logs': { + 'enabled': logs_enabled, + 'containerCollectAll': logs_collect_all, + 'autoMultiLineDetection': logs_auto_multi_line, + }, + 'apm': {'socketEnabled': apm_enabled, 'portEnabled': False}, + 'dogstatsd': { + 'port': dogstatsd_port, + 'useSocketVolume': False, + 'nonLocalTraffic': dogstatsd_non_local_traffic, + 'originDetection': dogstatsd_origin_detection, + 'tagCardinality': dogstatsd_tag_cardinality, + }, + 'expvarPort': expvar_port, + 'containerLifecycle': {'enabled': container_lifecycle_enabled}, + # This is Datadog's service discovery product, not Kubernetes + # integration Autodiscovery, and otherwise starts system-probe. + 'discovery': {'enabled': False, 'networkStats': {'enabled': False}}, + 'processAgent': { + 'enabled': False, + 'processCollection': False, + 'processDiscovery': False, + 'containerCollection': False, + }, + 'orchestratorExplorer': { + 'enabled': False, + 'kubelet_configuration_check': {'enabled': False}, + }, + 'operator': {'enabled': False}, + 'remoteConfiguration': {'enabled': remote_configuration_enabled}, + } + if site is not None: + datadog_values['site'] = site + if log_level is not None: + datadog_values['logLevel'] = log_level + + image_values: dict[str, Any] = { + 'repository': image.repository, + 'doNotCheckTag': True, + 'pullPolicy': self.image_pull_policy, + } + if image.tag is not None: + image_values['tag'] = image.tag + if image.digest is not None: + image_values['digest'] = image.digest + + return { + 'fullnameOverride': self.namespace, + 'targetSystem': 'linux', + 'commonLabels': {OWNER_LABEL: self.owner_id}, + 'datadog': datadog_values, + 'clusterAgent': { + 'enabled': False, + 'admissionController': {'enabled': False}, + }, + 'agents': { + 'enabled': True, + 'image': image_values, + 'instanceLabelOverride': self.owner_id, + 'podLabels': self.pod_labels, + 'tolerations': [{'operator': 'Exists'}], + # DaemonSets automatically tolerate an unschedulable node. Pin + # this initial single-Agent implementation to the one node that + # topology validation selected, even if cordoned nodes exist. + 'affinity': { + 'nodeAffinity': { + 'requiredDuringSchedulingIgnoredDuringExecution': { + 'nodeSelectorTerms': [ + { + 'matchFields': [ + { + 'key': 'metadata.name', + 'operator': 'In', + 'values': [node_name], + } + ] + } + ] + } + } + }, + 'containers': { + 'agent': { + # The chart normally bypasses the image's s6 entrypoint with `agent run`. + # Use the normal image entrypoint so ddev can restart the Agent process + # without replacing the container and losing editable installations. + 'command': ['/bin/entrypoint.sh'], + # Editable integration packages are copied to /home and installed + # into the embedded Python environment at runtime. + 'securityContext': {'readOnlyRootFilesystem': False}, + # Intake connectivity is intentionally not part of local E2E + # readiness. KubernetesAgent performs `agent status` checks. + 'livenessProbe': self._local_probe(), + 'readinessProbe': self._local_probe(), + 'startupProbe': self._local_probe(), + 'env': [{'name': key, 'value': value} for key, value in sorted(env_vars.items())], + } + }, + }, + } + + @staticmethod + def _local_probe() -> dict[str, Any]: + # Override the chart's intake-dependent probes without retaining its + # default 15-second delays after Helm's deep merge. + return { + 'exec': {'command': ['/bin/true']}, + 'initialDelaySeconds': 0, + 'periodSeconds': 1, + 'timeoutSeconds': 1, + 'successThreshold': 1, + 'failureThreshold': 3, + } + + def validate_namespace_absent(self) -> None: + process = self._captured_kubectl(['get', 'namespace', self.namespace, '--ignore-not-found=true', '-o', 'name']) + if process.returncode: + raise RuntimeError( + f'Unable to inspect Kubernetes Agent namespace `{self.namespace}`: {self.process_output(process)}' + ) + if self.process_output(process).strip(): + raise RuntimeError( + f'Refusing to overwrite Kubernetes resources not owned by this environment: namespace/{self.namespace}' + ) + + process = self._captured_kubectl( + [ + 'get', + f'clusterrole/{self.namespace}', + f'clusterrolebinding/{self.namespace}', + '--ignore-not-found=true', + '-o', + 'name', + ] + ) + if process.returncode: + raise RuntimeError( + f'Unable to inspect Kubernetes Agent cluster-scoped resources: {self.process_output(process)}' + ) + if self.process_output(process).strip(): + raise RuntimeError( + 'Refusing to overwrite stale Kubernetes Agent cluster-scoped resources: ' + + self.process_output(process).strip() + ) + + def create_namespace(self) -> None: + payload = { + 'apiVersion': 'v1', + 'kind': 'Namespace', + 'metadata': {'name': self.namespace, 'labels': self.namespace_labels}, + } + process = self._captured_kubectl(['create', '-f', '-', '-o', 'json'], input=json.dumps(payload).encode()) + if process.returncode: + raise RuntimeError(f'Unable to create Kubernetes Agent namespace: {self.process_output(process)}') + try: + namespace_uid = json.loads(self.process_output(process))['metadata']['uid'] + except (KeyError, TypeError, json.JSONDecodeError) as e: + raise RuntimeError(f'Unable to parse the created Kubernetes Agent namespace: {e}') from e + if not isinstance(namespace_uid, str) or not namespace_uid: + raise RuntimeError('Created Kubernetes Agent namespace is missing its UID') + self.metadata[_NAMESPACE_UID_METADATA] = namespace_uid + + def install(self, values: dict[str, Any]) -> None: + process = self._captured_helm( + [ + 'install', + RELEASE_NAME, + CHART_NAME, + '--repo', + CHART_REPOSITORY, + '--version', + CHART_VERSION, + '--namespace', + self.namespace, + '--kubeconfig', + self.kubeconfig, + '--atomic', + '--wait', + '--timeout', + f'{self.wait_timeout}s', + '-f', + '-', + ], + input=json.dumps(values).encode(), + ) + if process.returncode: + raise RuntimeError(f'Unable to install Kubernetes Agent Helm release: {self.process_output(process)}') + + def wait_for_daemonset(self) -> None: + process = self._captured_kubectl( + [ + 'rollout', + 'status', + f'daemonset/{self.namespace}', + '--namespace', + self.namespace, + f'--timeout={self.wait_timeout}s', + ] + ) + if process.returncode: + raise RuntimeError(f'Kubernetes Agent DaemonSet did not become ready: {self.process_output(process)}') + + @overload + def agent_pod(self, *, ready: bool = True, required: Literal[True] = True) -> PodIdentity: ... + + @overload + def agent_pod(self, *, ready: bool = True, required: Literal[False]) -> PodIdentity | None: ... + + def agent_pod(self, *, ready: bool = True, required: bool = True) -> PodIdentity | None: + selector = f'{_AGENT_COMPONENT_LABEL}={_AGENT_COMPONENT},{OWNER_LABEL}={self.owner_id}' + process = self._captured_kubectl( + ['get', 'pods', '--namespace', self.namespace, '--selector', selector, '-o', 'json'] + ) + if process.returncode: + raise RuntimeError(f'Unable to inspect Kubernetes Agent pods: {self.process_output(process)}') + + try: + items = json.loads(self.process_output(process))['items'] + except (KeyError, TypeError, json.JSONDecodeError) as e: + raise RuntimeError(f'Unable to parse Kubernetes Agent pod data: {e}') from e + + candidates: list[PodIdentity] = [] + descriptions: list[str] = [] + for item in items: + metadata = item.get('metadata', {}) + spec = item.get('spec', {}) + status = item.get('status', {}) + name = metadata.get('name', '') + node_name = spec.get('nodeName', '') + phase = status.get('phase', '') + is_ready = any( + condition.get('type') == 'Ready' and condition.get('status') == 'True' + for condition in status.get('conditions', []) + ) + terminating = bool(metadata.get('deletionTimestamp')) + descriptions.append( + f'{name} (node={node_name or ""}, phase={phase}, ready={is_ready}, terminating={terminating})' + ) + container_names = {container.get('name') for container in spec.get('containers', [])} + if terminating or CONTAINER_NAME not in container_names: + continue + if ready and (phase != 'Running' or not is_ready): + continue + candidates.append(PodIdentity(name=name, uid=metadata.get('uid', ''), node_name=node_name)) + + if not candidates and not required: + return None + if len(candidates) != 1: + details = ', '.join(descriptions) if descriptions else 'none' + raise AgentPodSelectionError( + f'Expected exactly one {"ready " if ready else ""}Kubernetes Agent pod, ' + f'found {len(candidates)}; observed: {details}', + candidate_count=len(candidates), + ) + return candidates[0] + + def wait_for_agent_pod(self) -> PodIdentity: + deadline = time.monotonic() + self.wait_timeout + last_error = None + while time.monotonic() < deadline: + try: + return self.agent_pod() + except AgentPodSelectionError as e: + if e.candidate_count: + raise + last_error = e + time.sleep(1) + raise RuntimeError('Kubernetes Agent pod did not become selectable before the timeout') from last_error + + def namespace_identity(self) -> tuple[str | None, str | None] | None: + process = self._captured_kubectl(['get', 'namespace', self.namespace, '--ignore-not-found=true', '-o', 'json']) + if process.returncode: + raise RuntimeError( + f'Unable to inspect Kubernetes Agent namespace `{self.namespace}`: {self.process_output(process)}' + ) + output = self.process_output(process).strip() + if not output: + return None + try: + namespace = json.loads(output) + metadata = namespace['metadata'] + return metadata.get('labels', {}).get(OWNER_LABEL), metadata.get('uid') + except (KeyError, TypeError, json.JSONDecodeError) as e: + raise RuntimeError(f'Unable to parse Kubernetes Agent namespace `{self.namespace}`: {e}') from e + + def namespace_is_owned(self) -> bool: + identity = self.namespace_identity() + if identity is None: + return False + owner_id, namespace_uid = identity + expected_uid = self.metadata.get(_NAMESPACE_UID_METADATA) + return owner_id == self.owner_id and (expected_uid is None or namespace_uid == expected_uid) + + def require_namespace_owned(self) -> None: + if not self.namespace_is_owned(): + raise RuntimeError( + f'Refusing Kubernetes Agent teardown because namespace `{self.namespace}` ownership changed' + ) + + def owned_cluster_resources_exist(self) -> bool: + process = self._captured_kubectl( + [ + 'get', + 'clusterrole,clusterrolebinding', + '--selector', + f'{OWNER_LABEL}={self.owner_id}', + '-o', + 'name', + ] + ) + if process.returncode: + raise RuntimeError( + f'Unable to inspect Kubernetes Agent cluster-scoped resources: {self.process_output(process)}' + ) + return bool(self.process_output(process).strip()) + + def owned_namespaced_resources_exist(self) -> bool: + process = self._captured_kubectl( + [ + 'get', + 'all,secret,configmap,serviceaccount,role,rolebinding', + '--namespace', + self.namespace, + '--selector', + f'{OWNER_LABEL}={self.owner_id}', + '-o', + 'name', + ] + ) + if process.returncode: + raise RuntimeError( + f'Unable to inspect Kubernetes Agent namespaced resources: {self.process_output(process)}' + ) + return bool(self.process_output(process).strip()) + + def uninstall(self) -> None: + self.require_namespace_owned() + # Do not use `helm list` as a preflight: Helm 3 omits some statuses by + # default, while neither Helm 3 nor Helm 4 can explicitly select the + # valid `unknown` status. An unconditional uninstall is status-agnostic. + process = self._captured_helm( + [ + 'uninstall', + RELEASE_NAME, + '--namespace', + self.namespace, + '--kubeconfig', + self.kubeconfig, + '--cascade', + 'foreground', + '--wait', + '--timeout', + f'{self.wait_timeout}s', + ] + ) + if process.returncode: + output = self.process_output(process) + if re.search(r'\brelease\b.*\bnot found\b', output, flags=re.IGNORECASE | re.DOTALL): + if self.owned_cluster_resources_exist() or self.owned_namespaced_resources_exist(): + raise RuntimeError( + 'Kubernetes Agent Helm release metadata is missing while owner-labeled chart resources remain' + ) + return + raise RuntimeError(f'Unable to uninstall Kubernetes Agent Helm release: {output}') + + def delete_namespace(self) -> None: + self.require_namespace_owned() + process = self._captured_kubectl( + [ + 'delete', + 'namespace', + self.namespace, + '--ignore-not-found=true', + '--wait=true', + f'--timeout={self.wait_timeout}s', + ] + ) + if process.returncode: + raise RuntimeError(f'Unable to remove Kubernetes Agent namespace: {self.process_output(process)}') diff --git a/ddev/tests/e2e/agent/test_kubernetes.py b/ddev/tests/e2e/agent/test_kubernetes.py index ba57ae9b9684f..8580cddb7b02c 100644 --- a/ddev/tests/e2e/agent/test_kubernetes.py +++ b/ddev/tests/e2e/agent/test_kubernetes.py @@ -7,9 +7,11 @@ import pytest from ddev.e2e.agent.kubernetes import KubernetesAgent +from ddev.e2e.agent.kubernetes_helm import CHART_REPOSITORY, CHART_VERSION from ddev.integration.core import Integration from ddev.repo.config import RepositoryConfig +POD_NAME = 'ddev-agent-abcde' RESTART_COMMAND = ( 'old_pid=$(pidof agent) || exit 1; ' 'set -- $old_pid; old_pid=$1; ' @@ -23,6 +25,20 @@ ) +def pod_data(*, ready=True, terminating=False, name=POD_NAME, node='kind-control-plane'): + metadata = {'name': name, 'uid': f'{name}-uid'} + if terminating: + metadata['deletionTimestamp'] = '2026-07-22T00:00:00Z' + return { + 'metadata': metadata, + 'spec': {'nodeName': node, 'containers': [{'name': 'agent'}]}, + 'status': { + 'phase': 'Running', + 'conditions': [{'type': 'Ready', 'status': 'True' if ready else 'False'}], + }, + } + + @pytest.fixture(scope='module') def get_integration(local_repo): def _get_integration(name): @@ -53,6 +69,7 @@ def metadata(auto_conf): 'kubernetes': { 'kubeconfig': '/tmp/kubeconfig', 'auto_conf': str(auto_conf), + 'wait_timeout': 120, }, } @@ -72,6 +89,14 @@ def run(command, **kwargs): if command[-4:] == ['get', 'nodes', '-o', 'json']: nodes = {'items': [{'metadata': {'name': 'kind-control-plane'}, 'spec': {}}]} return successful_process(command, stdout=json.dumps(nodes).encode()) + if 'get' in command and 'pods' in command and command[-2:] == ['-o', 'json']: + return successful_process(command, stdout=json.dumps({'items': [pod_data()]}).encode()) + if command[:3] == ['helm', 'version', '--short']: + return successful_process(command, stdout=b'v3.19.0\n') + if 'create' in command and command[-4:] == ['-f', '-', '-o', 'json']: + return successful_process(command, stdout=b'{"metadata":{"uid":"namespace-uid"}}') + if command[:2] == ['helm', 'list']: + return successful_process(command, stdout=b'[]') return successful_process(command) return mocker.patch.object(app.platform, 'run_command', side_effect=run) @@ -81,7 +106,13 @@ def command_calls(run_command): return [call.args[0] for call in run_command.call_args_list] -def test_start_uses_selected_image_rbac_config_and_local_packages( +def operational_calls(run_command): + return [ + call.args[0] for call in run_command.call_args_list if not ('get' in call.args[0] and 'pods' in call.args[0]) + ] + + +def test_start_installs_pinned_helm_chart_and_prepares_selected_agent( agent, metadata, config_file, auto_conf, temp_dir, run_command ): local_base = temp_dir / 'datadog_checks_base' @@ -97,52 +128,93 @@ def test_start_uses_selected_image_rbac_config_and_local_packages( calls = command_calls(run_command) prefix = ['kubectl', '--kubeconfig', '/tmp/kubeconfig'] - assert calls[0] == [*prefix, 'get', 'nodes', '-o', 'json'] + assert calls[0] == ['helm', 'version', '--short'] + assert calls[1] == [*prefix, 'get', 'nodes', '-o', 'json'] + assert calls[2] == [*prefix, 'get', 'namespace', agent._namespace, '--ignore-not-found=true', '-o', 'name'] - create_calls = [call for call in run_command.call_args_list if call.args[0] == [*prefix, 'create', '-f', '-']] + create_calls = [ + call for call in run_command.call_args_list if call.args[0] == [*prefix, 'create', '-f', '-', '-o', 'json'] + ] + assert len(create_calls) == 1 namespace = json.loads(create_calls[0].kwargs['input']) - manifest = json.loads(create_calls[1].kwargs['input']) - resources = {item['kind']: item for item in [namespace, *manifest['items']]} - assert namespace['kind'] == 'Namespace' - assert resources['Pod']['spec']['containers'][0]['image'] == 'registry.example.com/datadog-agent:test' - assert resources['Pod']['spec']['containers'][0]['imagePullPolicy'] == 'Always' - assert resources['Pod']['spec']['serviceAccountName'] == 'ddev-agent' - for kind in ('Namespace', 'ServiceAccount', 'ClusterRole', 'ClusterRoleBinding', 'Pod'): - assert resources[kind]['metadata']['labels']['ddev.datadoghq.com/environment'] == 'test-owner' - assert resources['ClusterRole']['rules'] == [ - {'apiGroups': [''], 'resources': ['nodes'], 'verbs': ['get', 'list', 'watch']}, - { - 'apiGroups': [''], - 'resources': ['nodes/metrics', 'nodes/spec', 'nodes/stats', 'nodes/proxy'], - 'verbs': ['get'], - }, - { - 'apiGroups': [''], - 'resources': ['pods', 'endpoints', 'services'], - 'verbs': ['get', 'list', 'watch'], + assert namespace == { + 'apiVersion': 'v1', + 'kind': 'Namespace', + 'metadata': { + 'name': agent._namespace, + 'labels': { + 'app.kubernetes.io/managed-by': 'ddev', + 'ddev.datadoghq.com/environment': 'test-owner', + }, }, + } + + helm_call = next(call for call in run_command.call_args_list if call.args[0][:2] == ['helm', 'install']) + assert helm_call.args[0] == [ + 'helm', + 'install', + 'ddev-agent', + 'datadog', + '--repo', + CHART_REPOSITORY, + '--version', + CHART_VERSION, + '--namespace', + agent._namespace, + '--kubeconfig', + '/tmp/kubeconfig', + '--atomic', + '--wait', + '--timeout', + '120s', + '-f', + '-', ] - env = {item['name']: item.get('value') for item in resources['Pod']['spec']['containers'][0]['env']} - assert env['DD_API_KEY'] == 'a' * 32 - assert env['DD_SITE'] == 'datadoghq.com' - assert env['DD_AUTOCONFIG_FROM_ENVIRONMENT'] == 'true' - assert 'DD_KUBERNETES_KUBELET_HOST' in env - assert 'DD_KUBERNETES_KUBELET_NODENAME' in env + values = json.loads(helm_call.kwargs['input']) + assert values['fullnameOverride'] == agent._namespace + assert values['commonLabels'] == {'ddev.datadoghq.com/environment': 'test-owner'} + assert values['agents']['instanceLabelOverride'] == 'test-owner' + assert values['agents']['image'] == { + 'repository': 'registry.example.com/datadog-agent', + 'tag': 'test', + 'doNotCheckTag': True, + 'pullPolicy': 'Always', + } + assert values['agents']['podLabels']['ddev.datadoghq.com/environment'] == 'test-owner' + assert values['agents']['podLabels']['app.kubernetes.io/component'] == 'agent' + assert values['agents']['podLabels']['app'] == agent._namespace + assert values['agents']['affinity']['nodeAffinity']['requiredDuringSchedulingIgnoredDuringExecution'][ + 'nodeSelectorTerms' + ] == [{'matchFields': [{'key': 'metadata.name', 'operator': 'In', 'values': ['kind-control-plane']}]}] + assert values['datadog']['site'] == 'datadoghq.com' + assert values['datadog']['dogstatsd']['useSocketVolume'] is False + assert values['datadog']['operator']['enabled'] is False + assert values['clusterAgent']['enabled'] is False + assert helm_call.kwargs['env']['HELM_CACHE_HOME'].endswith('/helm/cache') + assert [ + *prefix, + 'rollout', + 'status', + f'daemonset/{agent._namespace}', + '--namespace', + agent._namespace, + '--timeout=120s', + ] in calls assert [ *prefix, 'cp', '--container', 'agent', local_base.name, - f'{agent._namespace}/ddev-agent:/home/datadog_checks_base', + f'{agent._namespace}/{POD_NAME}:/home/datadog_checks_base', ] in calls - assert [ + pip_command = [ *prefix, 'exec', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', @@ -153,14 +225,18 @@ def test_start_uses_selected_image_rbac_config_and_local_packages( '--disable-pip-version-check', '-e', '/home/datadog_checks_base[kube]', - ] in calls + ] + assert pip_command in calls + pip_call = next(call for call in run_command.call_args_list if call.args[0] == pip_command) + assert pip_call.kwargs['stdout'] == subprocess.PIPE + assert pip_call.kwargs['stderr'] == subprocess.STDOUT assert [ *prefix, 'cp', '--container', 'agent', config_file.name, - f'{agent._namespace}/ddev-agent:/etc/datadog-agent/conf.d/velero.d/conf.yaml', + f'{agent._namespace}/{POD_NAME}:/etc/datadog-agent/conf.d/velero.d/conf.yaml', ] in calls assert [ *prefix, @@ -168,14 +244,14 @@ def test_start_uses_selected_image_rbac_config_and_local_packages( '--container', 'agent', auto_conf.name, - f'{agent._namespace}/ddev-agent:/etc/datadog-agent/conf.d/velero.d/auto_conf.yaml', + f'{agent._namespace}/{POD_NAME}:/etc/datadog-agent/conf.d/velero.d/auto_conf.yaml', ] in calls assert [ *prefix, 'exec', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', @@ -183,10 +259,24 @@ def test_start_uses_selected_image_rbac_config_and_local_packages( '-c', RESTART_COMMAND, ] in calls + assert [ + *prefix, + 'exec', + '--namespace', + agent._namespace, + f'pod/{POD_NAME}', + '--container', + 'agent', + '--', + 'touch', + '/home/.ddev-agent-prepared', + ] in calls + assert metadata['kubernetes']['_namespace_uid'] == 'namespace-uid' assert metadata['kubernetes']['local_packages'] == [ {'path': str(local_base), 'name': 'datadog_checks_base', 'features': '[kube]'}, {'path': str(integration), 'name': 'velero', 'features': '[deps]'}, ] + assert not any('clusterrole' in command or 'clusterrolebinding' in command for command in calls) local_base_copy = next(call for call in run_command.call_args_list if call.args[0][-2] == local_base.name) assert local_base_copy.kwargs['cwd'] == local_base.resolve().parent @@ -203,22 +293,29 @@ def test_rejects_invalid_wait_timeout_before_creating_resources(agent, metadata, def test_rejects_multi_node_clusters(agent, app, mocker): nodes = {'items': [{'metadata': {'name': 'one'}, 'spec': {}}, {'metadata': {'name': 'two'}, 'spec': {}}]} - run_command = mocker.patch.object( - app.platform, - 'run_command', - return_value=successful_process([], stdout=json.dumps(nodes).encode()), - ) + + def run(command, **kwargs): + if command[:3] == ['helm', 'version', '--short']: + return successful_process(command) + return successful_process(command, stdout=json.dumps(nodes).encode()) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) with pytest.raises(NotImplementedError, match='exactly one schedulable node'): agent.start(agent_build='', local_packages={}, env_vars={}) - run_command.assert_called_once() + assert command_calls(run_command) == [ + ['helm', 'version', '--short'], + ['kubectl', '--kubeconfig', '/tmp/kubeconfig', 'get', 'nodes', '-o', 'json'], + ] -def test_rejects_preexisting_owned_resources(agent, app, mocker): +def test_rejects_preexisting_namespace(agent, app, mocker): nodes = {'items': [{'metadata': {'name': 'kind-control-plane'}, 'spec': {}}]} def run(command, **kwargs): + if command[:3] == ['helm', 'version', '--short']: + return successful_process(command) if command[-4:] == ['get', 'nodes', '-o', 'json']: return successful_process(command, stdout=json.dumps(nodes).encode()) if 'namespace' in command and '--ignore-not-found=true' in command: @@ -230,43 +327,79 @@ def run(command, **kwargs): with pytest.raises(RuntimeError, match='Refusing to overwrite Kubernetes resources'): agent.start(agent_build='', local_packages={}, env_vars={}) - assert not any(call.args[0][-3:] == ['create', '-f', '-'] for call in run_command.call_args_list) + assert not any('create' in call.args[0] for call in run_command.call_args_list) + + +def test_rejects_stale_cluster_resources_before_creating_namespace(agent, app, mocker): + nodes = {'items': [{'metadata': {'name': 'kind-control-plane'}, 'spec': {}}]} + + def run(command, **kwargs): + if command[:3] == ['helm', 'version', '--short']: + return successful_process(command) + if command[-4:] == ['get', 'nodes', '-o', 'json']: + return successful_process(command, stdout=json.dumps(nodes).encode()) + if f'clusterrole/{agent._namespace}' in command: + return successful_process( + command, stdout=f'clusterrole.rbac.authorization.k8s.io/{agent._namespace}\n'.encode() + ) + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + with pytest.raises(RuntimeError, match='stale Kubernetes Agent cluster-scoped resources'): + agent.start(agent_build='', local_packages={}, env_vars={}) + + assert not any('create' in call.args[0] for call in run_command.call_args_list) -def test_creation_fails_if_a_resource_appears_after_preflight(agent, app, mocker): +def test_namespace_creation_remains_atomic(agent, app, mocker): nodes = {'items': [{'metadata': {'name': 'kind-control-plane'}, 'spec': {}}]} def run(command, **kwargs): + if command[:3] == ['helm', 'version', '--short']: + return successful_process(command) if command[-4:] == ['get', 'nodes', '-o', 'json']: return successful_process(command, stdout=json.dumps(nodes).encode()) - if command[-3:] == ['create', '-f', '-']: - return subprocess.CompletedProcess(command, 1, stdout=b'', stderr=b'namespace already exists') + if command[-4:] == ['-f', '-', '-o', 'json']: + return subprocess.CompletedProcess(command, 1, stdout=b'namespace already exists') return successful_process(command) run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) - with pytest.raises(RuntimeError, match='Unable to create Kubernetes Agent resources'): + with pytest.raises(RuntimeError, match='Unable to create Kubernetes Agent namespace'): agent.start(agent_build='', local_packages={}, env_vars={}) calls = command_calls(run_command) - assert sum(command[-3:] == ['create', '-f', '-'] for command in calls) == 1 - assert not any('apply' in command or 'wait' in command for command in calls) + assert sum(command[-4:] == ['-f', '-', '-o', 'json'] for command in calls) == 1 + assert not any(command[:2] == ['helm', 'install'] for command in calls) -def test_invoke_synchronizes_config_and_environment(agent, metadata, config_file, auto_conf, run_command): +def test_invoke_synchronizes_config_and_environment(agent, config_file, auto_conf, run_command): config_file.write_text('instances:\n - openmetrics_endpoint: http://velero:8085/metrics\n') agent.invoke(['check', 'velero', '--json'], env_vars={'ZED': 'last', 'ALPHA': 'first'}) - calls = command_calls(run_command) prefix = ['kubectl', '--kubeconfig', '/tmp/kubeconfig'] - assert calls == [ + assert operational_calls(run_command) == [ + [ + *prefix, + 'exec', + '--namespace', + agent._namespace, + f'pod/{POD_NAME}', + '--container', + 'agent', + '--', + 'test', + '-f', + '/home/.ddev-agent-prepared', + ], [ *prefix, 'exec', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', @@ -280,14 +413,14 @@ def test_invoke_synchronizes_config_and_environment(agent, metadata, config_file '--container', 'agent', config_file.name, - f'{agent._namespace}/ddev-agent:/etc/datadog-agent/conf.d/velero.d/conf.yaml', + f'{agent._namespace}/{POD_NAME}:/etc/datadog-agent/conf.d/velero.d/conf.yaml', ], [ *prefix, 'exec', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', @@ -301,19 +434,33 @@ def test_invoke_synchronizes_config_and_environment(agent, metadata, config_file '--container', 'agent', auto_conf.name, - f'{agent._namespace}/ddev-agent:/etc/datadog-agent/conf.d/velero.d/auto_conf.yaml', + f'{agent._namespace}/{POD_NAME}:/etc/datadog-agent/conf.d/velero.d/auto_conf.yaml', + ], + [ + *prefix, + 'exec', + '--namespace', + agent._namespace, + f'pod/{POD_NAME}', + '--container', + 'agent', + '--', + 'test', + '-f', + '/home/.ddev-agent-prepared', ], [ *prefix, 'exec', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', 'env', 'ALPHA=first', + 'DD_LOG_LEVEL=off', 'ZED=last', 'agent', 'check', @@ -323,12 +470,105 @@ def test_invoke_synchronizes_config_and_environment(agent, metadata, config_file ] +def test_invoke_reinstalls_after_daemonset_replaces_container(agent, metadata, temp_dir, app, mocker): + local_package = temp_dir / 'velero-source' + local_package.ensure_dir_exists() + metadata['kubernetes']['local_packages'] = [ + {'path': str(local_package), 'name': 'velero-source', 'features': '[deps]'} + ] + metadata['start_commands'] = ['echo recovery-start'] + prepared = False + + def run(command, **kwargs): + nonlocal prepared + if 'get' in command and 'pods' in command: + return successful_process( + command, stdout=json.dumps({'items': [pod_data(name='replacement-agent')]}).encode() + ) + if command[-3:] == ['test', '-f', '/home/.ddev-agent-prepared']: + return successful_process(command) if prepared else subprocess.CompletedProcess(command, 1) + if command[-2:] == ['touch', '/home/.ddev-agent-prepared']: + prepared = True + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + agent.invoke(['status']) + + calls = command_calls(run_command) + assert any('pip' in command and '/home/velero-source[deps]' in command for command in calls) + marker_probes = [ + call for call in run_command.call_args_list if call.args[0][-3:] == ['test', '-f', '/home/.ddev-agent-prepared'] + ] + assert marker_probes + assert all(call.kwargs['stdout'] == subprocess.PIPE for call in marker_probes) + recovery_start = next( + call for call in run_command.call_args_list if call.args[0][-2:] == ['echo', 'recovery-start'] + ) + assert recovery_start.kwargs['stdout'] == subprocess.PIPE + assert recovery_start.kwargs['stderr'] == subprocess.STDOUT + assert any(command[-2:] == ['touch', '/home/.ddev-agent-prepared'] for command in calls) + invoke = next(command for command in reversed(calls) if command[-2:] == ['agent', 'status']) + assert 'pod/replacement-agent' in invoke + + +def test_invoke_retries_preparation_when_pod_changes_during_synchronization(agent, app, mocker): + state = {'pod_queries': 0, 'replacement_prepared': False} + + def run(command, **kwargs): + if 'get' in command and 'pods' in command: + state['pod_queries'] += 1 + name = 'old-agent' if state['pod_queries'] == 1 else 'replacement-agent' + return successful_process(command, stdout=json.dumps({'items': [pod_data(name=name)]}).encode()) + if command[-3:] == ['test', '-f', '/home/.ddev-agent-prepared']: + if 'pod/old-agent' in command or state['replacement_prepared']: + return successful_process(command) + return subprocess.CompletedProcess(command, 1) + if command[-2:] == ['touch', '/home/.ddev-agent-prepared']: + state['replacement_prepared'] = True + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + agent.invoke(['status']) + + calls = command_calls(run_command) + assert state['pod_queries'] == 5 + assert any( + 'pod/replacement-agent' in command and command[-2:] == ['touch', '/home/.ddev-agent-prepared'] + for command in calls + ) + invoke = next(command for command in reversed(calls) if command[-2:] == ['agent', 'status']) + assert 'pod/replacement-agent' in invoke + + +def test_invoke_waits_for_replacement_when_no_agent_pod_is_ready(agent, app, mocker): + mocker.patch('ddev.e2e.agent.kubernetes_helm.time.sleep') + pod_queries = 0 + + def run(command, **kwargs): + nonlocal pod_queries + if 'get' in command and 'pods' in command: + pod_queries += 1 + items = [] if pod_queries == 1 else [pod_data(name='replacement-agent')] + return successful_process(command, stdout=json.dumps({'items': items}).encode()) + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + agent.invoke(['status']) + + calls = command_calls(run_command) + assert pod_queries == 3 + invoke = next(command for command in reversed(calls) if command[-2:] == ['agent', 'status']) + assert 'pod/replacement-agent' in invoke + + def test_invoke_removes_pod_config_when_host_config_is_absent(agent, config_file, run_command): config_file.remove() agent.invoke(['status']) - calls = command_calls(run_command) assert [ 'kubectl', '--kubeconfig', @@ -336,50 +576,50 @@ def test_invoke_removes_pod_config_when_host_config_is_absent(agent, config_file 'exec', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', 'rm', '-f', '/etc/datadog-agent/conf.d/velero.d/conf.yaml', - ] in calls + ] in command_calls(run_command) -def test_stop_is_idempotent_when_pod_is_absent(agent, run_command): +def test_stop_is_idempotent_when_namespace_is_absent(agent, run_command): agent.stop() - calls = command_calls(run_command) - prefix = ['kubectl', '--kubeconfig', '/tmp/kubeconfig'] - selector = 'ddev.datadoghq.com/environment=test-owner' - assert calls == [ - [*prefix, 'get', 'namespace', agent._namespace, '--ignore-not-found=true', '-o', 'json'], + assert command_calls(run_command) == [ [ - *prefix, - 'delete', + 'kubectl', + '--kubeconfig', + '/tmp/kubeconfig', + 'get', 'namespace', - '--selector', - selector, + agent._namespace, '--ignore-not-found=true', - '--wait=true', - '--timeout=120s', + '-o', + 'json', ], [ - *prefix, - 'delete', + 'kubectl', + '--kubeconfig', + '/tmp/kubeconfig', + 'get', 'clusterrole,clusterrolebinding', '--selector', - selector, - '--ignore-not-found=true', + 'ddev.datadoghq.com/environment=test-owner', + '-o', + 'name', ], ] -def test_stop_does_not_enter_or_delete_another_environment_namespace(agent, app, mocker): +def test_stop_does_not_touch_another_environment_namespace(agent, app, mocker): namespace = {'metadata': {'labels': {'ddev.datadoghq.com/environment': 'another-owner'}}} def run(command, **kwargs): - if command[-1:] == ['json'] and 'namespace' in command: + if 'namespace' in command: return successful_process(command, stdout=json.dumps(namespace).encode()) return successful_process(command) @@ -388,11 +628,55 @@ def run(command, **kwargs): agent.stop() calls = command_calls(run_command) - assert not any('pod' in command for command in calls) - delete_calls = [command for command in calls if 'delete' in command] - assert len(delete_calls) == 2 - assert all('ddev.datadoghq.com/environment=test-owner' in command for command in delete_calls) - assert all('another-owner' not in command for command in delete_calls) + assert len(calls) == 2 + assert calls[0][-5:-3] == ['namespace', agent._namespace] + assert not any(command[0] == 'helm' or 'delete' in command for command in calls) + + +def test_stop_preserves_state_when_namespace_is_missing_but_cluster_resources_remain(agent, app, mocker): + def run(command, **kwargs): + if 'clusterrole,clusterrolebinding' in command: + return successful_process( + command, stdout=f'clusterrole.rbac.authorization.k8s.io/{agent._namespace}\n'.encode() + ) + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + with pytest.raises(RuntimeError, match='cluster-scoped resources remain'): + agent.stop() + + calls = command_calls(run_command) + assert not any(command[0] == 'helm' or 'delete' in command for command in calls) + + +def test_stop_revalidates_namespace_uid_before_helm_uninstall(agent, metadata, app, mocker): + metadata['kubernetes']['_namespace_uid'] = 'owned-uid' + namespace_queries = 0 + + def run(command, **kwargs): + nonlocal namespace_queries + if command[-2:] == ['-o', 'json'] and 'namespace' in command: + namespace_queries += 1 + uid = 'owned-uid' if namespace_queries == 1 else 'replacement-uid' + namespace = { + 'metadata': { + 'uid': uid, + 'labels': {'ddev.datadoghq.com/environment': 'test-owner'}, + } + } + return successful_process(command, stdout=json.dumps(namespace).encode()) + if 'get' in command and 'pods' in command: + return successful_process(command, stdout=b'{"items":[]}') + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + with pytest.raises(RuntimeError, match='ownership changed'): + agent.stop() + + assert namespace_queries == 2 + assert not any(command[0] == 'helm' or 'delete' in command for command in command_calls(run_command)) def test_restart_resynchronizes_config_auto_conf_and_editable_sources( @@ -414,7 +698,7 @@ def test_restart_resynchronizes_config_auto_conf_and_editable_sources( '--container', 'agent', local_package.name, - f'{agent._namespace}/ddev-agent:/home/velero-source', + f'{agent._namespace}/{POD_NAME}:/home/velero-source', ] config_copy = [ *prefix, @@ -422,14 +706,14 @@ def test_restart_resynchronizes_config_auto_conf_and_editable_sources( '--container', 'agent', config_file.name, - f'{agent._namespace}/ddev-agent:/etc/datadog-agent/conf.d/velero.d/conf.yaml', + f'{agent._namespace}/{POD_NAME}:/etc/datadog-agent/conf.d/velero.d/conf.yaml', ] restart = [ *prefix, 'exec', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', @@ -443,17 +727,87 @@ def test_restart_resynchronizes_config_auto_conf_and_editable_sources( assert not any('pip' in command for command in calls) +def test_stop_uninstalls_release_before_namespace(agent, app, mocker): + namespace = {'metadata': {'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + + def run(command, **kwargs): + if command[-2:] == ['-o', 'json'] and 'namespace' in command: + return successful_process(command, stdout=json.dumps(namespace).encode()) + if 'get' in command and 'pods' in command: + return successful_process(command, stdout=json.dumps({'items': [pod_data()]}).encode()) + if command[:2] == ['helm', 'list']: + return successful_process(command, stdout=b'[{"name":"ddev-agent"}]') + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + agent.stop() + + calls = command_calls(run_command) + uninstall = next(command for command in calls if command[:2] == ['helm', 'uninstall']) + delete = next(command for command in calls if 'delete' in command) + assert calls.index(uninstall) < calls.index(delete) + assert not any('clusterrole' in command for command in calls) + + +def test_stop_runs_hook_in_selected_non_ready_pod(agent, app, mocker): + agent.metadata['stop_commands'] = ['echo stopping'] + namespace = {'metadata': {'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + + def run(command, **kwargs): + if command[-2:] == ['-o', 'json'] and 'namespace' in command: + return successful_process(command, stdout=json.dumps(namespace).encode()) + if 'get' in command and 'pods' in command: + return successful_process(command, stdout=json.dumps({'items': [pod_data(ready=False)]}).encode()) + if command[:2] == ['helm', 'list']: + return successful_process(command, stdout=b'[]') + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + agent.stop() + + stop_hook = next(command for command in command_calls(run_command) if command[-2:] == ['echo', 'stopping']) + assert f'pod/{POD_NAME}' in stop_hook + + +def test_stop_preserves_namespace_when_helm_uninstall_fails(agent, app, mocker): + namespace = {'metadata': {'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + + def run(command, **kwargs): + if command[-2:] == ['-o', 'json'] and 'namespace' in command: + return successful_process(command, stdout=json.dumps(namespace).encode()) + if 'get' in command and 'pods' in command: + return successful_process(command, stdout=json.dumps({'items': []}).encode()) + if command[:2] == ['helm', 'list']: + return successful_process(command, stdout=b'[{"name":"ddev-agent"}]') + if command[:2] == ['helm', 'uninstall']: + return subprocess.CompletedProcess(command, 1, stdout=b'helm failed') + return successful_process(command) + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + with pytest.raises(RuntimeError, match='preserving the Helm release namespace'): + agent.stop() + + calls = command_calls(run_command) + assert any(command[:2] == ['helm', 'uninstall'] for command in calls) + assert not any('delete' in command for command in calls) + + def test_stop_cleans_resources_after_stop_command_failure(agent, app, mocker): agent.metadata['stop_commands'] = ['false'] + namespace = {'metadata': {'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} def run(command, **kwargs): - if command[-1:] == ['json'] and 'namespace' in command: - namespace = {'metadata': {'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + if command[-2:] == ['-o', 'json'] and 'namespace' in command: return successful_process(command, stdout=json.dumps(namespace).encode()) - if command[-1:] == ['name'] and 'pod' in command: - return successful_process(command, stdout=b'pod/ddev-agent\n') + if 'get' in command and 'pods' in command: + return successful_process(command, stdout=json.dumps({'items': [pod_data()]}).encode()) if command[-1:] == ['false']: raise subprocess.CalledProcessError(1, command) + if command[:2] == ['helm', 'list']: + return successful_process(command, stdout=b'[]') return successful_process(command) run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) @@ -463,29 +817,41 @@ def run(command, **kwargs): calls = command_calls(run_command) assert any('namespace' in command and 'delete' in command for command in calls) - assert any('clusterrole,clusterrolebinding' in command and 'delete' in command for command in calls) -def test_shell_and_logs_use_backend_commands(agent, run_command): +def test_shell_and_logs_use_dynamically_selected_pod(agent, run_command): agent.enter_shell() agent.show_logs() - calls = command_calls(run_command) + calls = operational_calls(run_command) prefix = ['kubectl', '--kubeconfig', '/tmp/kubeconfig'] assert calls == [ + [ + *prefix, + 'exec', + '--namespace', + agent._namespace, + f'pod/{POD_NAME}', + '--container', + 'agent', + '--', + 'test', + '-f', + '/home/.ddev-agent-prepared', + ], [ *prefix, 'exec', '-it', '--namespace', agent._namespace, - 'pod/ddev-agent', + f'pod/{POD_NAME}', '--container', 'agent', '--', 'bash', ], - [*prefix, 'logs', '--namespace', agent._namespace, 'pod/ddev-agent', '--container', 'agent'], + [*prefix, 'logs', '--namespace', agent._namespace, f'pod/{POD_NAME}', '--container', 'agent'], ] assert run_command.call_args_list[-1].kwargs['check'] is True @@ -496,6 +862,10 @@ def test_shell_and_logs_use_backend_commands(agent, run_command): ({}, 'must contain a `kubernetes` mapping'), ({'kubernetes': {}}, 'non-empty `kubeconfig`'), ({'kubernetes': {'kubeconfig': '/tmp/config', 'namespace': 'INVALID'}}, 'Invalid Kubernetes Agent namespace'), + ( + {'kubernetes': {'kubeconfig': '/tmp/config', 'namespace': '1-invalid-service-name'}}, + 'Invalid Kubernetes Agent namespace', + ), ], ) def test_metadata_validation(app, get_integration, config_file, metadata, match): diff --git a/ddev/tests/e2e/agent/test_kubernetes_helm.py b/ddev/tests/e2e/agent/test_kubernetes_helm.py new file mode 100644 index 0000000000000..3a95aa74d6167 --- /dev/null +++ b/ddev/tests/e2e/agent/test_kubernetes_helm.py @@ -0,0 +1,384 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +import json +import subprocess + +import pytest + +from ddev.e2e.agent.kubernetes_helm import AgentImage, HelmDaemonSetDeployment, parse_agent_image + + +def successful_process(command, *, stdout=b'', stderr=None): + return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr=stderr) + + +@pytest.fixture +def metadata(): + return {'pod_labels': {'purpose': 'discovery-e2e'}} + + +@pytest.fixture +def deployment(app, temp_dir, metadata): + return HelmDaemonSetDeployment( + platform=app.platform, + kubeconfig='/tmp/kubeconfig', + namespace='ddev-agent-velero-12345678', + owner_id='test-owner', + kubernetes_metadata=metadata, + state_dir=temp_dir, + wait_timeout=90, + ) + + +@pytest.mark.parametrize( + 'reference, expected', + [ + ( + 'registry.datadoghq.com/agent-dev:master-py3', + AgentImage(repository='registry.datadoghq.com/agent-dev', tag='master-py3'), + ), + ('datadog/agent:7.81.1', AgentImage(repository='datadog/agent', tag='7.81.1')), + ( + 'localhost:5000/datadog-agent:test', + AgentImage(repository='localhost:5000/datadog-agent', tag='test'), + ), + ( + f'registry.example.com/agent@sha256:{"a" * 64}', + AgentImage(repository='registry.example.com/agent', digest=f'sha256:{"a" * 64}'), + ), + ], +) +def test_parse_agent_image(reference, expected): + assert parse_agent_image(reference) == expected + + +@pytest.mark.parametrize( + 'reference', + [ + '', + 'registry.example.com/agent', + 'registry.example.com/agent:', + 'registry.example.com/agent@sha256:not-hex', + f'registry.example.com/agent@sha256:{"a" * 63}', + 'registry.example.com/agent:bad tag', + ], +) +def test_parse_agent_image_rejects_invalid_references(reference): + with pytest.raises(ValueError, match='image'): + parse_agent_image(reference) + + +def test_values_disable_auxiliary_workloads_and_map_image_environment_and_labels(deployment, metadata): + metadata['pod_labels']['ddev.datadoghq.com/environment'] = 'another-owner' + metadata['pod_labels']['app.kubernetes.io/component'] = 'broken-selector' + metadata['pod_labels']['app'] = 'broken-selector' + metadata['image_pull_policy'] = 'Never' + + values = deployment.values( + 'localhost:5000/datadog-agent:test', + { + 'DD_API_KEY': 'secret', + 'DD_SITE': 'datadoghq.eu', + 'DD_KUBELET_TLS_VERIFY': 'true', + 'DD_LOG_LEVEL': 'debug', + }, + node_name='kind-control-plane', + ) + + assert values['fullnameOverride'] == deployment.namespace + assert values['targetSystem'] == 'linux' + assert values['commonLabels'] == {'ddev.datadoghq.com/environment': 'test-owner'} + assert values['clusterAgent'] == { + 'enabled': False, + 'admissionController': {'enabled': False}, + } + assert values['agents']['enabled'] is True + assert values['agents']['instanceLabelOverride'] == 'test-owner' + assert values['agents']['image'] == { + 'repository': 'localhost:5000/datadog-agent', + 'tag': 'test', + 'doNotCheckTag': True, + 'pullPolicy': 'Never', + } + assert values['agents']['podLabels'] == { + 'purpose': 'discovery-e2e', + 'ddev.datadoghq.com/environment': 'test-owner', + 'app.kubernetes.io/component': 'agent', + 'app': deployment.namespace, + } + assert values['agents']['tolerations'] == [{'operator': 'Exists'}] + assert values['agents']['affinity'] == { + 'nodeAffinity': { + 'requiredDuringSchedulingIgnoredDuringExecution': { + 'nodeSelectorTerms': [ + { + 'matchFields': [ + { + 'key': 'metadata.name', + 'operator': 'In', + 'values': ['kind-control-plane'], + } + ] + } + ] + } + } + } + + datadog = values['datadog'] + assert datadog['apiKey'] == 'secret' + assert datadog['site'] == 'datadoghq.eu' + assert datadog['logLevel'] == 'debug' + assert datadog['kubelet']['tlsVerify'] is True + for feature in ( + datadog['clusterChecks'], + datadog['kubeStateMetricsCore'], + datadog['logs'], + datadog['orchestratorExplorer'], + datadog['operator'], + datadog['remoteConfiguration'], + ): + assert feature['enabled'] is False + assert datadog['collectEvents'] is False + assert datadog['leaderElection'] is False + assert datadog['useHostPID'] is False + assert datadog['apm'] == {'socketEnabled': False, 'portEnabled': False} + assert datadog['processAgent'] == { + 'enabled': False, + 'processCollection': False, + 'processDiscovery': False, + 'containerCollection': False, + } + assert datadog['dogstatsd'] == { + 'port': 8125, + 'useSocketVolume': False, + 'nonLocalTraffic': True, + 'originDetection': False, + 'tagCardinality': 'low', + } + assert datadog['expvarPort'] == 5000 + assert datadog['containerLifecycle']['enabled'] is False + assert datadog['discovery'] == {'enabled': False, 'networkStats': {'enabled': False}} + + assert values['agents']['containers']['agent']['command'] == ['/bin/entrypoint.sh'] + agent_container = values['agents']['containers']['agent'] + assert agent_container['securityContext'] == {'readOnlyRootFilesystem': False} + local_probe = { + 'exec': {'command': ['/bin/true']}, + 'initialDelaySeconds': 0, + 'periodSeconds': 1, + 'timeoutSeconds': 1, + 'successThreshold': 1, + 'failureThreshold': 3, + } + for probe in ('livenessProbe', 'readinessProbe', 'startupProbe'): + assert agent_container[probe] == local_probe + env = {item['name']: item['value'] for item in agent_container['env']} + assert env == {'DD_AUTOCONFIG_FROM_ENVIRONMENT': 'true'} + for chart_mapped_name in ('DD_API_KEY', 'DD_SITE', 'DD_KUBELET_TLS_VERIFY', 'DD_APM_ENABLED', 'DD_LOG_LEVEL'): + assert chart_mapped_name not in env + for chart_generated_name in ('DD_HOSTNAME', 'DD_KUBERNETES_KUBELET_HOST', 'DD_KUBERNETES_KUBELET_NODENAME'): + assert chart_generated_name not in env + + +def test_values_map_digest_reference(deployment): + digest = f'sha256:{"b" * 64}' + + values = deployment.values(f'registry.example.com/agent@{digest}', {}, node_name='kind-control-plane') + + assert values['agents']['image'] == { + 'repository': 'registry.example.com/agent', + 'digest': digest, + 'doNotCheckTag': True, + 'pullPolicy': 'Always', + } + + +@pytest.mark.parametrize( + 'metadata_update, env_vars, match', + [ + ({'pod_labels': []}, {}, 'pod_labels'), + ({'image_pull_policy': 'Sometimes'}, {}, 'image_pull_policy'), + ({}, {'DD_KUBELET_TLS_VERIFY': 'maybe'}, 'DD_KUBELET_TLS_VERIFY'), + ({}, {'DD_APM_ENABLED': 'true'}, 'must remain false'), + ({}, {'DD_LOGS_ENABLED': 'true'}, 'log collection'), + ({}, {'DD_APM_RECEIVER_PORT': '8127'}, 'managed by the Helm chart'), + ], +) +def test_values_validate_metadata_before_install(deployment, metadata, metadata_update, env_vars, match): + metadata.update(metadata_update) + + with pytest.raises(ValueError, match=match): + deployment.values('registry.example.com/agent:test', env_vars, node_name='kind-control-plane') + + +def test_helm_environment_is_isolated_under_environment_state(deployment, temp_dir): + environment = deployment.helm_environment + + assert environment['HELM_CACHE_HOME'] == str(temp_dir / 'helm' / 'cache') + assert environment['HELM_CONFIG_HOME'] == str(temp_dir / 'helm' / 'config') + assert environment['HELM_DATA_HOME'] == str(temp_dir / 'helm' / 'data') + assert all((temp_dir / 'helm' / directory).is_dir() for directory in ('cache', 'config', 'data')) + + +def test_check_helm_reports_missing_executable(deployment, app, mocker): + mocker.patch.object(app.platform, 'run_command', side_effect=FileNotFoundError('helm')) + + with pytest.raises(RuntimeError, match='requires the `helm` executable'): + deployment.check_helm() + + +def test_agent_pod_selects_ready_owned_agent_and_ignores_terminating_pod(deployment, app, mocker): + pods = { + 'items': [ + { + 'metadata': {'name': 'old-agent', 'uid': 'old', 'deletionTimestamp': 'now'}, + 'spec': {'nodeName': 'kind-control-plane', 'containers': [{'name': 'agent'}]}, + 'status': {'phase': 'Running', 'conditions': [{'type': 'Ready', 'status': 'True'}]}, + }, + { + 'metadata': {'name': 'new-agent', 'uid': 'new'}, + 'spec': {'nodeName': 'kind-control-plane', 'containers': [{'name': 'agent'}]}, + 'status': {'phase': 'Running', 'conditions': [{'type': 'Ready', 'status': 'True'}]}, + }, + ] + } + run_command = mocker.patch.object( + app.platform, + 'run_command', + return_value=successful_process([], stdout=json.dumps(pods).encode()), + ) + + pod = deployment.agent_pod() + + assert pod.name == 'new-agent' + assert pod.uid == 'new' + assert pod.node_name == 'kind-control-plane' + command = run_command.call_args.args[0] + assert command == [ + 'kubectl', + '--kubeconfig', + '/tmp/kubeconfig', + 'get', + 'pods', + '--namespace', + deployment.namespace, + '--selector', + 'app.kubernetes.io/component=agent,ddev.datadoghq.com/environment=test-owner', + '-o', + 'json', + ] + + +@pytest.mark.parametrize( + 'items, ready, expected_count', + [ + ([], True, 0), + ( + [ + { + 'metadata': {'name': 'unready', 'uid': 'one'}, + 'spec': {'nodeName': 'kind-control-plane', 'containers': [{'name': 'agent'}]}, + 'status': {'phase': 'Running', 'conditions': [{'type': 'Ready', 'status': 'False'}]}, + } + ], + True, + 0, + ), + ( + [ + { + 'metadata': {'name': name, 'uid': name}, + 'spec': {'nodeName': 'kind-control-plane', 'containers': [{'name': 'agent'}]}, + 'status': {'phase': 'Running', 'conditions': [{'type': 'Ready', 'status': 'True'}]}, + } + for name in ('one', 'two') + ], + True, + 2, + ), + ], +) +def test_agent_pod_rejects_zero_or_multiple_candidates(deployment, app, mocker, items, ready, expected_count): + mocker.patch.object( + app.platform, + 'run_command', + return_value=successful_process([], stdout=json.dumps({'items': items}).encode()), + ) + + with pytest.raises(RuntimeError, match=rf'found {expected_count}'): + deployment.agent_pod(ready=ready) + + +def test_uninstall_skips_missing_release_without_status_preflight(deployment, app, mocker): + namespace = {'metadata': {'uid': 'namespace-uid', 'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + + def run(command, **kwargs): + if ( + 'clusterrole,clusterrolebinding' in command + or 'all,secret,configmap,serviceaccount,role,rolebinding' in command + ): + return successful_process(command) + if command[0] == 'kubectl': + return successful_process(command, stdout=json.dumps(namespace).encode()) + return subprocess.CompletedProcess(command, 1, stdout=b'Error: release: not found') + + run_command = mocker.patch.object(app.platform, 'run_command', side_effect=run) + + deployment.uninstall() + + assert len(run_command.call_args_list) == 4 + assert any(call.args[0][:2] == ['helm', 'uninstall'] for call in run_command.call_args_list) + + +def test_uninstall_preserves_state_when_release_metadata_is_missing_but_cluster_resources_remain( + deployment, app, mocker +): + namespace = {'metadata': {'uid': 'namespace-uid', 'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + + def run(command, **kwargs): + if 'clusterrole,clusterrolebinding' in command: + return successful_process(command, stdout=b'clusterrole.rbac.authorization.k8s.io/ddev-agent') + if command[0] == 'kubectl': + return successful_process(command, stdout=json.dumps(namespace).encode()) + return subprocess.CompletedProcess(command, 1, stdout=b'Error: release: not found') + + mocker.patch.object(app.platform, 'run_command', side_effect=run) + + with pytest.raises(RuntimeError, match='release metadata is missing'): + deployment.uninstall() + + +def test_uninstall_preserves_state_when_release_metadata_is_missing_but_namespaced_resources_remain( + deployment, app, mocker +): + namespace = {'metadata': {'uid': 'namespace-uid', 'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + + def run(command, **kwargs): + if 'clusterrole,clusterrolebinding' in command: + return successful_process(command) + if 'all,secret,configmap,serviceaccount,role,rolebinding' in command: + return successful_process(command, stdout=b'service/ddev-agent') + if command[0] == 'kubectl': + return successful_process(command, stdout=json.dumps(namespace).encode()) + return subprocess.CompletedProcess(command, 1, stdout=b'Error: release: not found') + + mocker.patch.object(app.platform, 'run_command', side_effect=run) + + with pytest.raises(RuntimeError, match='release metadata is missing'): + deployment.uninstall() + + +def test_uninstall_reports_non_missing_release_error(deployment, app, mocker): + namespace = {'metadata': {'uid': 'namespace-uid', 'labels': {'ddev.datadoghq.com/environment': 'test-owner'}}} + + def run(command, **kwargs): + if command[0] == 'kubectl': + return successful_process(command, stdout=json.dumps(namespace).encode()) + return subprocess.CompletedProcess(command, 1, stdout=b'Kubernetes API unavailable') + + mocker.patch.object(app.platform, 'run_command', side_effect=run) + + with pytest.raises(RuntimeError, match='Unable to uninstall Kubernetes Agent Helm release'): + deployment.uninstall() diff --git a/docs/developer/ddev/plugins.md b/docs/developer/ddev/plugins.md index fbf1c4599fa1d..02ee190a81fc0 100644 --- a/docs/developer/ddev/plugins.md +++ b/docs/developer/ddev/plugins.md @@ -141,5 +141,5 @@ is desired. This fixture is responsible for starting and stopping environments a - `env_vars` - A `dict` of environment variables and their values that will be present when starting the Agent. - `docker_volumes` - A `list` of `str` representing [Docker volume mounts][docker-volume-docs] if `agent_type` is `docker` e.g. `/local/path:/agent/container/path:ro`. - `docker_platform` - The container architecture to use if `agent_type` is `docker`. Currently, we support `linux` (default) and `windows`. -- `kubernetes` - Configuration used when `agent_type` is `kubernetes`. It requires `kubeconfig` and optionally accepts `namespace`, `auto_conf`, `pod_labels`, `image_pull_policy`, and `wait_timeout`. The backend owns the namespace, which must not already exist. +- `kubernetes` - Configuration used when `agent_type` is `kubernetes`. It requires `kubeconfig` and optionally accepts `namespace`, `auto_conf`, `pod_labels`, `image_pull_policy`, and `wait_timeout`. The backend requires `kubectl` and Helm, installs a pinned official Datadog chart, and owns the namespace, which must not already exist. - `logs_config` - A `list` of configs that will be used by the Logs Agent. You will never need to use this directly, but rather via [higher level abstractions](test.md#logs). diff --git a/docs/developer/ddev/test.md b/docs/developer/ddev/test.md index 1e6ed2b49f1a9..bd507f16e494d 100644 --- a/docs/developer/ddev/test.md +++ b/docs/developer/ddev/test.md @@ -93,9 +93,10 @@ Note: Vagrant environments are not supported in CI environments due to virtualiz ### Kubernetes Agent -The Kubernetes Agent backend runs the Datadog Agent inside an existing Kubernetes test cluster. The cluster lifecycle -remains the responsibility of an environment helper such as `kind_run`; the backend only requires a kubeconfig and uses -standard `kubectl` operations, so it is not tied to Kind. +The Kubernetes Agent backend runs the Datadog Agent inside an existing Kubernetes test cluster as a Node Agent +DaemonSet from a pinned version of the official Datadog Helm chart. The cluster lifecycle remains the responsibility of +an environment helper such as `kind_run`; the backend consumes a kubeconfig and is not tied to Kind. Both `kubectl` and +Helm must be installed on the host. Helm cache, configuration, and data are isolated within the environment's ddev state. ```python @pytest.fixture(scope='session') @@ -118,11 +119,14 @@ def dd_environment(): The backend uses the Agent image selected by `ddev env start --agent` or `DDEV_E2E_AGENT`, installs and synchronizes local packages requested by `--dev` or `--base`, and implements Agent commands through `kubectl exec`. Static and -discovery E2E tests therefore continue to use `dd_agent_check` and `dd_agent_check_discovery`. +discovery E2E tests therefore continue to use `dd_agent_check` and `dd_agent_check_discovery`. The chart deployment +runs only the core Node Agent; the Operator, Cluster Agent, APM, logs, process collection, and other auxiliary workloads +remain disabled for this E2E backend. Agent images default to the `Always` pull policy so mutable release and development tags are refreshed. Environments that import a local image into the cluster can set `image_pull_policy` to `IfNotPresent` or `Never`. A custom `namespace` -must not already exist; the backend owns and deletes the namespace and its cluster-scoped RBAC resources. +must not already exist. The backend owns the namespace and Helm release; teardown uninstalls the chart, including its +cluster-scoped RBAC resources, before deleting the namespace. The initial implementation supports exactly one schedulable Kubernetes node. It rejects multi-node clusters until Agent targeting or fan-out semantics are defined. From 9504b5e727e2b15f030adaea99c71158d2a8733e Mon Sep 17 00:00:00 2001 From: Enrico Donnici Date: Fri, 24 Jul 2026 11:18:46 +0000 Subject: [PATCH 2/2] Make Helm cache assertion platform independent --- ddev/tests/e2e/agent/test_kubernetes.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ddev/tests/e2e/agent/test_kubernetes.py b/ddev/tests/e2e/agent/test_kubernetes.py index 8580cddb7b02c..897f1a94b1e9b 100644 --- a/ddev/tests/e2e/agent/test_kubernetes.py +++ b/ddev/tests/e2e/agent/test_kubernetes.py @@ -3,6 +3,7 @@ # Licensed under a 3-clause BSD style license (see LICENSE) import json import subprocess +from pathlib import Path import pytest @@ -190,7 +191,7 @@ def test_start_installs_pinned_helm_chart_and_prepares_selected_agent( assert values['datadog']['dogstatsd']['useSocketVolume'] is False assert values['datadog']['operator']['enabled'] is False assert values['clusterAgent']['enabled'] is False - assert helm_call.kwargs['env']['HELM_CACHE_HOME'].endswith('/helm/cache') + assert Path(helm_call.kwargs['env']['HELM_CACHE_HOME']).parts[-2:] == ('helm', 'cache') assert [ *prefix,