Coverage for gco/services/manifest_processor.py: 91.55%
673 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""
2Manifest Processor Service for GCO (Global Capacity Orchestrator on AWS).
4This service processes Kubernetes manifest submissions, validates them against
5security and resource constraints, and applies them to the cluster.
7Key Features:
8- Validates manifests for required fields and structure
9- Enforces namespace restrictions (only allowed namespaces)
10- Enforces resource limits (CPU, memory, GPU per manifest)
11- Validates security context (no privileged containers)
12- Validates image sources (trusted registries only)
13- Supports dry-run mode for validation without applying
15Security Validations:
16- Namespace must be in allowed list (default: gco-jobs)
17- No privileged containers or privilege escalation
18- Images must be from trusted registries
19- Resource requests/limits within configured maximums
21Environment Variables:
22 CLUSTER_NAME: Name of the EKS cluster
23 REGION: AWS region of the cluster
24 MAX_CPU_PER_MANIFEST: Maximum CPU (millicores) per manifest (default: 10000)
25 MAX_MEMORY_PER_MANIFEST: Maximum memory per manifest (default: 32Gi)
26 MAX_GPU_PER_MANIFEST: Maximum GPUs per manifest (default: 4)
27 ALLOWED_NAMESPACES: Comma-separated list of allowed namespaces
28 VALIDATION_ENABLED: Enable/disable validation (default: true)
30Usage:
31 processor = create_manifest_processor_from_env()
32 response = await processor.process_manifest_submission(request)
33"""
35from __future__ import annotations
37import copy
38import hashlib
39import logging
40import os
41import re
42from typing import Any, cast
44import yaml
45from kubernetes import client, config, dynamic
46from kubernetes.client.models import V1Job
47from kubernetes.client.rest import ApiException
48from kubernetes.dynamic.exceptions import ResourceNotFoundError
50from gco.models import (
51 ManifestSubmissionRequest,
52 ManifestSubmissionResponse,
53 ResourceStatus,
54)
55from gco.services.structured_logging import configure_structured_logging, sanitize_log_value
57# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
58# Generated at (UTC): 2026-07-18T01:03:40Z
59# Flowchart(s) generated from this file:
60# * ``ManifestProcessor.apply_queued_job`` -> ``diagrams/code_diagrams/gco/services/manifest_processor.ManifestProcessor_apply_queued_job.html``
61# (PNG: ``diagrams/code_diagrams/gco/services/manifest_processor.ManifestProcessor_apply_queued_job.png``)
62# Regenerate with ``python diagrams/code_diagrams/generate.py``.
63# <pyflowchart-code-diagram> END
66# NOTE: No logging.basicConfig() here. This module is imported by the CLI
67# (cli/jobs.py, cli/commands/*_cmd.py) as a library for YAML loading helpers.
68# Calling basicConfig() at import time would configure the root logger with
69# INFO-level output, causing noisy botocore/urllib3 INFO messages on every
70# CLI command. Container entry points (manifest_api.py) do their own
71# basicConfig() call.
72logger = logging.getLogger(__name__)
75# Accelerator resource keys and their corresponding node taint keys. GCO
76# nodepools taint accelerator nodes with these keys (authoritative list:
77# regional_stack._ADDON_NODE_TOLERATIONS), so a job requesting one of these
78# resources must carry a matching toleration or it will never schedule.
79# Taint key == resource key for all three. Kept in sync with the mirror in
80# gco/services/queue_processor.py::ACCELERATOR_TAINTS.
81ACCELERATOR_TAINTS = ("nvidia.com/gpu", "aws.amazon.com/neuron", "vpc.amazonaws.com/efa")
83# Authoritative resource-kind policy shared by the REST and SQS submission
84# paths. Keep the fallback here so both services fail closed to the same set
85# when ALLOWED_KINDS is not explicitly configured.
86DEFAULT_ALLOWED_KINDS = (
87 "Job",
88 "CronJob",
89 "Deployment",
90 "StatefulSet",
91 "DaemonSet",
92 "Service",
93 "ConfigMap",
94 "Pod",
95)
98class RetryableQueuedJobApplyError(RuntimeError):
99 """A deterministic queued Job apply can be retried or adopted safely."""
102class QueuedJobNotCreatedError(ValueError):
103 """A queued Job was rejected before any Kubernetes operation began."""
106def _is_retryable_kubernetes_api_error(error: ApiException) -> bool:
107 """Classify throttling, server, and transport-like Kubernetes API failures."""
108 try:
109 status = int(error.status or 0)
110 except TypeError, ValueError:
111 status = 0
112 return status == 0 or status in {408, 429} or status >= 500
115def validate_resource_kind(
116 manifest: dict[str, Any], allowed_kinds: set[str] | tuple[str, ...] = DEFAULT_ALLOWED_KINDS
117) -> tuple[bool, str | None]:
118 """Validate a manifest kind against the shared submission allowlist."""
119 kind = manifest.get("kind", "")
120 allowed = set(allowed_kinds)
121 if kind not in allowed:
122 return (
123 False,
124 f"Resource kind '{kind}' is not allowed. Allowed kinds: {sorted(allowed)}",
125 )
126 return True, None
129def _positive_quantity(value: Any) -> bool:
130 """True if a K8s resource quantity is present and greater than zero."""
131 if value is None:
132 return False
133 try:
134 return float(value) > 0
135 except TypeError, ValueError:
136 # A non-numeric quantity is still an explicit request.
137 return True
140def _toleration_matches(tolerations: list[dict[str, Any]], taint_key: str) -> bool:
141 """Return True if *tolerations* tolerates the ``<taint_key>=true:NoSchedule`` taint.
143 A toleration matches when its ``key`` equals *taint_key*, its effect is
144 empty (matches all effects) or ``NoSchedule``, and it either uses
145 ``operator: Exists`` or ``operator: Equal`` with ``value: "true"``.
146 Kept in sync with the mirror in queue_processor._toleration_matches.
147 """
148 for tol in tolerations:
149 if not isinstance(tol, dict) or tol.get("key") != taint_key: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 continue
151 effect = tol.get("effect", "")
152 if effect not in ("", "NoSchedule"):
153 continue
154 operator = tol.get("operator", "Equal")
155 if operator == "Exists":
156 return True
157 if operator == "Equal" and str(tol.get("value")) == "true": 157 ↛ 148line 157 didn't jump to line 148 because the condition on line 157 was always true
158 return True
159 return False
162# ---------------------------------------------------------------------------
163# YAML Alias Rejection Loader
164# ---------------------------------------------------------------------------
167class NoAliasSafeLoader(yaml.SafeLoader):
168 """A YAML SafeLoader that rejects anchors and aliases.
170 YAML anchors (``&anchor``) and aliases (``*anchor``) can be used to
171 construct exponentially large data structures (billion-laughs attack).
172 This loader raises an error when any alias is encountered, preventing
173 such attacks at the parsing stage.
174 """
176 def compose_node(self, parent: Any, index: Any) -> Any:
177 if self.check_event(yaml.AliasEvent): # type: ignore[no-untyped-call]
178 event = self.get_event() # type: ignore[no-untyped-call]
179 raise yaml.composer.ComposerError(
180 None,
181 None,
182 "YAML aliases are not allowed "
183 "(security policy: yaml_allow_aliases=false), "
184 f"found alias *{event.anchor}",
185 event.start_mark,
186 )
187 return super().compose_node(parent, index)
190def safe_load_yaml(stream: str | Any, *, allow_aliases: bool = False) -> Any:
191 """Load a single YAML document with optional alias rejection.
193 Args:
194 stream: YAML string or file-like object.
195 allow_aliases: If False (default), reject YAML anchors/aliases.
197 Returns:
198 Parsed YAML document.
200 Raises:
201 yaml.YAMLError: If the document is invalid or contains aliases
202 when ``allow_aliases`` is False.
203 """
204 loader_cls = yaml.SafeLoader if allow_aliases else NoAliasSafeLoader
205 # Loader is always a SafeLoader subclass (SafeLoader or NoAliasSafeLoader),
206 # so this is equivalent to yaml.safe_load. Bandit's B506 check does not
207 # recognize the custom loader as safe.
208 return yaml.load(stream, Loader=loader_cls) # nosec B506
211def safe_load_all_yaml(stream: str | Any, *, allow_aliases: bool = False) -> list[Any]:
212 """Load all YAML documents from a stream with optional alias rejection.
214 Args:
215 stream: YAML string or file-like object.
216 allow_aliases: If False (default), reject YAML anchors/aliases.
218 Returns:
219 List of parsed YAML documents (``None`` documents are skipped).
221 Raises:
222 yaml.YAMLError: If any document is invalid or contains aliases
223 when ``allow_aliases`` is False.
224 """
225 loader_cls = yaml.SafeLoader if allow_aliases else NoAliasSafeLoader
226 # Loader is always a SafeLoader subclass, so this is equivalent to
227 # yaml.safe_load_all. Bandit's B506 check does not recognize the custom
228 # loader as safe.
229 return [
230 doc
231 for doc in yaml.load_all(stream, Loader=loader_cls)
232 if doc is not None # nosec B506
233 ]
236class ManifestProcessor:
237 """
238 Processes Kubernetes manifest submissions and applies them to the cluster
239 """
241 def __init__(self, cluster_id: str, region: str, config_dict: dict[str, Any]):
242 self.cluster_id = cluster_id
243 self.region = region
244 self.config = config_dict
246 # Initialize Kubernetes clients
247 try:
248 # Try to load in-cluster config first (when running in pod)
249 config.load_incluster_config()
250 logger.info("Loaded in-cluster Kubernetes configuration")
251 except config.ConfigException:
252 try:
253 # Fall back to local kubeconfig (for development)
254 config.load_kube_config()
255 logger.info("Loaded local Kubernetes configuration")
256 except config.ConfigException as e:
257 logger.error(f"Failed to load Kubernetes configuration: {e}")
258 raise
260 # Initialize API clients
261 self.api_client = client.ApiClient()
262 self.api_client.configuration.request_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
263 self.core_v1 = client.CoreV1Api()
264 self.apps_v1 = client.AppsV1Api()
265 self.batch_v1 = client.BatchV1Api()
266 self.networking_v1 = client.NetworkingV1Api()
267 self.custom_objects = client.CustomObjectsApi()
269 # Dynamic client for CRDs - lazy initialized to avoid cluster connection during init
270 self._dynamic_client: dynamic.DynamicClient | None = None
272 # Timeout for Kubernetes API calls (seconds)
273 self._k8s_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
275 # Resource quotas and limits
276 self.max_cpu_per_manifest = self._parse_cpu_string(
277 config_dict.get("max_cpu_per_manifest", "10")
278 )
279 self.max_memory_per_manifest = self._parse_memory_string(
280 config_dict.get("max_memory_per_manifest", "32Gi")
281 )
282 self.max_gpu_per_manifest = int(config_dict.get("max_gpu_per_manifest", 4))
283 # Hard-reject accelerator jobs that lack a matching node toleration.
284 # Kept in sync with queue_processor.REQUIRE_ACCELERATOR_TOLERATION.
285 self.require_accelerator_toleration = config_dict.get(
286 "require_accelerator_toleration", True
287 )
288 self.allowed_namespaces = set(config_dict.get("allowed_namespaces", ["gco-jobs"]))
289 self.validation_enabled = config_dict.get("validation_enabled", True)
291 # Trusted registries for image validation (configurable via cdk.json)
292 self.trusted_registries = config_dict.get(
293 "trusted_registries",
294 [
295 "docker.io",
296 "gcr.io",
297 "quay.io",
298 "registry.k8s.io",
299 "k8s.gcr.io",
300 "public.ecr.aws",
301 "nvcr.io",
302 ],
303 )
304 self.trusted_dockerhub_orgs = config_dict.get(
305 "trusted_dockerhub_orgs",
306 [
307 "nvidia",
308 "pytorch",
309 "rayproject",
310 "tensorflow",
311 "huggingface",
312 "amazon",
313 "bitnami",
314 "gco",
315 ],
316 )
318 # Warn about trusted_registries entries that look like Docker Hub orgs (no dot or colon)
319 for registry in self.trusted_registries:
320 if not self._is_registry_domain(registry): 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 logger.warning(
322 f"Trusted registry '{registry}' has no domain separator (dot or colon) — "
323 f"consider moving it to trusted_dockerhub_orgs instead"
324 )
326 # YAML parsing limits (configurable via cdk.json)
327 self.yaml_max_depth = int(config_dict.get("yaml_max_depth", 50))
329 # Allowed resource kinds (configurable via cdk.json)
330 self.allowed_kinds = set(config_dict.get("allowed_kinds", DEFAULT_ALLOWED_KINDS))
332 # Security policy — toggleable checks (configurable via cdk.json)
333 security_policy = config_dict.get("manifest_security_policy", {})
334 self.block_privileged = security_policy.get("block_privileged", True)
335 self.block_privilege_escalation = security_policy.get("block_privilege_escalation", True)
336 self.block_host_network = security_policy.get("block_host_network", True)
337 self.block_host_pid = security_policy.get("block_host_pid", True)
338 self.block_host_ipc = security_policy.get("block_host_ipc", True)
339 self.block_host_path = security_policy.get("block_host_path", True)
340 self.block_added_capabilities = security_policy.get("block_added_capabilities", True)
341 self.block_run_as_root = security_policy.get("block_run_as_root", False)
343 # ------------------------------------------------------------------
344 # Security defaults injection
345 # ------------------------------------------------------------------
347 @staticmethod
348 def _extract_pod_spec(manifest: dict[str, Any]) -> dict[str, Any] | None:
349 """Extract the pod spec from a manifest, handling all workload types.
351 Supports:
352 - Deployment / StatefulSet / DaemonSet / ReplicaSet → spec.template.spec
353 - Job → spec.template.spec
354 - CronJob → spec.jobTemplate.spec.template.spec
355 - Bare Pod → spec (when ``containers`` key is present)
357 Returns:
358 The pod spec dict (mutable reference), or ``None`` if the manifest
359 does not contain a recognisable pod spec.
360 """
361 spec = manifest.get("spec")
362 if spec is None or not isinstance(spec, dict):
363 return None
365 kind = manifest.get("kind", "")
367 # CronJob: spec.jobTemplate.spec.template.spec
368 if kind == "CronJob":
369 job_template = spec.get("jobTemplate")
370 if isinstance(job_template, dict): 370 ↛ 378line 370 didn't jump to line 378 because the condition on line 370 was always true
371 job_spec = job_template.get("spec")
372 if isinstance(job_spec, dict): 372 ↛ 378line 372 didn't jump to line 378 because the condition on line 372 was always true
373 template = job_spec.get("template")
374 if isinstance(template, dict): 374 ↛ 378line 374 didn't jump to line 378 because the condition on line 374 was always true
375 pod_spec = template.get("spec")
376 if isinstance(pod_spec, dict): 376 ↛ 378line 376 didn't jump to line 378 because the condition on line 376 was always true
377 return pod_spec
378 return None
380 # Deployment / StatefulSet / DaemonSet / ReplicaSet / Job:
381 # spec.template.spec
382 if "template" in spec:
383 template = spec.get("template")
384 if isinstance(template, dict): 384 ↛ 388line 384 didn't jump to line 388 because the condition on line 384 was always true
385 pod_spec = template.get("spec")
386 if isinstance(pod_spec, dict): 386 ↛ 388line 386 didn't jump to line 388 because the condition on line 386 was always true
387 return pod_spec
388 return None
390 # Bare Pod: spec contains "containers" directly
391 if "containers" in spec:
392 return cast(dict[str, Any], spec)
394 return None
396 def _inject_security_defaults(self, manifest: dict[str, Any]) -> dict[str, Any]:
397 """Inject security defaults into user-submitted manifests.
399 Currently injects:
400 - ``automountServiceAccountToken: false`` in the pod spec (unless the
401 user has explicitly set it).
403 The method mutates *manifest* in-place and returns it for convenience.
404 """
405 pod_spec = self._extract_pod_spec(manifest)
406 if pod_spec is not None:
407 # Use setdefault so we don't override an explicit user choice
408 pod_spec.setdefault("automountServiceAccountToken", False)
409 return manifest
411 @property
412 def dynamic_client(self) -> dynamic.DynamicClient:
413 """Lazy-initialized dynamic client for CRD support."""
414 if self._dynamic_client is None:
415 self._dynamic_client = dynamic.DynamicClient(self.api_client)
416 return self._dynamic_client
418 def _parse_cpu_string(self, cpu_str: str) -> int:
419 """Parse CPU string to millicores"""
420 if not cpu_str:
421 return 0
423 cpu_str = cpu_str.strip()
424 if cpu_str.endswith("m"):
425 return int(cpu_str[:-1])
426 return int(cpu_str) * 1000
428 def _parse_memory_string(self, memory_str: str) -> int:
429 """Parse memory string to bytes"""
430 if not memory_str:
431 return 0
433 memory_str = memory_str.strip()
435 if memory_str.endswith("Ki"):
436 return int(memory_str[:-2]) * 1024
437 if memory_str.endswith("Mi"):
438 return int(memory_str[:-2]) * 1024 * 1024
439 if memory_str.endswith("Gi"):
440 return int(memory_str[:-2]) * 1024 * 1024 * 1024
441 if memory_str.endswith("Ti"):
442 return int(memory_str[:-2]) * 1024 * 1024 * 1024 * 1024
443 if memory_str.endswith("k"):
444 return int(memory_str[:-1]) * 1000
445 if memory_str.endswith("M"):
446 return int(memory_str[:-1]) * 1000 * 1000
447 if memory_str.endswith("G"):
448 return int(memory_str[:-1]) * 1000 * 1000 * 1000
449 return int(memory_str)
451 def _check_yaml_depth(self, obj: Any, current_depth: int = 0) -> bool:
452 """Check if a parsed YAML/JSON object exceeds max nesting depth.
454 Recursively walks dicts and lists. Returns False if depth exceeds
455 ``self.yaml_max_depth``.
457 Args:
458 obj: The parsed object to check (dict, list, or scalar).
459 current_depth: Current recursion depth (callers should leave at 0).
461 Returns:
462 True if the object is within the depth limit, False otherwise.
463 """
464 if current_depth > self.yaml_max_depth:
465 return False
466 if isinstance(obj, dict):
467 return all(self._check_yaml_depth(v, current_depth + 1) for v in obj.values())
468 if isinstance(obj, list):
469 return all(self._check_yaml_depth(item, current_depth + 1) for item in obj)
470 return True
472 def validate_manifest(
473 self,
474 manifest: dict[str, Any],
475 default_namespace: str | None = None,
476 ) -> tuple[bool, str | None]:
477 """Validate a Kubernetes manifest for security and resource constraints.
479 ``default_namespace`` is the request-level destination for manifests
480 that omit ``metadata.namespace``. Validation and apply must resolve the
481 same effective namespace or the request default could bypass the
482 namespace allowlist.
484 Returns: ``(is_valid, error_message)``.
485 """
486 if not self.validation_enabled:
487 return True, None
489 try:
490 # YAML depth check — reject excessively nested documents
491 if not self._check_yaml_depth(manifest):
492 return (
493 False,
494 f"Manifest exceeds maximum nesting depth of {self.yaml_max_depth} levels",
495 )
497 # Basic structure validation
498 required_fields = ["apiVersion", "kind", "metadata"]
499 for field in required_fields:
500 if field not in manifest:
501 return False, f"Missing required field: {field}"
503 # Validate metadata
504 metadata = manifest.get("metadata", {})
505 if "name" not in metadata:
506 return False, "Missing metadata.name field"
508 # Validate namespace
509 namespace = metadata.get("namespace", default_namespace or "gco-jobs")
510 if namespace not in self.allowed_namespaces:
511 return (
512 False,
513 f"Namespace '{namespace}' not allowed. Allowed namespaces: {list(self.allowed_namespaces)}",
514 )
516 # Validate resource kind using the policy shared with the SQS path.
517 kind = manifest.get("kind", "")
518 kind_valid, kind_error = validate_resource_kind(manifest, self.allowed_kinds)
519 if not kind_valid:
520 return False, kind_error
522 # Validate resource limits for workload resources
523 if kind in [
524 "Deployment",
525 "Job",
526 "CronJob",
527 "StatefulSet",
528 "DaemonSet",
529 ]:
530 resource_valid, resource_error = self._validate_resource_limits(manifest)
531 if not resource_valid:
532 return False, resource_error
534 # Require accelerator jobs to carry a matching toleration.
535 if self.require_accelerator_toleration: 535 ↛ 541line 535 didn't jump to line 541 because the condition on line 535 was always true
536 tol_valid, tol_error = self._validate_tolerations(manifest)
537 if not tol_valid:
538 return False, tol_error
540 # Security validations
541 sec_valid, sec_error = self._validate_security_context(manifest)
542 if not sec_valid:
543 return False, f"Security context validation failed: {sec_error}"
545 # Validate image sources (prevent pulling from untrusted registries)
546 img_valid, img_error = self._validate_image_sources(manifest)
547 if not img_valid:
548 return False, img_error or "Untrusted image sources detected"
550 return True, None
552 except Exception as e:
553 logger.error(f"Error validating manifest: {e}")
554 return False, f"Validation error: {e!s}"
556 def _validate_resource_limits(self, manifest: dict[str, Any]) -> tuple[bool, str]:
557 """Validate resource limits in manifest.
559 Returns:
560 Tuple of (is_valid, error_message). error_message is empty if valid.
561 """
562 try:
563 errors: list[str] = []
564 spec = manifest.get("spec", {})
566 # Get pod spec (handle different resource types)
567 pod_spec = {}
568 if "template" in spec: # Deployment, StatefulSet, etc.
569 pod_spec = spec.get("template", {}).get("spec", {})
570 elif "jobTemplate" in spec: # CronJob 570 ↛ 574line 570 didn't jump to line 574 because the condition on line 570 was always true
571 pod_spec = (
572 spec.get("jobTemplate", {}).get("spec", {}).get("template", {}).get("spec", {})
573 )
574 elif "containers" in spec: # Pod
575 pod_spec = spec
577 total_cpu = 0
578 total_memory = 0
579 total_gpu = 0
581 for _container_type, container in self._get_all_containers(pod_spec):
582 resources = container.get("resources", {})
583 requests = resources.get("requests", {})
584 limits = resources.get("limits", {})
586 # Check CPU (use limits if available, otherwise requests)
587 cpu = limits.get("cpu") or requests.get( # nosec B113 - dict.get(), not HTTP requests
588 "cpu", "0"
589 )
590 total_cpu += self._parse_cpu_string(cpu)
592 # Check Memory
593 memory = limits.get("memory") or requests.get( # nosec B113 - dict.get(), not HTTP requests
594 "memory", "0"
595 )
596 total_memory += self._parse_memory_string(memory)
598 # Check GPU
599 gpu = limits.get("nvidia.com/gpu") or requests.get( # nosec B113 - dict.get(), not HTTP requests
600 "nvidia.com/gpu", "0"
601 )
602 total_gpu += int(gpu)
604 # Validate against limits
605 if total_cpu > self.max_cpu_per_manifest:
606 logger.warning(f"CPU limit exceeded: {total_cpu}m > {self.max_cpu_per_manifest}m")
607 errors.append(f"CPU {total_cpu}m exceeds max {self.max_cpu_per_manifest}m")
609 if total_memory > self.max_memory_per_manifest:
610 logger.warning(
611 f"Memory limit exceeded: {total_memory} > {self.max_memory_per_manifest}"
612 )
613 mem_gb = self.max_memory_per_manifest / (1024**3)
614 req_gb = total_memory / (1024**3)
615 errors.append(f"Memory {req_gb:.0f}Gi exceeds max {mem_gb:.0f}Gi")
617 if total_gpu > self.max_gpu_per_manifest:
618 logger.warning(f"GPU limit exceeded: {total_gpu} > {self.max_gpu_per_manifest}")
619 errors.append(f"GPU {total_gpu} exceeds max {self.max_gpu_per_manifest}")
621 if errors:
622 hint = (
623 "To raise limits, update resource_quotas in cdk.json "
624 "and redeploy (see examples/README.md#troubleshooting)"
625 )
626 return False, "; ".join(errors) + f". {hint}"
628 return True, ""
630 except Exception as e:
631 logger.error(f"Error validating resource limits: {e}")
632 return False, f"Resource limit validation error: {e}"
634 def _validate_tolerations(self, manifest: dict[str, Any]) -> tuple[bool, str | None]:
635 """Require accelerator jobs to carry a matching node toleration.
637 GCO nodepools taint accelerator nodes with ``nvidia.com/gpu``,
638 ``aws.amazon.com/neuron``, and ``vpc.amazonaws.com/efa`` (NoSchedule).
639 A pod requesting one of these resources but lacking a matching
640 toleration would stay Pending forever, so we reject it at admission
641 with an actionable message instead.
643 Returns:
644 Tuple of (is_valid, error_message). error_message is None if valid.
645 """
646 pod_spec = self._extract_pod_spec(manifest)
647 if not pod_spec:
648 return True, None
650 requested = self._requested_accelerators(pod_spec)
651 if not requested:
652 return True, None
654 tolerations = pod_spec.get("tolerations", []) or []
655 for taint in requested:
656 if not _toleration_matches(tolerations, taint):
657 hint = (
658 f"add a matching toleration (e.g. key '{taint}', operator "
659 "'Exists', effect 'NoSchedule'); see examples/gpu-job.yaml"
660 )
661 return (
662 False,
663 f"Job requests accelerator '{taint}' but no matching "
664 f"toleration for taint {taint}=true:NoSchedule was found. {hint}",
665 )
666 return True, None
668 def _requested_accelerators(self, pod_spec: dict[str, Any]) -> set[str]:
669 """Return the set of accelerator taint keys any container requests
670 a nonzero quantity of."""
671 requested: set[str] = set()
672 for _container_type, container in self._get_all_containers(pod_spec):
673 resources = container.get("resources", {}) or {}
674 for section in ("requests", "limits"):
675 values = resources.get(section, {}) or {}
676 for taint in ACCELERATOR_TAINTS:
677 if _positive_quantity(values.get(taint)):
678 requested.add(taint)
679 return requested
681 def _get_all_containers(self, pod_spec: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
682 """Get all containers from pod spec including init and ephemeral containers.
684 Returns:
685 List of (container_type, container_dict) tuples where container_type
686 is one of 'container', 'initContainer', or 'ephemeralContainer'.
687 """
688 result = []
689 for c in pod_spec.get("containers", []):
690 result.append(("container", c))
691 for c in pod_spec.get("initContainers", []):
692 result.append(("initContainer", c))
693 for c in pod_spec.get("ephemeralContainers", []):
694 result.append(("ephemeralContainer", c))
695 return result
697 def _validate_security_context(self, manifest: dict[str, Any]) -> tuple[bool, str | None]:
698 """Validate security context settings.
700 Returns:
701 Tuple of (is_valid, error_message). error_message is None if valid.
702 """
703 try:
704 # Basic security checks - prevent privileged containers
705 spec = manifest.get("spec", {})
707 # Get pod spec (handle different resource types)
708 pod_spec = None
709 if "template" in spec:
710 pod_spec = spec.get("template", {}).get("spec", {})
711 elif "jobTemplate" in spec:
712 pod_spec = (
713 spec.get("jobTemplate", {}).get("spec", {}).get("template", {}).get("spec", {})
714 )
715 elif "containers" in spec:
716 pod_spec = spec
718 if pod_spec:
719 # --- Pod-level checks ---
720 if self.block_host_network and pod_spec.get("hostNetwork", False):
721 return False, "hostNetwork is not permitted"
723 if self.block_host_pid and pod_spec.get("hostPID", False):
724 return False, "hostPID is not permitted"
726 if self.block_host_ipc and pod_spec.get("hostIPC", False):
727 return False, "hostIPC is not permitted"
729 # Check volumes for hostPath
730 if self.block_host_path:
731 for volume in pod_spec.get("volumes", []):
732 if volume.get("hostPath") is not None:
733 return False, "hostPath volumes are not permitted"
735 # Check pod security context
736 security_context = pod_spec.get("securityContext", {})
737 if self.block_privileged and security_context.get("privileged", False):
738 return False, "privileged pod security context is not permitted"
740 if self.block_run_as_root:
741 run_as_user = security_context.get("runAsUser")
742 if run_as_user is not None and run_as_user == 0:
743 return False, "running as root (runAsUser: 0) is not permitted"
745 # --- Container-level checks ---
746 for container_type, container in self._get_all_containers(pod_spec):
747 container_name = container.get("name", "unknown")
748 container_security = container.get("securityContext", {})
749 if self.block_privileged and container_security.get("privileged", False):
750 return (
751 False,
752 f"{container_type} '{container_name}': privileged containers are not permitted",
753 )
754 if self.block_privilege_escalation and container_security.get(
755 "allowPrivilegeEscalation", False
756 ):
757 return (
758 False,
759 f"{container_type} '{container_name}': allowPrivilegeEscalation is not permitted",
760 )
762 # Check for added capabilities
763 if self.block_added_capabilities:
764 added_caps = container_security.get("capabilities", {}).get("add", [])
765 if added_caps:
766 return (
767 False,
768 f"{container_type} '{container_name}': added capabilities are not permitted",
769 )
771 # Check for runAsUser: 0 (root) — off by default
772 if self.block_run_as_root:
773 run_as_user = container_security.get("runAsUser")
774 if run_as_user is not None and run_as_user == 0:
775 return (
776 False,
777 f"{container_type} '{container_name}': running as root (runAsUser: 0) is not permitted",
778 )
780 return True, None
782 except Exception as e:
783 logger.error(f"Error validating security context: {e}")
784 return False, f"Security context error: {e}"
786 @staticmethod
787 def _is_registry_domain(entry: str) -> bool:
788 """Check if a registry entry is a proper domain (contains dot or colon).
790 A proper registry domain contains either a dot (e.g., 'docker.io', 'gcr.io')
791 or a colon (e.g., 'localhost:5000'). Entries without these are Docker Hub
792 organization names (e.g., 'nvidia', 'gco').
793 """
794 return "." in entry or ":" in entry
796 def _validate_image_sources(self, manifest: dict[str, Any]) -> tuple[bool, str | None]:
797 """Validate container image sources.
799 Uses proper domain matching instead of prefix matching to prevent
800 dependency confusion attacks (e.g., 'gco-malicious/evil' should NOT
801 match a trusted registry entry 'gco').
803 Matching logic:
804 1. If image has no '/' → official Docker Hub image (always allowed)
805 2. If image has '/' and part before first '/' contains a dot or colon
806 → it's a registry domain → match against trusted_registries
807 3. If image has '/' but first segment has no dot/colon
808 → it's a Docker Hub org → match against trusted_dockerhub_orgs
809 4. Digest references (@sha256:) are accepted from any trusted source
811 Returns:
812 Tuple of (is_valid, error_message). error_message is None if valid.
813 """
814 try:
815 trusted_registries = self.trusted_registries
816 trusted_dockerhub_orgs = self.trusted_dockerhub_orgs
818 spec = manifest.get("spec", {})
820 # Get pod spec (handle different resource types)
821 pod_spec = {}
822 if "template" in spec:
823 pod_spec = spec.get("template", {}).get("spec", {})
824 elif "jobTemplate" in spec:
825 pod_spec = (
826 spec.get("jobTemplate", {}).get("spec", {}).get("template", {}).get("spec", {})
827 )
828 elif "containers" in spec:
829 pod_spec = spec
831 for container_type, container in self._get_all_containers(pod_spec):
832 image = container.get("image", "")
833 if not image:
834 continue
836 is_trusted = False
838 # Case 1: Official Docker Hub image (no slash) — e.g., "python:3.14", "busybox"
839 if "/" not in image:
840 is_trusted = True
841 else:
842 # Image has a slash — determine if first segment is a domain or org
843 first_segment = image.split("/")[0]
845 if self._is_registry_domain(first_segment):
846 # Case 2: First segment looks like a domain (has dot or colon)
847 # Match against trusted_registries as exact domain match
848 for registry in trusted_registries:
849 if first_segment == registry:
850 is_trusted = True
851 break
852 # Also support multi-level registry paths like "public.ecr.aws"
853 # where the image might be "public.ecr.aws/lambda/python:3.14"
854 if image.startswith(registry + "/"): 854 ↛ 855line 854 didn't jump to line 855 because the condition on line 854 was never true
855 is_trusted = True
856 break
857 else:
858 # Case 3: First segment has no dot/colon — it's a Docker Hub org
859 # Match against trusted_dockerhub_orgs
860 if first_segment in trusted_dockerhub_orgs:
861 is_trusted = True
863 if not is_trusted:
864 container_name = container.get("name", "unknown")
865 # image comes from the user-submitted manifest; sanitize it
866 # before logging to prevent log injection / forging (CWE-117).
867 logger.warning("Untrusted image source: %s", sanitize_log_value(image))
868 return (
869 False,
870 f"{container_type} '{container_name}': Untrusted image source '{image}'",
871 )
873 return True, None
875 except Exception as e:
876 logger.error(f"Error validating image sources: {e}")
877 return False, f"Image source validation error: {e}"
879 async def process_manifest_submission(
880 self, request: ManifestSubmissionRequest
881 ) -> ManifestSubmissionResponse:
882 """
883 Process a manifest submission request
884 """
885 logger.info(f"Processing manifest submission with {len(request.manifests)} manifests")
887 resources = []
888 errors = []
889 overall_success = True
891 try:
892 # Process each manifest
893 for i, manifest_data in enumerate(request.manifests):
894 try:
895 # Validate manifest
896 is_valid, error_msg = self.validate_manifest(manifest_data, request.namespace)
897 if not is_valid:
898 error_msg = f"Manifest {i + 1} validation failed: {error_msg}"
899 errors.append(error_msg)
900 logger.error(error_msg)
902 # Create failed resource status
903 resource_status = ResourceStatus(
904 api_version=manifest_data.get("apiVersion", "unknown"),
905 kind=manifest_data.get("kind", "unknown"),
906 name=manifest_data.get("metadata", {}).get("name", f"manifest-{i + 1}"),
907 namespace=manifest_data.get("metadata", {}).get(
908 "namespace", request.namespace or "gco-jobs"
909 ),
910 status="failed",
911 message=error_msg,
912 )
913 resources.append(resource_status)
914 overall_success = False
915 continue
917 # Apply manifest if validation passed
918 if not request.dry_run:
919 resource_status = await self._apply_manifest(
920 manifest_data, request.namespace
921 )
922 resources.append(resource_status)
924 if not resource_status.is_successful():
925 overall_success = False
926 else:
927 # Dry run - just validate
928 resource_status = ResourceStatus(
929 api_version=manifest_data.get("apiVersion", "unknown"),
930 kind=manifest_data.get("kind", "unknown"),
931 name=manifest_data.get("metadata", {}).get("name", "unknown"),
932 namespace=manifest_data.get("metadata", {}).get(
933 "namespace", request.namespace or "gco-jobs"
934 ),
935 status="unchanged",
936 message="Dry run - validation passed",
937 )
938 resources.append(resource_status)
940 except Exception as e:
941 error_msg = f"Error processing manifest {i + 1}: {e!s}"
942 errors.append(error_msg)
943 logger.error(error_msg)
944 overall_success = False
946 # Create failed resource status
947 resource_status = ResourceStatus(
948 api_version=manifest_data.get("apiVersion", "unknown"),
949 kind=manifest_data.get("kind", "unknown"),
950 name=manifest_data.get("metadata", {}).get("name", f"manifest-{i + 1}"),
951 namespace=manifest_data.get("metadata", {}).get(
952 "namespace", request.namespace or "gco-jobs"
953 ),
954 status="failed",
955 message=str(e),
956 )
957 resources.append(resource_status)
959 except Exception as e:
960 error_msg = f"Fatal error processing manifest submission: {e!s}"
961 errors.append(error_msg)
962 logger.error(error_msg)
963 overall_success = False
965 response = ManifestSubmissionResponse(
966 success=overall_success,
967 cluster_id=self.cluster_id,
968 region=self.region,
969 resources=resources,
970 errors=errors if errors else None,
971 )
973 logger.info(
974 f"Manifest submission completed - Success: {overall_success}, "
975 f"Resources: {len(resources)}, Errors: {len(errors)}"
976 )
978 return response
980 @staticmethod
981 def queued_job_name(original_name: str, queue_job_id: str) -> str:
982 """Return a DNS-label-safe Kubernetes name deterministically fenced by queue ID."""
983 suffix = hashlib.sha256(queue_job_id.encode("utf-8")).hexdigest()[:16]
984 prefix = re.sub(r"[^a-z0-9-]+", "-", original_name.lower()).strip("-")
985 prefix = prefix[: 63 - len(suffix) - 1].rstrip("-") or "gco-job"
986 return f"{prefix}-{suffix}"
988 def apply_queued_job(
989 self,
990 manifest_data: dict[str, Any],
991 namespace: str,
992 queue_job_id: str,
993 ) -> ResourceStatus:
994 """Create or adopt exactly one deterministic ``batch/v1`` Job.
996 This path deliberately bypasses generic manifest upsert semantics: it
997 never deletes, renames, or replaces an existing Job. An ambiguous API
998 result is safe to retry because the same queue ID always resolves to the
999 same Kubernetes name and adoption requires the full queue ID annotation.
1000 """
1001 manifest = copy.deepcopy(manifest_data)
1002 if manifest.get("apiVersion") != "batch/v1" or manifest.get("kind") != "Job": 1002 ↛ 1003line 1002 didn't jump to line 1003 because the condition on line 1002 was never true
1003 raise QueuedJobNotCreatedError(
1004 "Central queue accepts only apiVersion 'batch/v1', kind 'Job'"
1005 )
1007 metadata = manifest.get("metadata")
1008 if not isinstance(metadata, dict): 1008 ↛ 1009line 1008 didn't jump to line 1009 because the condition on line 1008 was never true
1009 raise QueuedJobNotCreatedError("Queued Job metadata must be an object")
1010 declared_namespace = metadata.get("namespace")
1011 if declared_namespace is not None and declared_namespace != namespace: 1011 ↛ 1012line 1011 didn't jump to line 1012 because the condition on line 1011 was never true
1012 raise QueuedJobNotCreatedError("Queued Job namespace does not match the queue envelope")
1013 original_name = metadata.get("name")
1014 if not isinstance(original_name, str) or not original_name: 1014 ↛ 1015line 1014 didn't jump to line 1015 because the condition on line 1014 was never true
1015 raise QueuedJobNotCreatedError("Queued Job metadata.name is required")
1017 deterministic_name = self.queued_job_name(original_name, queue_job_id)
1018 metadata["name"] = deterministic_name
1019 metadata["namespace"] = namespace
1020 labels = metadata.setdefault("labels", {})
1021 annotations = metadata.setdefault("annotations", {})
1022 if not isinstance(labels, dict) or not isinstance(annotations, dict): 1022 ↛ 1023line 1022 didn't jump to line 1023 because the condition on line 1022 was never true
1023 raise QueuedJobNotCreatedError(
1024 "Queued Job metadata labels and annotations must be objects"
1025 )
1026 labels["gco.io/managed-by"] = "central-queue"
1027 labels["gco.io/queue-job-key"] = hashlib.sha256(queue_job_id.encode("utf-8")).hexdigest()[
1028 :32
1029 ]
1030 annotations["gco.io/queue-job-id"] = queue_job_id
1031 annotations["gco.io/original-job-name"] = original_name
1033 is_valid, validation_error = self.validate_manifest(manifest, namespace)
1034 if not is_valid:
1035 raise QueuedJobNotCreatedError(f"Queued Job validation failed: {validation_error}")
1036 self._inject_security_defaults(manifest)
1038 try:
1039 job = self.batch_v1.read_namespaced_job(
1040 name=deterministic_name,
1041 namespace=namespace,
1042 _request_timeout=self._k8s_timeout,
1043 )
1044 operation = "unchanged"
1045 message = "Existing deterministic Kubernetes Job adopted"
1046 except ApiException as error:
1047 if error.status != 404: 1047 ↛ 1048line 1047 didn't jump to line 1048 because the condition on line 1047 was never true
1048 if _is_retryable_kubernetes_api_error(error):
1049 raise RetryableQueuedJobApplyError(
1050 "Kubernetes Job lookup was inconclusive; retry deterministic adoption"
1051 ) from error
1052 raise
1053 try:
1054 job = self.batch_v1.create_namespaced_job(
1055 namespace=namespace,
1056 body=manifest,
1057 _request_timeout=self._k8s_timeout,
1058 )
1059 operation = "created"
1060 message = "Deterministic Kubernetes Job created"
1061 except ApiException as create_error:
1062 if create_error.status == 409:
1063 try:
1064 job = self.batch_v1.read_namespaced_job(
1065 name=deterministic_name,
1066 namespace=namespace,
1067 _request_timeout=self._k8s_timeout,
1068 )
1069 except ApiException as adoption_error:
1070 if adoption_error.status == 404 or _is_retryable_kubernetes_api_error(
1071 adoption_error
1072 ):
1073 raise RetryableQueuedJobApplyError(
1074 "Concurrent Kubernetes Job adoption was inconclusive"
1075 ) from adoption_error
1076 raise
1077 except Exception as adoption_error:
1078 raise RetryableQueuedJobApplyError(
1079 "Concurrent Kubernetes Job adoption was inconclusive"
1080 ) from adoption_error
1081 operation = "unchanged"
1082 message = "Concurrent deterministic Kubernetes Job adopted"
1083 elif _is_retryable_kubernetes_api_error(create_error):
1084 raise RetryableQueuedJobApplyError(
1085 "Kubernetes Job create result was inconclusive; retry deterministic adoption"
1086 ) from create_error
1087 else:
1088 raise
1089 except Exception as create_error:
1090 # A timeout or connection loss can occur after the API server
1091 # persisted the Job. Never mark that ambiguous result terminal.
1092 raise RetryableQueuedJobApplyError(
1093 "Kubernetes Job create result was inconclusive; retry deterministic adoption"
1094 ) from create_error
1095 except Exception as read_error:
1096 raise RetryableQueuedJobApplyError(
1097 "Kubernetes Job lookup was inconclusive; retry deterministic adoption"
1098 ) from read_error
1100 actual_annotations = getattr(job.metadata, "annotations", None) or {}
1101 if actual_annotations.get("gco.io/queue-job-id") != queue_job_id: 1101 ↛ 1102line 1101 didn't jump to line 1102 because the condition on line 1101 was never true
1102 raise RuntimeError(
1103 f"Kubernetes Job name collision for {namespace}/{deterministic_name}"
1104 )
1105 uid = str(getattr(job.metadata, "uid", "") or "")
1106 actual_name = str(getattr(job.metadata, "name", "") or deterministic_name)
1107 actual_namespace = str(getattr(job.metadata, "namespace", "") or namespace)
1108 if not uid: 1108 ↛ 1109line 1108 didn't jump to line 1109 because the condition on line 1108 was never true
1109 raise RuntimeError("Kubernetes API returned a queued Job without a UID")
1110 return ResourceStatus(
1111 api_version="batch/v1",
1112 kind="Job",
1113 name=actual_name,
1114 namespace=actual_namespace,
1115 status=operation,
1116 message=message,
1117 uid=uid,
1118 )
1120 def read_queued_job(self, name: str, namespace: str) -> V1Job:
1121 """Read one reconciled Job through the processor's bounded client contract."""
1122 return self.batch_v1.read_namespaced_job(
1123 name=name,
1124 namespace=namespace,
1125 _request_timeout=self._k8s_timeout,
1126 )
1128 async def _apply_manifest(
1129 self, manifest_data: dict[str, Any], default_namespace: str | None = None
1130 ) -> ResourceStatus:
1131 """
1132 Apply a single manifest to the cluster.
1134 For Jobs and CronJobs, if the resource already exists and is completed/failed,
1135 it will be automatically deleted and recreated (since these resources are immutable).
1136 """
1137 try:
1138 api_version: str = manifest_data.get("apiVersion", "unknown")
1139 kind: str = manifest_data.get("kind", "unknown")
1140 metadata = manifest_data.get("metadata", {})
1141 name: str = metadata.get("name", "unknown")
1142 namespace: str = metadata.get("namespace", default_namespace or "gco-jobs")
1144 # Ensure namespace is set in manifest
1145 if "namespace" not in metadata and namespace:
1146 manifest_data["metadata"]["namespace"] = namespace
1148 # Inject security defaults (e.g., automountServiceAccountToken: false)
1149 self._inject_security_defaults(manifest_data)
1151 # Check if resource already exists
1152 existing_resource = await self._get_existing_resource(
1153 api_version, kind, name, namespace
1154 )
1156 if existing_resource:
1157 # Jobs are immutable — if one already exists and is finished,
1158 # delete it first so we can recreate cleanly.
1159 # If the job is still active, auto-rename to avoid collision.
1160 if kind == "Job":
1161 if self._is_job_finished(existing_resource):
1162 logger.info(
1163 f"Job {name} already exists and is finished, deleting before recreating"
1164 )
1165 await self.delete_resource(api_version, kind, name, namespace)
1166 import asyncio
1168 await asyncio.sleep(1)
1169 await self._create_resource(manifest_data)
1170 status = "created"
1171 message = "Previous completed job replaced with new submission"
1172 else:
1173 # Active job — rename to avoid destroying it
1174 import uuid
1176 suffix = uuid.uuid4().hex[:5]
1177 new_name = f"{name}-{suffix}"
1178 manifest_data["metadata"]["name"] = new_name
1179 logger.warning(
1180 f"Job {name} is still active, renamed new submission to {new_name}"
1181 )
1182 await self._create_resource(manifest_data)
1183 status = "created"
1184 message = (
1185 f"Job '{name}' is still running. "
1186 f"New submission renamed to '{new_name}'."
1187 )
1188 name = new_name
1189 else:
1190 # Update existing resource (works for mutable resources)
1191 updated_resource = await self._update_resource(manifest_data)
1192 status = "updated" if updated_resource else "unchanged"
1193 message = (
1194 "Resource updated successfully" if updated_resource else "No changes needed"
1195 )
1196 else:
1197 # Create new resource
1198 await self._create_resource(manifest_data)
1199 status = "created"
1200 message = "Resource created successfully"
1202 return ResourceStatus(
1203 api_version=api_version,
1204 kind=kind,
1205 name=name,
1206 namespace=namespace,
1207 status=status,
1208 message=message,
1209 )
1211 except ApiException as e:
1212 logger.error(f"Kubernetes API error applying manifest: {e}")
1213 return ResourceStatus(
1214 api_version=manifest_data.get("apiVersion", "unknown"),
1215 kind=manifest_data.get("kind", "unknown"),
1216 name=manifest_data.get("metadata", {}).get("name", "unknown"),
1217 namespace=manifest_data.get("metadata", {}).get("namespace", "gco-jobs"),
1218 status="failed",
1219 message=f"API error: {e.reason}",
1220 )
1221 except Exception as e:
1222 logger.error(f"Error applying manifest: {e}")
1223 return ResourceStatus(
1224 api_version=manifest_data.get("apiVersion", "unknown"),
1225 kind=manifest_data.get("kind", "unknown"),
1226 name=manifest_data.get("metadata", {}).get("name", "unknown"),
1227 namespace=manifest_data.get("metadata", {}).get("namespace", "gco-jobs"),
1228 status="failed",
1229 message=str(e),
1230 )
1232 def _is_job_finished(self, job_resource: dict[str, Any]) -> bool:
1233 """Check if a Kubernetes Job resource is in a terminal state (Complete or Failed)."""
1234 status = job_resource.get("status", {})
1235 conditions = status.get("conditions") or []
1236 for condition in conditions:
1237 cond_type = condition.get("type", "")
1238 cond_status = condition.get("status", "")
1239 if cond_type in ("Complete", "Failed") and cond_status == "True":
1240 return True
1241 return False
1243 async def _get_existing_resource(
1244 self, api_version: str, kind: str, name: str, namespace: str
1245 ) -> dict[str, Any] | None:
1246 """Check if a resource already exists using dynamic client"""
1247 try:
1248 # Get the API resource
1249 api_resource = self._get_api_resource(api_version, kind)
1251 # Try to get the resource
1252 if namespace and api_resource.namespaced:
1253 resource = api_resource.get(name=name, namespace=namespace)
1254 else:
1255 resource = api_resource.get(name=name)
1257 if resource is not None: 1257 ↛ 1268line 1257 didn't jump to line 1268 because the condition on line 1257 was always true
1258 return dict(resource.to_dict())
1260 except ApiException as e:
1261 if e.status == 404:
1262 return None # Resource doesn't exist
1263 raise
1264 except ValueError:
1265 # Unknown resource type
1266 return None
1268 return None
1270 def _get_api_resource(self, api_version: str, kind: str) -> Any:
1271 """Get the API resource for a given apiVersion and kind using dynamic client."""
1272 try:
1273 return self.dynamic_client.resources.get(api_version=api_version, kind=kind)
1274 except ResourceNotFoundError as e:
1275 logger.error(
1276 "Resource type not found: %s/%s",
1277 sanitize_log_value(api_version),
1278 sanitize_log_value(kind),
1279 )
1280 raise ValueError(f"Unknown resource type: {api_version}/{kind}") from e
1282 async def _create_resource(self, manifest_data: dict[str, Any]) -> Any:
1283 """Create a resource and return the API object, including server identity."""
1284 try:
1285 api_version = manifest_data.get("apiVersion", "")
1286 kind = manifest_data.get("kind", "")
1287 namespace = manifest_data.get("metadata", {}).get("namespace")
1289 api_resource = self._get_api_resource(api_version, kind)
1290 if namespace and api_resource.namespaced:
1291 return api_resource.create(body=manifest_data, namespace=namespace)
1292 return api_resource.create(body=manifest_data)
1293 except Exception as e:
1294 logger.error(f"Error creating resource: {e}")
1295 raise
1297 async def _update_resource(self, manifest_data: dict[str, Any]) -> bool:
1298 """Update an existing resource using dynamic client"""
1299 try:
1300 api_version = manifest_data.get("apiVersion", "")
1301 kind = manifest_data.get("kind", "")
1302 name = manifest_data.get("metadata", {}).get("name", "")
1303 namespace = manifest_data.get("metadata", {}).get("namespace")
1305 # Get the API resource
1306 api_resource = self._get_api_resource(api_version, kind)
1308 # Update the resource using patch (server-side apply)
1309 if namespace and api_resource.namespaced:
1310 api_resource.patch(
1311 body=manifest_data,
1312 name=name,
1313 namespace=namespace,
1314 content_type="application/merge-patch+json",
1315 )
1316 else:
1317 api_resource.patch(
1318 body=manifest_data,
1319 name=name,
1320 content_type="application/merge-patch+json",
1321 )
1323 return True
1324 except Exception as e:
1325 logger.error(f"Error updating resource: {e}")
1326 raise
1328 async def delete_resource(
1329 self, api_version: str, kind: str, name: str, namespace: str
1330 ) -> ResourceStatus:
1331 """
1332 Delete a resource from the cluster using dynamic client
1333 """
1334 try:
1335 # Get the API resource
1336 api_resource = self._get_api_resource(api_version, kind)
1338 # Delete the resource
1339 if namespace and api_resource.namespaced:
1340 api_resource.delete(name=name, namespace=namespace)
1341 else:
1342 api_resource.delete(name=name)
1344 return ResourceStatus(
1345 api_version=api_version,
1346 kind=kind,
1347 name=name,
1348 namespace=namespace,
1349 status="deleted",
1350 message="Resource deleted successfully",
1351 )
1353 except ValueError as e:
1354 # Unknown resource type
1355 return ResourceStatus(
1356 api_version=api_version,
1357 kind=kind,
1358 name=name,
1359 namespace=namespace,
1360 status="failed",
1361 message=str(e),
1362 )
1363 except ApiException as e:
1364 if e.status == 404:
1365 return ResourceStatus(
1366 api_version=api_version,
1367 kind=kind,
1368 name=name,
1369 namespace=namespace,
1370 status="unchanged",
1371 message="Resource not found (already deleted)",
1372 )
1373 return ResourceStatus(
1374 api_version=api_version,
1375 kind=kind,
1376 name=name,
1377 namespace=namespace,
1378 status="failed",
1379 message=f"Delete failed: {e.reason}",
1380 )
1381 except Exception as e:
1382 return ResourceStatus(
1383 api_version=api_version,
1384 kind=kind,
1385 name=name,
1386 namespace=namespace,
1387 status="failed",
1388 message=str(e),
1389 )
1391 async def list_jobs(
1392 self, namespace: str | None = None, status_filter: str | None = None
1393 ) -> list[dict[str, Any]]:
1394 """
1395 List Kubernetes Jobs from allowed namespaces.
1397 Args:
1398 namespace: Filter by specific namespace (must be in allowed_namespaces)
1399 status_filter: Filter by status: "running", "completed", "failed"
1401 Returns:
1402 List of job dictionaries with metadata and status
1403 """
1404 jobs = []
1406 # Determine which namespaces to query
1407 if namespace:
1408 if namespace not in self.allowed_namespaces:
1409 raise ValueError(
1410 f"Namespace '{namespace}' not allowed. "
1411 f"Allowed namespaces: {list(self.allowed_namespaces)}"
1412 )
1413 namespaces_to_query = [namespace]
1414 else:
1415 namespaces_to_query = list(self.allowed_namespaces)
1417 for ns in namespaces_to_query:
1418 try:
1419 job_list = self.batch_v1.list_namespaced_job(
1420 namespace=ns, _request_timeout=self._k8s_timeout
1421 )
1422 for job in job_list.items:
1423 job_dict = self._job_to_dict(job)
1425 # Apply status filter
1426 if status_filter:
1427 job_status = self._get_job_status(job)
1428 if job_status != status_filter:
1429 continue
1431 jobs.append(job_dict)
1432 except ApiException as e:
1433 logger.warning(
1434 "Failed to list jobs in namespace %s: %s", sanitize_log_value(ns), e.reason
1435 )
1436 continue
1438 return jobs
1440 def _job_to_dict(self, job: V1Job) -> dict[str, Any]:
1441 """Convert a Kubernetes Job object to a dictionary."""
1442 metadata = job.metadata
1443 status = job.status
1444 spec = job.spec
1446 return {
1447 "metadata": {
1448 "name": metadata.name,
1449 "namespace": metadata.namespace,
1450 "creationTimestamp": (
1451 metadata.creation_timestamp.isoformat() if metadata.creation_timestamp else None
1452 ),
1453 "labels": metadata.labels or {},
1454 "uid": metadata.uid,
1455 },
1456 "spec": {
1457 "parallelism": spec.parallelism,
1458 "completions": spec.completions,
1459 "backoffLimit": spec.backoff_limit,
1460 },
1461 "status": {
1462 "active": status.active or 0,
1463 "succeeded": status.succeeded or 0,
1464 "failed": status.failed or 0,
1465 "startTime": status.start_time.isoformat() if status.start_time else None,
1466 "completionTime": (
1467 status.completion_time.isoformat() if status.completion_time else None
1468 ),
1469 "conditions": [
1470 {
1471 "type": c.type,
1472 "status": c.status,
1473 "reason": c.reason,
1474 "message": c.message,
1475 }
1476 for c in (status.conditions or [])
1477 ],
1478 },
1479 }
1481 def _get_job_status(self, job: V1Job) -> str:
1482 """Determine the status of a job: running, completed, or failed."""
1483 status = job.status
1484 conditions = status.conditions or []
1486 for condition in conditions:
1487 if condition.type == "Complete" and condition.status == "True":
1488 return "completed"
1489 if condition.type == "Failed" and condition.status == "True": 1489 ↛ 1486line 1489 didn't jump to line 1486 because the condition on line 1489 was always true
1490 return "failed"
1492 if (status.active or 0) > 0:
1493 return "running"
1495 return "pending"
1497 async def get_resource_status(
1498 self, api_version: str, kind: str, name: str, namespace: str
1499 ) -> dict[str, Any] | None:
1500 """
1501 Get the status of a specific resource
1502 """
1503 try:
1504 resource = await self._get_existing_resource(api_version, kind, name, namespace)
1505 if resource:
1506 return {
1507 "api_version": api_version,
1508 "kind": kind,
1509 "name": name,
1510 "namespace": namespace,
1511 "exists": True,
1512 "status": resource.get("status", {}),
1513 "metadata": resource.get("metadata", {}),
1514 "spec": resource.get("spec", {}),
1515 }
1516 return {
1517 "api_version": api_version,
1518 "kind": kind,
1519 "name": name,
1520 "namespace": namespace,
1521 "exists": False,
1522 }
1523 except Exception as e:
1524 logger.error(f"Error getting resource status: {e}")
1525 return None
1528def create_manifest_processor_from_env() -> ManifestProcessor:
1529 """
1530 Create ManifestProcessor instance from environment variables
1531 """
1532 cluster_id = os.getenv("CLUSTER_NAME", "unknown-cluster")
1533 region = os.getenv("REGION", "unknown-region")
1535 # Enable structured JSON logging for CloudWatch Insights
1536 configure_structured_logging(
1537 service_name="manifest-processor",
1538 cluster_id=cluster_id,
1539 region=region,
1540 )
1542 # Load configuration from environment
1543 config_dict = {
1544 "max_cpu_per_manifest": os.getenv("MAX_CPU_PER_MANIFEST", "10"),
1545 "max_memory_per_manifest": os.getenv("MAX_MEMORY_PER_MANIFEST", "32Gi"),
1546 "max_gpu_per_manifest": int(os.getenv("MAX_GPU_PER_MANIFEST", "4")),
1547 "require_accelerator_toleration": os.getenv(
1548 "REQUIRE_ACCELERATOR_TOLERATION", "true"
1549 ).lower()
1550 == "true",
1551 "allowed_namespaces": (
1552 ["gco-jobs"]
1553 if os.getenv("ALLOWED_NAMESPACES") is None
1554 else [
1555 namespace.strip()
1556 for namespace in os.environ["ALLOWED_NAMESPACES"].split(",")
1557 if namespace.strip()
1558 ]
1559 ),
1560 "validation_enabled": os.getenv("VALIDATION_ENABLED", "true").lower() == "true",
1561 }
1563 allowed_kinds_env = os.getenv("ALLOWED_KINDS")
1564 if allowed_kinds_env is not None:
1565 # An absent variable uses the authoritative defaults; an explicitly
1566 # empty value is a deliberate deny-all policy and must stay empty.
1567 config_dict["allowed_kinds"] = [
1568 kind.strip() for kind in allowed_kinds_env.split(",") if kind.strip()
1569 ]
1571 # Image registry allowlist — sourced from the same CDK env vars the
1572 # queue_processor reads, so an attacker who holds sqs:SendMessage on
1573 # the regional queue can't reach an image source the REST path
1574 # rejects. When unset (or empty) the ManifestProcessor falls back
1575 # to its hardcoded default. Empty/missing values are dropped to
1576 # match the queue_processor's parsing rules.
1577 trusted_registries_env = os.getenv("TRUSTED_REGISTRIES", "")
1578 trusted_registries = [r.strip() for r in trusted_registries_env.split(",") if r.strip()]
1579 if trusted_registries: 1579 ↛ 1580line 1579 didn't jump to line 1580 because the condition on line 1579 was never true
1580 config_dict["trusted_registries"] = trusted_registries
1582 trusted_dockerhub_orgs_env = os.getenv("TRUSTED_DOCKERHUB_ORGS", "")
1583 trusted_dockerhub_orgs = [o.strip() for o in trusted_dockerhub_orgs_env.split(",") if o.strip()]
1584 if trusted_dockerhub_orgs: 1584 ↛ 1585line 1584 didn't jump to line 1585 because the condition on line 1584 was never true
1585 config_dict["trusted_dockerhub_orgs"] = trusted_dockerhub_orgs
1587 return ManifestProcessor(cluster_id, region, config_dict)