Coverage for gco / services / manifest_processor.py: 100.00%
562 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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 per manifest (default: 384 cores — see
25 gco.resource_governance.DEFAULT_MANIFEST_RESOURCE_CAPS)
26 MAX_MEMORY_PER_MANIFEST: Maximum memory per manifest (default: 4096Gi)
27 MAX_GPU_PER_MANIFEST: Maximum GPUs per manifest (default: 16)
28 ALLOWED_NAMESPACES: Comma-separated list of allowed namespaces
29 VALIDATION_ENABLED: Enable/disable validation (default: true)
31Usage:
32 processor = create_manifest_processor_from_env()
33 response = await processor.process_manifest_submission(request)
34"""
36from __future__ import annotations
38import copy
39import hashlib
40import logging
41import os
42import re
43from typing import Any, cast
45import yaml
46from kubernetes import client, config, dynamic
47from kubernetes.client.models import V1Job
48from kubernetes.client.rest import ApiException
49from kubernetes.dynamic.exceptions import ResourceNotFoundError
51# Pure admission helpers and the policy constants they read live in
52# gco.job_admission, which imports no Kubernetes client. Re-exported here
53# unchanged: these names are part of this module's import surface for the
54# queue processor, the REST API, the CLI and the offline example validator,
55# and relocating them is not meant to break any of them. The `X as X` form
56# is deliberate -- under mypy's no_implicit_reexport a plain import would
57# not be visible to those importers.
58from gco.job_admission import (
59 ACCELERATOR_TAINTS as ACCELERATOR_TAINTS,
60)
61from gco.job_admission import (
62 ADDON_KIND_HINTS as ADDON_KIND_HINTS,
63)
64from gco.job_admission import (
65 DEFAULT_ALLOWED_KINDS as DEFAULT_ALLOWED_KINDS,
66)
67from gco.job_admission import (
68 DEFAULT_TRUSTED_DOCKERHUB_ORGS as DEFAULT_TRUSTED_DOCKERHUB_ORGS,
69)
70from gco.job_admission import (
71 DEFAULT_TRUSTED_REGISTRIES as DEFAULT_TRUSTED_REGISTRIES,
72)
73from gco.job_admission import (
74 RESOURCE_API_VERSIONS as RESOURCE_API_VERSIONS,
75)
76from gco.job_admission import (
77 TRAINJOB_API_VERSION as TRAINJOB_API_VERSION,
78)
79from gco.job_admission import (
80 JobValidationPolicy as JobValidationPolicy,
81)
82from gco.job_admission import (
83 TrainJobPodSpecs as TrainJobPodSpecs,
84)
85from gco.job_admission import (
86 _collect_embedded_pod_specs as _collect_embedded_pod_specs,
87)
88from gco.job_admission import (
89 _extract_validation_pod_spec as _extract_validation_pod_spec,
90)
91from gco.job_admission import (
92 _is_trusted_registry_domain as _is_trusted_registry_domain,
93)
94from gco.job_admission import (
95 _iter_all_containers as _iter_all_containers,
96)
97from gco.job_admission import (
98 _positive_quantity as _positive_quantity,
99)
100from gco.job_admission import (
101 _toleration_matches as _toleration_matches,
102)
103from gco.job_admission import (
104 _untrusted_container_in_pod_spec as _untrusted_container_in_pod_spec,
105)
106from gco.job_admission import (
107 check_resource_caps as check_resource_caps,
108)
109from gco.job_admission import (
110 check_security_context as check_security_context,
111)
112from gco.job_admission import (
113 check_tolerations as check_tolerations,
114)
115from gco.job_admission import (
116 extract_pod_spec as extract_pod_spec,
117)
118from gco.job_admission import (
119 extract_trainjob_pod_specs as extract_trainjob_pod_specs,
120)
121from gco.job_admission import (
122 parse_cpu_millicores as parse_cpu_millicores,
123)
124from gco.job_admission import (
125 parse_memory_bytes as parse_memory_bytes,
126)
127from gco.job_admission import (
128 requested_accelerators as requested_accelerators,
129)
130from gco.job_admission import (
131 trainjob_validation_pod_specs as trainjob_validation_pod_specs,
132)
133from gco.job_admission import (
134 validate_image_sources as validate_image_sources,
135)
136from gco.job_admission import (
137 validate_resource_kind as validate_resource_kind,
138)
139from gco.job_admission import (
140 weighted_pod_specs as weighted_pod_specs,
141)
142from gco.manifest_security_policy import (
143 MANIFEST_SECURITY_POLICY_DEFAULTS,
144 parse_boolean_environment,
145 validate_manifest_security_policy,
146)
147from gco.models import (
148 ManifestSubmissionRequest,
149 ManifestSubmissionResponse,
150 ResourceStatus,
151)
152from gco.resource_governance import DEFAULT_MANIFEST_RESOURCE_CAPS
153from gco.services.structured_logging import configure_structured_logging, sanitize_log_value
155# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
156# Generated at (UTC): 2026-09-01T14:42:56Z
157# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
158# Flowchart(s) generated from this file:
159# * ``ManifestProcessor.apply_queued_job`` -> ``diagrams/code_diagrams/gco/services/manifest_processor.ManifestProcessor_apply_queued_job.html``
160# (PNG: ``diagrams/code_diagrams/gco/services/manifest_processor.ManifestProcessor_apply_queued_job.png``)
161# * ``ManifestProcessor.validate_manifest`` -> ``diagrams/code_diagrams/gco/services/manifest_processor.ManifestProcessor_validate_manifest.html``
162# (PNG: ``diagrams/code_diagrams/gco/services/manifest_processor.ManifestProcessor_validate_manifest.png``)
163# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
164# <pyflowchart-code-diagram> END
167# NOTE: No logging.basicConfig() here. This module is imported by the CLI
168# (cli/jobs.py, cli/commands/*_cmd.py) as a library for YAML loading helpers.
169# Calling basicConfig() at import time would configure the root logger with
170# INFO-level output, causing noisy botocore/urllib3 INFO messages on every
171# CLI command. Container entry points (manifest_api.py) do their own
172# basicConfig() call.
173logger = logging.getLogger(__name__)
176class RetryableQueuedJobApplyError(RuntimeError):
177 """A deterministic queued Job apply can be retried or adopted safely."""
180class QueuedJobNotCreatedError(ValueError):
181 """A queued Job was rejected before any Kubernetes operation began."""
184def _is_retryable_kubernetes_api_error(error: ApiException) -> bool:
185 """Classify throttling, server, and transport-like Kubernetes API failures."""
186 try:
187 status = int(error.status or 0)
188 except TypeError, ValueError:
189 status = 0
190 return status == 0 or status in {408, 429} or status >= 500
193# ---------------------------------------------------------------------------
194# YAML Alias Rejection Loader
195# ---------------------------------------------------------------------------
198class NoAliasSafeLoader(yaml.SafeLoader):
199 """A YAML SafeLoader that rejects anchors and aliases.
201 YAML anchors (``&anchor``) and aliases (``*anchor``) can be used to
202 construct exponentially large data structures (billion-laughs attack).
203 This loader raises an error when any alias is encountered, preventing
204 such attacks at the parsing stage.
205 """
207 def compose_node(self, parent: Any, index: Any) -> Any:
208 if self.check_event(yaml.AliasEvent):
209 event = self.get_event() # type: ignore[no-untyped-call]
210 raise yaml.composer.ComposerError(
211 None,
212 None,
213 "YAML aliases are not allowed "
214 "(security policy: yaml_allow_aliases=false), "
215 f"found alias *{event.anchor}",
216 event.start_mark,
217 )
218 return super().compose_node(parent, index)
221def safe_load_yaml(stream: str | Any, *, allow_aliases: bool = False) -> Any:
222 """Load a single YAML document with optional alias rejection.
224 Args:
225 stream: YAML string or file-like object.
226 allow_aliases: If False (default), reject YAML anchors/aliases.
228 Returns:
229 Parsed YAML document.
231 Raises:
232 yaml.YAMLError: If the document is invalid or contains aliases
233 when ``allow_aliases`` is False.
234 """
235 loader_cls = yaml.SafeLoader if allow_aliases else NoAliasSafeLoader
236 # Loader is always a SafeLoader subclass (SafeLoader or NoAliasSafeLoader),
237 # so this is equivalent to yaml.safe_load. Bandit's B506 check does not
238 # recognize the custom loader as safe.
239 return yaml.load(stream, Loader=loader_cls) # nosec B506
242def safe_load_all_yaml(stream: str | Any, *, allow_aliases: bool = False) -> list[Any]:
243 """Load all YAML documents from a stream with optional alias rejection.
245 Args:
246 stream: YAML string or file-like object.
247 allow_aliases: If False (default), reject YAML anchors/aliases.
249 Returns:
250 List of parsed YAML documents (``None`` documents are skipped).
252 Raises:
253 yaml.YAMLError: If any document is invalid or contains aliases
254 when ``allow_aliases`` is False.
255 """
256 loader_cls = yaml.SafeLoader if allow_aliases else NoAliasSafeLoader
257 # Loader is always a SafeLoader subclass, so this is equivalent to
258 # yaml.safe_load_all. Bandit's B506 check does not recognize the custom
259 # loader as safe.
260 return [
261 doc
262 for doc in yaml.load_all(stream, Loader=loader_cls)
263 if doc is not None # nosec B506
264 ]
267class ManifestProcessor:
268 """
269 Processes Kubernetes manifest submissions and applies them to the cluster
270 """
272 def __init__(self, cluster_id: str, region: str, config_dict: dict[str, Any]):
273 self.cluster_id = cluster_id
274 self.region = region
275 self.config = config_dict
277 # Initialize Kubernetes clients
278 try:
279 # Try to load in-cluster config first (when running in pod)
280 config.load_incluster_config()
281 logger.info("Loaded in-cluster Kubernetes configuration")
282 except config.ConfigException:
283 try:
284 # Fall back to local kubeconfig (for development)
285 config.load_kube_config()
286 logger.info("Loaded local Kubernetes configuration")
287 except config.ConfigException as e:
288 logger.error(f"Failed to load Kubernetes configuration: {e}")
289 raise
291 # Initialize API clients
292 self.api_client = client.ApiClient()
293 self.api_client.configuration.request_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
294 self.core_v1 = client.CoreV1Api()
295 self.apps_v1 = client.AppsV1Api()
296 self.batch_v1 = client.BatchV1Api()
297 self.networking_v1 = client.NetworkingV1Api()
298 self.custom_objects = client.CustomObjectsApi()
300 # Dynamic client for CRDs - lazy initialized to avoid cluster connection during init
301 self._dynamic_client: dynamic.DynamicClient | None = None
303 # Timeout for Kubernetes API calls (seconds)
304 self._k8s_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
306 # Resource quotas and limits. Defaults come from the shared source of
307 # truth (gco.resource_governance.DEFAULT_MANIFEST_RESOURCE_CAPS): two
308 # full accelerator-node slices, validated at synth against the
309 # LimitRange / namespace-quota layering invariant.
310 self.max_cpu_per_manifest = self._parse_cpu_string(
311 config_dict.get(
312 "max_cpu_per_manifest",
313 DEFAULT_MANIFEST_RESOURCE_CAPS["max_cpu_per_manifest"],
314 )
315 )
316 self.max_memory_per_manifest = self._parse_memory_string(
317 config_dict.get(
318 "max_memory_per_manifest",
319 DEFAULT_MANIFEST_RESOURCE_CAPS["max_memory_per_manifest"],
320 )
321 )
322 self.max_gpu_per_manifest = int(
323 config_dict.get(
324 "max_gpu_per_manifest",
325 DEFAULT_MANIFEST_RESOURCE_CAPS["max_gpu_per_manifest"],
326 )
327 )
328 # Hard-reject accelerator jobs that lack a matching node toleration.
329 # Kept in sync with queue_processor.REQUIRE_ACCELERATOR_TOLERATION.
330 require_accelerator_toleration = config_dict.get("require_accelerator_toleration", True)
331 if type(require_accelerator_toleration) is not bool:
332 raise ValueError("require_accelerator_toleration must be a boolean")
333 self.require_accelerator_toleration = require_accelerator_toleration
334 self.allowed_namespaces = set(config_dict.get("allowed_namespaces", ["gco-jobs"]))
335 validation_enabled = config_dict.get("validation_enabled", True)
336 if type(validation_enabled) is not bool:
337 raise ValueError("validation_enabled must be a boolean")
338 self.validation_enabled = validation_enabled
340 # Trusted registries for image validation (configurable via cdk.json)
341 self.trusted_registries = config_dict.get(
342 "trusted_registries", list(DEFAULT_TRUSTED_REGISTRIES)
343 )
344 self.trusted_dockerhub_orgs = config_dict.get(
345 "trusted_dockerhub_orgs", list(DEFAULT_TRUSTED_DOCKERHUB_ORGS)
346 )
348 # Warn about trusted_registries entries that look like Docker Hub orgs (no dot or colon)
349 for registry in self.trusted_registries:
350 if not self._is_registry_domain(registry):
351 logger.warning(
352 f"Trusted registry '{registry}' has no domain separator (dot or colon) — "
353 f"consider moving it to trusted_dockerhub_orgs instead"
354 )
356 # YAML parsing limits (configurable via cdk.json)
357 self.yaml_max_depth = int(config_dict.get("yaml_max_depth", 50))
359 # Allowed resource kinds (configurable via cdk.json)
360 self.allowed_kinds = set(config_dict.get("allowed_kinds", DEFAULT_ALLOWED_KINDS))
362 # Security policy — toggleable checks (configurable via cdk.json)
363 security_policy = validate_manifest_security_policy(
364 config_dict.get("manifest_security_policy", {})
365 )
366 self.block_privileged = security_policy.get("block_privileged", True)
367 self.block_privilege_escalation = security_policy.get("block_privilege_escalation", True)
368 self.block_host_network = security_policy.get("block_host_network", True)
369 self.block_host_pid = security_policy.get("block_host_pid", True)
370 self.block_host_ipc = security_policy.get("block_host_ipc", True)
371 self.block_host_path = security_policy.get("block_host_path", True)
372 self.block_added_capabilities = security_policy.get("block_added_capabilities", True)
373 self.block_run_as_root = security_policy.get("block_run_as_root", False)
375 # ------------------------------------------------------------------
376 # Effective-policy introspection (read-only)
377 # ------------------------------------------------------------------
379 def job_validation_policy(self) -> JobValidationPolicy:
380 """Bundle the attributes the pure admission checks read.
382 Built fresh per call rather than cached at ``__init__``: the caps and
383 toggles are plain attributes and a test (or a future reload path) may
384 set them after construction, and a stale snapshot here would enforce a
385 policy the instance no longer reports.
386 """
387 return JobValidationPolicy(
388 max_cpu_millicores=self.max_cpu_per_manifest,
389 max_memory_bytes=self.max_memory_per_manifest,
390 max_gpu_count=self.max_gpu_per_manifest,
391 allowed_namespaces=frozenset(self.allowed_namespaces),
392 allowed_kinds=frozenset(self.allowed_kinds),
393 # Sorted, like effective_job_validation_policy() reports them.
394 # Matching is by equality and prefix so order cannot change an
395 # admission outcome, but a canonical order is what lets two policies
396 # be compared with == -- which cross-region drift detection relies
397 # on, and which would otherwise report configuration order as drift.
398 trusted_registries=tuple(sorted(self.trusted_registries)),
399 trusted_dockerhub_orgs=tuple(sorted(self.trusted_dockerhub_orgs)),
400 require_accelerator_toleration=self.require_accelerator_toleration,
401 security={
402 "block_privileged": self.block_privileged,
403 "block_privilege_escalation": self.block_privilege_escalation,
404 "block_host_network": self.block_host_network,
405 "block_host_pid": self.block_host_pid,
406 "block_host_ipc": self.block_host_ipc,
407 "block_host_path": self.block_host_path,
408 "block_added_capabilities": self.block_added_capabilities,
409 "block_run_as_root": self.block_run_as_root,
410 },
411 validation_enabled=self.validation_enabled,
412 yaml_max_depth=self.yaml_max_depth,
413 )
415 def effective_job_validation_policy(self) -> dict[str, Any]:
416 """Return the validation policy this instance actually enforces.
418 Read straight off the instance attributes that ``validate_manifest``
419 and its helpers compare against, so the answer is the *deployed*
420 policy rather than whatever ``cdk.json`` currently says on some
421 operator's disk. The two can differ for two independent reasons: the
422 cluster may have been deployed from a different checkout, and CDK
423 augments ``trusted_registries`` with the project's own ECR registry
424 hostnames at synth time, so the effective allowlist is strictly
425 larger than the configured one.
427 Numeric caps are reported in the units the validator compares in
428 (millicores, bytes, whole GPUs) alongside the raw configured strings,
429 because ``384`` vCPU and ``384000`` millicores are the same cap and a
430 consumer doing a local pre-check needs to know which it is holding.
432 Sets are returned as sorted lists so the payload is stable across
433 calls and diffable between regions.
434 """
435 return {
436 "validation_enabled": self.validation_enabled,
437 "manifest_caps": {
438 "max_cpu_millicores": self.max_cpu_per_manifest,
439 "max_memory_bytes": self.max_memory_per_manifest,
440 "max_gpu_count": self.max_gpu_per_manifest,
441 "configured": {
442 "max_cpu_per_manifest": os.getenv("MAX_CPU_PER_MANIFEST"),
443 "max_memory_per_manifest": os.getenv("MAX_MEMORY_PER_MANIFEST"),
444 "max_gpu_per_manifest": os.getenv("MAX_GPU_PER_MANIFEST"),
445 },
446 },
447 "allowed_namespaces": sorted(self.allowed_namespaces),
448 "allowed_kinds": sorted(self.allowed_kinds),
449 "allowed_api_versions": {
450 kind: sorted(versions)
451 for kind, versions in sorted(RESOURCE_API_VERSIONS.items())
452 if kind in self.allowed_kinds
453 },
454 "trusted_registries": sorted(self.trusted_registries),
455 "trusted_dockerhub_orgs": sorted(self.trusted_dockerhub_orgs),
456 "require_accelerator_toleration": self.require_accelerator_toleration,
457 "yaml_max_depth": self.yaml_max_depth,
458 "manifest_security_policy": {
459 "block_privileged": self.block_privileged,
460 "block_privilege_escalation": self.block_privilege_escalation,
461 "block_host_network": self.block_host_network,
462 "block_host_pid": self.block_host_pid,
463 "block_host_ipc": self.block_host_ipc,
464 "block_host_path": self.block_host_path,
465 "block_added_capabilities": self.block_added_capabilities,
466 "block_run_as_root": self.block_run_as_root,
467 },
468 }
470 def cluster_resource_governance(self) -> dict[str, Any]:
471 """Return the live ResourceQuota / LimitRange ceilings per namespace.
473 The per-manifest caps in
474 :meth:`effective_job_validation_policy` are only the **first** of three
475 layers. A manifest that clears the front door can still be rejected by
476 the namespace's LimitRange (per-container ceiling) or its ResourceQuota
477 (aggregate ceiling). Reporting only the first layer would let a caller
478 conclude a job is admissible when it is not, so this reads the other
479 two straight from the Kubernetes API.
481 Fail-soft by design: a Kubernetes read failure yields
482 ``status="unavailable"`` with the reason attached rather than raising,
483 because a partial policy answer is more useful than a 500 — and the
484 caller can see explicitly that the layer is missing instead of
485 inferring absence from a silently truncated payload.
486 """
487 namespaces: dict[str, Any] = {}
488 for namespace in sorted(self.allowed_namespaces):
489 entry: dict[str, Any] = {}
490 try:
491 quotas = self.core_v1.list_namespaced_resource_quota(
492 namespace, _request_timeout=self._k8s_timeout
493 )
494 entry["resource_quotas"] = {
495 item.metadata.name: dict(item.status.hard or {})
496 if item.status and item.status.hard
497 else dict(item.spec.hard or {})
498 for item in quotas.items
499 }
501 limit_ranges = self.core_v1.list_namespaced_limit_range(
502 namespace, _request_timeout=self._k8s_timeout
503 )
504 entry["limit_ranges"] = {
505 item.metadata.name: [
506 {
507 "type": limit.type,
508 "max": dict(limit.max or {}),
509 "min": dict(limit.min or {}),
510 "default": dict(limit.default or {}),
511 "defaultRequest": dict(limit.default_request or {}),
512 }
513 for limit in (item.spec.limits or [])
514 ]
515 for item in limit_ranges.items
516 }
517 entry["status"] = "ok"
518 except ApiException as e:
519 logger.warning(
520 "Failed to read resource governance for namespace %s: %s",
521 sanitize_log_value(namespace),
522 e.reason,
523 )
524 entry = {"status": "unavailable", "reason": f"{e.status} {e.reason}"}
525 except Exception as e: # noqa: BLE001 - any read failure is "unavailable"
526 logger.warning(
527 "Failed to read resource governance for namespace %s: %s",
528 sanitize_log_value(namespace),
529 e,
530 )
531 entry = {"status": "unavailable", "reason": str(e)}
532 namespaces[namespace] = entry
533 return namespaces
535 def _resource_access_error(self, api_version: str, kind: str, namespace: str) -> str | None:
536 """Return an authorization error for a CRUD resource identifier."""
537 if namespace not in self.allowed_namespaces:
538 return f"Namespace '{namespace}' is not allowed"
539 if kind not in self.allowed_kinds:
540 return f"Resource kind '{kind}' is not allowed"
541 allowed_versions = RESOURCE_API_VERSIONS.get(kind)
542 if not allowed_versions or api_version not in allowed_versions:
543 return (
544 f"API version '{api_version}' is not allowed for kind '{kind}'. "
545 f"Allowed versions: {sorted(allowed_versions or ())}"
546 )
547 return None
549 # ------------------------------------------------------------------
550 # Security defaults injection
551 # ------------------------------------------------------------------
553 @staticmethod
554 def _extract_pod_spec(manifest: dict[str, Any]) -> dict[str, Any] | None:
555 """Extract the pod spec from a manifest, handling all workload types.
557 Supports:
558 - Deployment / StatefulSet / DaemonSet / ReplicaSet → spec.template.spec
559 - Job → spec.template.spec
560 - CronJob → spec.jobTemplate.spec.template.spec
561 - Bare Pod → spec (when ``containers`` key is present)
563 Returns:
564 The pod spec dict (mutable reference), or ``None`` if the manifest
565 does not contain a recognisable pod spec.
566 """
567 spec = manifest.get("spec")
568 if spec is None or not isinstance(spec, dict):
569 return None
571 kind = manifest.get("kind", "")
573 # CronJob: spec.jobTemplate.spec.template.spec
574 if kind == "CronJob":
575 job_template = spec.get("jobTemplate")
576 if isinstance(job_template, dict):
577 job_spec = job_template.get("spec")
578 if isinstance(job_spec, dict):
579 template = job_spec.get("template")
580 if isinstance(template, dict):
581 pod_spec = template.get("spec")
582 if isinstance(pod_spec, dict):
583 return pod_spec
584 return None
586 # Deployment / StatefulSet / DaemonSet / ReplicaSet / Job:
587 # spec.template.spec
588 if "template" in spec:
589 template = spec.get("template")
590 if isinstance(template, dict):
591 pod_spec = template.get("spec")
592 if isinstance(pod_spec, dict):
593 return pod_spec
594 return None
596 # Bare Pod: spec contains "containers" directly
597 if "containers" in spec:
598 return cast(dict[str, Any], spec)
600 return None
602 def _inject_security_defaults(self, manifest: dict[str, Any]) -> dict[str, Any]:
603 """Inject security defaults into user-submitted manifests.
605 Currently injects:
606 - ``automountServiceAccountToken: false`` in the pod spec (unless the
607 user has explicitly set it).
609 The method mutates *manifest* in-place and returns it for convenience.
611 For a TrainJob the default is injected into every pod spec embedded
612 in ``runtimePatches`` (live references into the manifest); the base
613 pod template comes from the shipped ClusterTrainingRuntime, which
614 already disables the token.
615 """
616 if manifest.get("kind") == "TrainJob":
617 for embedded in extract_trainjob_pod_specs(manifest).embedded:
618 embedded.setdefault("automountServiceAccountToken", False)
619 return manifest
620 pod_spec = self._extract_pod_spec(manifest)
621 if pod_spec is not None:
622 # Use setdefault so we don't override an explicit user choice
623 pod_spec.setdefault("automountServiceAccountToken", False)
624 return manifest
626 @property
627 def dynamic_client(self) -> dynamic.DynamicClient:
628 """Lazy-initialized dynamic client for CRD support."""
629 if self._dynamic_client is None:
630 self._dynamic_client = dynamic.DynamicClient(self.api_client)
631 return self._dynamic_client
633 def _parse_cpu_string(self, cpu_str: str) -> int:
634 """Parse CPU string to millicores"""
635 return parse_cpu_millicores(cpu_str)
637 def _parse_memory_string(self, memory_str: str) -> int:
638 """Parse memory string to bytes"""
639 return parse_memory_bytes(memory_str)
641 def _check_yaml_depth(self, obj: Any, current_depth: int = 0) -> bool:
642 """Check if a parsed YAML/JSON object exceeds max nesting depth.
644 Recursively walks dicts and lists. Returns False if depth exceeds
645 ``self.yaml_max_depth``.
647 Args:
648 obj: The parsed object to check (dict, list, or scalar).
649 current_depth: Current recursion depth (callers should leave at 0).
651 Returns:
652 True if the object is within the depth limit, False otherwise.
653 """
654 if current_depth > self.yaml_max_depth:
655 return False
656 if isinstance(obj, dict):
657 return all(self._check_yaml_depth(v, current_depth + 1) for v in obj.values())
658 if isinstance(obj, list):
659 return all(self._check_yaml_depth(item, current_depth + 1) for item in obj)
660 return True
662 def validate_manifest(
663 self,
664 manifest: dict[str, Any],
665 default_namespace: str | None = None,
666 ) -> tuple[bool, str | None]:
667 """Validate a Kubernetes manifest for security and resource constraints.
669 ``default_namespace`` is the request-level destination for manifests
670 that omit ``metadata.namespace``. Validation and apply must resolve the
671 same effective namespace or the request default could bypass the
672 namespace allowlist.
674 Returns: ``(is_valid, error_message)``.
675 """
676 if not self.validation_enabled:
677 return True, None
679 try:
680 # YAML depth check — reject excessively nested documents
681 if not self._check_yaml_depth(manifest):
682 return (
683 False,
684 f"Manifest exceeds maximum nesting depth of {self.yaml_max_depth} levels",
685 )
687 # Basic structure validation
688 required_fields = ["apiVersion", "kind", "metadata"]
689 for field in required_fields:
690 if field not in manifest:
691 return False, f"Missing required field: {field}"
693 # Validate metadata
694 metadata = manifest.get("metadata", {})
695 if "name" not in metadata:
696 return False, "Missing metadata.name field"
698 # Validate namespace
699 namespace = metadata.get("namespace", default_namespace or "gco-jobs")
700 if namespace not in self.allowed_namespaces:
701 return (
702 False,
703 f"Namespace '{namespace}' not allowed. Allowed namespaces: {list(self.allowed_namespaces)}",
704 )
706 # Validate resource kind using the policy shared with the SQS path.
707 kind = manifest.get("kind", "")
708 kind_valid, kind_error = validate_resource_kind(manifest, self.allowed_kinds)
709 if not kind_valid:
710 return False, kind_error
712 # Validate resource limits for workload resources
713 if kind in [
714 "Deployment",
715 "Job",
716 "CronJob",
717 "StatefulSet",
718 "DaemonSet",
719 "TrainJob",
720 ]:
721 resource_valid, resource_error = self._validate_resource_limits(manifest)
722 if not resource_valid:
723 return False, resource_error
725 # Require accelerator jobs to carry a matching toleration.
726 if self.require_accelerator_toleration:
727 tol_valid, tol_error = self._validate_tolerations(manifest)
728 if not tol_valid:
729 return False, tol_error
731 # Security validations
732 sec_valid, sec_error = self._validate_security_context(manifest)
733 if not sec_valid:
734 return False, f"Security context validation failed: {sec_error}"
736 # Validate image sources (prevent pulling from untrusted registries)
737 img_valid, img_error = self._validate_image_sources(manifest)
738 if not img_valid:
739 return False, img_error or "Untrusted image sources detected"
741 return True, None
743 except Exception as e:
744 logger.error(f"Error validating manifest: {e}")
745 return False, f"Validation error: {e!s}"
747 def _validate_resource_limits(self, manifest: dict[str, Any]) -> tuple[bool, str]:
748 """Validate resource limits in manifest.
750 Delegates to :func:`gco.job_admission.check_resource_caps` so the
751 offline and multi-region pre-checks judge a manifest with the same code
752 that gates it here.
754 Returns:
755 Tuple of (is_valid, error_message). error_message is empty if valid.
756 """
757 return check_resource_caps(manifest, self.job_validation_policy())
759 def _validate_tolerations(self, manifest: dict[str, Any]) -> tuple[bool, str | None]:
760 """Require accelerator jobs to carry a matching node toleration.
762 Delegates to :func:`gco.job_admission.check_tolerations`; see there for
763 why TrainJob needs every pod-spec view unioned before matching.
765 Returns:
766 Tuple of (is_valid, error_message). error_message is None if valid.
767 """
768 return check_tolerations(manifest)
770 def _requested_accelerators(self, pod_spec: dict[str, Any]) -> set[str]:
771 """Return the set of accelerator taint keys any container requests
772 a nonzero quantity of."""
773 return requested_accelerators(pod_spec)
775 def _get_all_containers(self, pod_spec: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
776 """Get all containers from pod spec including init and ephemeral containers.
778 Returns:
779 List of (container_type, container_dict) tuples where container_type
780 is one of 'container', 'initContainer', or 'ephemeralContainer'.
781 """
782 return _iter_all_containers(pod_spec)
784 def _validate_security_context(self, manifest: dict[str, Any]) -> tuple[bool, str | None]:
785 """Validate security context settings.
787 Delegates to :func:`gco.job_admission.check_security_context`, which
788 reads the same eight toggles off a policy object instead of ``self``.
790 Returns:
791 Tuple of (is_valid, error_message). error_message is None if valid.
792 """
793 return check_security_context(manifest, self.job_validation_policy())
795 @staticmethod
796 def _is_registry_domain(entry: str) -> bool:
797 """Check if a registry entry is a proper domain (contains dot or colon).
799 A proper registry domain contains either a dot (e.g., 'docker.io', 'gcr.io')
800 or a colon (e.g., 'localhost:5000'). Entries without these are Docker Hub
801 organization names (e.g., 'nvidia', 'gco').
802 """
803 return "." in entry or ":" in entry
805 def _validate_image_sources(self, manifest: dict[str, Any]) -> tuple[bool, str | None]:
806 """Validate container image sources against this deployment's allowlists.
808 Delegates to the module-level :func:`validate_image_sources` so the
809 REST/SQS services and offline validators share one implementation.
810 """
811 return validate_image_sources(
812 manifest,
813 trusted_registries=self.trusted_registries,
814 trusted_dockerhub_orgs=self.trusted_dockerhub_orgs,
815 )
817 async def process_manifest_submission(
818 self, request: ManifestSubmissionRequest
819 ) -> ManifestSubmissionResponse:
820 """
821 Process a manifest submission request
822 """
823 logger.info(f"Processing manifest submission with {len(request.manifests)} manifests")
825 resources = []
826 errors = []
827 overall_success = True
829 try:
830 # Process each manifest
831 for i, manifest_data in enumerate(request.manifests):
832 try:
833 # Validate manifest
834 is_valid, error_msg = self.validate_manifest(manifest_data, request.namespace)
835 if not is_valid:
836 error_msg = f"Manifest {i + 1} validation failed: {error_msg}"
837 errors.append(error_msg)
838 logger.error(error_msg)
840 # Create failed resource status
841 resource_status = ResourceStatus(
842 api_version=manifest_data.get("apiVersion", "unknown"),
843 kind=manifest_data.get("kind", "unknown"),
844 name=manifest_data.get("metadata", {}).get("name", f"manifest-{i + 1}"),
845 namespace=manifest_data.get("metadata", {}).get(
846 "namespace", request.namespace or "gco-jobs"
847 ),
848 status="failed",
849 message=error_msg,
850 )
851 resources.append(resource_status)
852 overall_success = False
853 continue
855 # Apply manifest if validation passed
856 if not request.dry_run:
857 resource_status = await self._apply_manifest(
858 manifest_data, request.namespace
859 )
860 resources.append(resource_status)
862 if not resource_status.is_successful():
863 overall_success = False
864 else:
865 # Dry run - just validate
866 resource_status = ResourceStatus(
867 api_version=manifest_data.get("apiVersion", "unknown"),
868 kind=manifest_data.get("kind", "unknown"),
869 name=manifest_data.get("metadata", {}).get("name", "unknown"),
870 namespace=manifest_data.get("metadata", {}).get(
871 "namespace", request.namespace or "gco-jobs"
872 ),
873 status="unchanged",
874 message="Dry run - validation passed",
875 )
876 resources.append(resource_status)
878 except Exception as e:
879 error_msg = f"Error processing manifest {i + 1}: {e!s}"
880 errors.append(error_msg)
881 logger.error(error_msg)
882 overall_success = False
884 # Create failed resource status
885 resource_status = ResourceStatus(
886 api_version=manifest_data.get("apiVersion", "unknown"),
887 kind=manifest_data.get("kind", "unknown"),
888 name=manifest_data.get("metadata", {}).get("name", f"manifest-{i + 1}"),
889 namespace=manifest_data.get("metadata", {}).get(
890 "namespace", request.namespace or "gco-jobs"
891 ),
892 status="failed",
893 message=str(e),
894 )
895 resources.append(resource_status)
897 except Exception as e:
898 error_msg = f"Fatal error processing manifest submission: {e!s}"
899 errors.append(error_msg)
900 logger.error(error_msg)
901 overall_success = False
903 response = ManifestSubmissionResponse(
904 success=overall_success,
905 cluster_id=self.cluster_id,
906 region=self.region,
907 resources=resources,
908 errors=errors if errors else None,
909 )
911 logger.info(
912 f"Manifest submission completed - Success: {overall_success}, "
913 f"Resources: {len(resources)}, Errors: {len(errors)}"
914 )
916 return response
918 @staticmethod
919 def queued_job_name(original_name: str, queue_job_id: str) -> str:
920 """Return a DNS-label-safe Kubernetes name deterministically fenced by queue ID."""
921 suffix = hashlib.sha256(queue_job_id.encode("utf-8")).hexdigest()[:16]
922 prefix = re.sub(r"[^a-z0-9-]+", "-", original_name.lower()).strip("-")
923 prefix = prefix[: 63 - len(suffix) - 1].rstrip("-") or "gco-job"
924 return f"{prefix}-{suffix}"
926 def apply_queued_job(
927 self,
928 manifest_data: dict[str, Any],
929 namespace: str,
930 queue_job_id: str,
931 ) -> ResourceStatus:
932 """Create or adopt exactly one deterministic ``batch/v1`` Job.
934 This path deliberately bypasses generic manifest upsert semantics: it
935 never deletes, renames, or replaces an existing Job. An ambiguous API
936 result is safe to retry because the same queue ID always resolves to the
937 same Kubernetes name and adoption requires the full queue ID annotation.
938 """
939 manifest = copy.deepcopy(manifest_data)
940 if manifest.get("apiVersion") != "batch/v1" or manifest.get("kind") != "Job":
941 raise QueuedJobNotCreatedError(
942 "Central queue accepts only apiVersion 'batch/v1', kind 'Job'"
943 )
945 metadata = manifest.get("metadata")
946 if not isinstance(metadata, dict):
947 raise QueuedJobNotCreatedError("Queued Job metadata must be an object")
948 declared_namespace = metadata.get("namespace")
949 if declared_namespace is not None and declared_namespace != namespace:
950 raise QueuedJobNotCreatedError("Queued Job namespace does not match the queue envelope")
951 original_name = metadata.get("name")
952 if not isinstance(original_name, str) or not original_name:
953 raise QueuedJobNotCreatedError("Queued Job metadata.name is required")
955 deterministic_name = self.queued_job_name(original_name, queue_job_id)
956 metadata["name"] = deterministic_name
957 metadata["namespace"] = namespace
958 labels = metadata.setdefault("labels", {})
959 annotations = metadata.setdefault("annotations", {})
960 if not isinstance(labels, dict) or not isinstance(annotations, dict):
961 raise QueuedJobNotCreatedError(
962 "Queued Job metadata labels and annotations must be objects"
963 )
964 labels["gco.io/managed-by"] = "central-queue"
965 labels["gco.io/queue-job-key"] = hashlib.sha256(queue_job_id.encode("utf-8")).hexdigest()[
966 :32
967 ]
968 annotations["gco.io/queue-job-id"] = queue_job_id
969 annotations["gco.io/original-job-name"] = original_name
971 is_valid, validation_error = self.validate_manifest(manifest, namespace)
972 if not is_valid:
973 raise QueuedJobNotCreatedError(f"Queued Job validation failed: {validation_error}")
974 self._inject_security_defaults(manifest)
976 try:
977 job = self.batch_v1.read_namespaced_job(
978 name=deterministic_name,
979 namespace=namespace,
980 _request_timeout=self._k8s_timeout,
981 )
982 operation = "unchanged"
983 message = "Existing deterministic Kubernetes Job adopted"
984 except ApiException as error:
985 if error.status != 404:
986 if _is_retryable_kubernetes_api_error(error):
987 raise RetryableQueuedJobApplyError(
988 "Kubernetes Job lookup was inconclusive; retry deterministic adoption"
989 ) from error
990 raise
991 try:
992 job = self.batch_v1.create_namespaced_job(
993 namespace=namespace,
994 body=manifest,
995 _request_timeout=self._k8s_timeout,
996 )
997 operation = "created"
998 message = "Deterministic Kubernetes Job created"
999 except ApiException as create_error:
1000 if create_error.status == 409:
1001 try:
1002 job = self.batch_v1.read_namespaced_job(
1003 name=deterministic_name,
1004 namespace=namespace,
1005 _request_timeout=self._k8s_timeout,
1006 )
1007 except ApiException as adoption_error:
1008 if adoption_error.status == 404 or _is_retryable_kubernetes_api_error(
1009 adoption_error
1010 ):
1011 raise RetryableQueuedJobApplyError(
1012 "Concurrent Kubernetes Job adoption was inconclusive"
1013 ) from adoption_error
1014 raise
1015 except Exception as adoption_error:
1016 raise RetryableQueuedJobApplyError(
1017 "Concurrent Kubernetes Job adoption was inconclusive"
1018 ) from adoption_error
1019 operation = "unchanged"
1020 message = "Concurrent deterministic Kubernetes Job adopted"
1021 elif _is_retryable_kubernetes_api_error(create_error):
1022 raise RetryableQueuedJobApplyError(
1023 "Kubernetes Job create result was inconclusive; retry deterministic adoption"
1024 ) from create_error
1025 else:
1026 raise
1027 except Exception as create_error:
1028 # A timeout or connection loss can occur after the API server
1029 # persisted the Job. Never mark that ambiguous result terminal.
1030 raise RetryableQueuedJobApplyError(
1031 "Kubernetes Job create result was inconclusive; retry deterministic adoption"
1032 ) from create_error
1033 except Exception as read_error:
1034 raise RetryableQueuedJobApplyError(
1035 "Kubernetes Job lookup was inconclusive; retry deterministic adoption"
1036 ) from read_error
1038 actual_annotations = getattr(job.metadata, "annotations", None) or {}
1039 if actual_annotations.get("gco.io/queue-job-id") != queue_job_id:
1040 raise RuntimeError(
1041 f"Kubernetes Job name collision for {namespace}/{deterministic_name}"
1042 )
1043 uid = str(getattr(job.metadata, "uid", "") or "")
1044 actual_name = str(getattr(job.metadata, "name", "") or deterministic_name)
1045 actual_namespace = str(getattr(job.metadata, "namespace", "") or namespace)
1046 if not uid:
1047 raise RuntimeError("Kubernetes API returned a queued Job without a UID")
1048 return ResourceStatus(
1049 api_version="batch/v1",
1050 kind="Job",
1051 name=actual_name,
1052 namespace=actual_namespace,
1053 status=operation,
1054 message=message,
1055 uid=uid,
1056 )
1058 def read_queued_job(self, name: str, namespace: str) -> V1Job:
1059 """Read one reconciled Job through the processor's bounded client contract."""
1060 return self.batch_v1.read_namespaced_job(
1061 name=name,
1062 namespace=namespace,
1063 _request_timeout=self._k8s_timeout,
1064 )
1066 async def _apply_manifest(
1067 self, manifest_data: dict[str, Any], default_namespace: str | None = None
1068 ) -> ResourceStatus:
1069 """
1070 Apply a single manifest to the cluster.
1072 For Jobs and CronJobs, if the resource already exists and is completed/failed,
1073 it will be automatically deleted and recreated (since these resources are immutable).
1074 """
1075 try:
1076 api_version: str = manifest_data.get("apiVersion", "unknown")
1077 kind: str = manifest_data.get("kind", "unknown")
1078 metadata = manifest_data.get("metadata", {})
1079 name: str = metadata.get("name", "unknown")
1080 namespace: str = metadata.get("namespace", default_namespace or "gco-jobs")
1082 # Ensure namespace is set in manifest
1083 if "namespace" not in metadata and namespace:
1084 manifest_data["metadata"]["namespace"] = namespace
1086 # Inject security defaults (e.g., automountServiceAccountToken: false)
1087 self._inject_security_defaults(manifest_data)
1089 # Check if resource already exists
1090 existing_resource = await self._get_existing_resource(
1091 api_version, kind, name, namespace
1092 )
1094 if existing_resource:
1095 # Jobs are immutable — if one already exists and is finished,
1096 # delete it first so we can recreate cleanly.
1097 # If the job is still active, auto-rename to avoid collision.
1098 if kind == "Job":
1099 if self._is_job_finished(existing_resource):
1100 logger.info(
1101 f"Job {name} already exists and is finished, deleting before recreating"
1102 )
1103 await self.delete_resource(api_version, kind, name, namespace)
1104 import asyncio
1106 await asyncio.sleep(1)
1107 await self._create_resource(manifest_data)
1108 status = "created"
1109 message = "Previous completed job replaced with new submission"
1110 else:
1111 # Active job — rename to avoid destroying it
1112 import uuid
1114 suffix = uuid.uuid4().hex[:5]
1115 new_name = f"{name}-{suffix}"
1116 manifest_data["metadata"]["name"] = new_name
1117 logger.warning(
1118 f"Job {name} is still active, renamed new submission to {new_name}"
1119 )
1120 await self._create_resource(manifest_data)
1121 status = "created"
1122 message = (
1123 f"Job '{name}' is still running. "
1124 f"New submission renamed to '{new_name}'."
1125 )
1126 name = new_name
1127 else:
1128 # Update existing resource (works for mutable resources)
1129 updated_resource = await self._update_resource(manifest_data)
1130 status = "updated" if updated_resource else "unchanged"
1131 message = (
1132 "Resource updated successfully" if updated_resource else "No changes needed"
1133 )
1134 else:
1135 # Create new resource
1136 await self._create_resource(manifest_data)
1137 status = "created"
1138 message = "Resource created successfully"
1140 return ResourceStatus(
1141 api_version=api_version,
1142 kind=kind,
1143 name=name,
1144 namespace=namespace,
1145 status=status,
1146 message=message,
1147 )
1149 except ApiException as e:
1150 logger.error(f"Kubernetes API error applying manifest: {e}")
1151 return ResourceStatus(
1152 api_version=manifest_data.get("apiVersion", "unknown"),
1153 kind=manifest_data.get("kind", "unknown"),
1154 name=manifest_data.get("metadata", {}).get("name", "unknown"),
1155 namespace=manifest_data.get("metadata", {}).get("namespace", "gco-jobs"),
1156 status="failed",
1157 message=f"API error: {e.reason}",
1158 )
1159 except Exception as e:
1160 logger.error(f"Error applying manifest: {e}")
1161 return ResourceStatus(
1162 api_version=manifest_data.get("apiVersion", "unknown"),
1163 kind=manifest_data.get("kind", "unknown"),
1164 name=manifest_data.get("metadata", {}).get("name", "unknown"),
1165 namespace=manifest_data.get("metadata", {}).get("namespace", "gco-jobs"),
1166 status="failed",
1167 message=str(e),
1168 )
1170 def _is_job_finished(self, job_resource: dict[str, Any]) -> bool:
1171 """Check if a Kubernetes Job resource is in a terminal state (Complete or Failed)."""
1172 status = job_resource.get("status", {})
1173 conditions = status.get("conditions") or []
1174 for condition in conditions:
1175 cond_type = condition.get("type", "")
1176 cond_status = condition.get("status", "")
1177 if cond_type in ("Complete", "Failed") and cond_status == "True":
1178 return True
1179 return False
1181 async def _get_existing_resource(
1182 self,
1183 api_version: str,
1184 kind: str,
1185 name: str,
1186 namespace: str,
1187 *,
1188 api_resource: Any | None = None,
1189 ) -> dict[str, Any] | None:
1190 """Check if a resource already exists using dynamic client."""
1191 try:
1192 # Reuse an already-authorized discovery result when supplied so a
1193 # status request cannot observe a different resource definition
1194 # between the scope check and the actual read.
1195 if api_resource is None:
1196 api_resource = self._get_api_resource(api_version, kind)
1198 # Try to get the resource
1199 if namespace and api_resource.namespaced:
1200 resource = api_resource.get(name=name, namespace=namespace)
1201 else:
1202 resource = api_resource.get(name=name)
1204 if resource is not None:
1205 return dict(resource.to_dict())
1207 except ApiException as e:
1208 if e.status == 404:
1209 return None # Resource doesn't exist
1210 raise
1211 except ValueError:
1212 # Unknown resource type
1213 return None
1215 return None
1217 def _get_api_resource(self, api_version: str, kind: str) -> Any:
1218 """Get the API resource for a given apiVersion and kind using dynamic client."""
1219 try:
1220 return self.dynamic_client.resources.get(api_version=api_version, kind=kind)
1221 except ResourceNotFoundError as e:
1222 logger.error(
1223 "Resource type not found: %s/%s",
1224 sanitize_log_value(api_version),
1225 sanitize_log_value(kind),
1226 )
1227 # A policy-allowed kind whose addon is not installed gets the
1228 # actionable remedy instead of an inscrutable discovery error.
1229 addon_hint = ADDON_KIND_HINTS.get(kind)
1230 if addon_hint:
1231 raise ValueError(addon_hint) from e
1232 raise ValueError(f"Unknown resource type: {api_version}/{kind}") from e
1234 async def _create_resource(self, manifest_data: dict[str, Any]) -> Any:
1235 """Create a resource and return the API object, including server identity."""
1236 try:
1237 api_version = manifest_data.get("apiVersion", "")
1238 kind = manifest_data.get("kind", "")
1239 namespace = manifest_data.get("metadata", {}).get("namespace")
1241 api_resource = self._get_api_resource(api_version, kind)
1242 if namespace and api_resource.namespaced:
1243 return api_resource.create(body=manifest_data, namespace=namespace)
1244 return api_resource.create(body=manifest_data)
1245 except Exception as e:
1246 logger.error(f"Error creating resource: {e}")
1247 raise
1249 async def _update_resource(self, manifest_data: dict[str, Any]) -> bool:
1250 """Update an existing resource using dynamic client"""
1251 try:
1252 api_version = manifest_data.get("apiVersion", "")
1253 kind = manifest_data.get("kind", "")
1254 name = manifest_data.get("metadata", {}).get("name", "")
1255 namespace = manifest_data.get("metadata", {}).get("namespace")
1257 # Get the API resource
1258 api_resource = self._get_api_resource(api_version, kind)
1260 # Update the resource using patch (server-side apply)
1261 if namespace and api_resource.namespaced:
1262 api_resource.patch(
1263 body=manifest_data,
1264 name=name,
1265 namespace=namespace,
1266 content_type="application/merge-patch+json",
1267 )
1268 else:
1269 api_resource.patch(
1270 body=manifest_data,
1271 name=name,
1272 content_type="application/merge-patch+json",
1273 )
1275 return True
1276 except Exception as e:
1277 logger.error(f"Error updating resource: {e}")
1278 raise
1280 async def delete_resource(
1281 self, api_version: str, kind: str, name: str, namespace: str
1282 ) -> ResourceStatus:
1283 """
1284 Delete a resource from the cluster using dynamic client
1285 """
1286 access_error = self._resource_access_error(api_version, kind, namespace)
1287 if access_error:
1288 return ResourceStatus(
1289 api_version=api_version,
1290 kind=kind,
1291 name=name,
1292 namespace=namespace,
1293 status="forbidden",
1294 message=access_error,
1295 )
1297 try:
1298 # Get the API resource
1299 api_resource = self._get_api_resource(api_version, kind)
1300 if not api_resource.namespaced:
1301 return ResourceStatus(
1302 api_version=api_version,
1303 kind=kind,
1304 name=name,
1305 namespace=namespace,
1306 status="forbidden",
1307 message="Cluster-scoped resource operations are not allowed",
1308 )
1310 # Every authorized GVK is namespaced; never drop the namespace.
1311 api_resource.delete(name=name, namespace=namespace)
1313 return ResourceStatus(
1314 api_version=api_version,
1315 kind=kind,
1316 name=name,
1317 namespace=namespace,
1318 status="deleted",
1319 message="Resource deleted successfully",
1320 )
1322 except ValueError as e:
1323 # Unknown resource type
1324 return ResourceStatus(
1325 api_version=api_version,
1326 kind=kind,
1327 name=name,
1328 namespace=namespace,
1329 status="failed",
1330 message=str(e),
1331 )
1332 except ApiException as e:
1333 if e.status == 404:
1334 return ResourceStatus(
1335 api_version=api_version,
1336 kind=kind,
1337 name=name,
1338 namespace=namespace,
1339 status="unchanged",
1340 message="Resource not found (already deleted)",
1341 )
1342 return ResourceStatus(
1343 api_version=api_version,
1344 kind=kind,
1345 name=name,
1346 namespace=namespace,
1347 status="failed",
1348 message=f"Delete failed: {e.reason}",
1349 )
1350 except Exception as e:
1351 return ResourceStatus(
1352 api_version=api_version,
1353 kind=kind,
1354 name=name,
1355 namespace=namespace,
1356 status="failed",
1357 message=str(e),
1358 )
1360 async def list_jobs(
1361 self, namespace: str | None = None, status_filter: str | None = None
1362 ) -> list[dict[str, Any]]:
1363 """
1364 List Kubernetes Jobs from allowed namespaces.
1366 Args:
1367 namespace: Filter by specific namespace (must be in allowed_namespaces)
1368 status_filter: Filter by status: "running", "completed", "failed"
1370 Returns:
1371 List of job dictionaries with metadata and status
1372 """
1373 jobs = []
1375 # Determine which namespaces to query
1376 if namespace:
1377 if namespace not in self.allowed_namespaces:
1378 raise ValueError(
1379 f"Namespace '{namespace}' not allowed. "
1380 f"Allowed namespaces: {list(self.allowed_namespaces)}"
1381 )
1382 namespaces_to_query = [namespace]
1383 else:
1384 namespaces_to_query = list(self.allowed_namespaces)
1386 for ns in namespaces_to_query:
1387 try:
1388 job_list = self.batch_v1.list_namespaced_job(
1389 namespace=ns, _request_timeout=self._k8s_timeout
1390 )
1391 for job in job_list.items:
1392 job_dict = self._job_to_dict(job)
1394 # Apply status filter
1395 if status_filter:
1396 job_status = self._get_job_status(job)
1397 if job_status != status_filter:
1398 continue
1400 jobs.append(job_dict)
1401 except ApiException as e:
1402 logger.warning(
1403 "Failed to list jobs in namespace %s: %s", sanitize_log_value(ns), e.reason
1404 )
1405 continue
1407 return jobs
1409 def _job_to_dict(self, job: V1Job) -> dict[str, Any]:
1410 """Convert a Kubernetes Job object to a dictionary."""
1411 metadata = job.metadata
1412 status = job.status
1413 spec = job.spec
1415 return {
1416 "metadata": {
1417 "name": metadata.name,
1418 "namespace": metadata.namespace,
1419 "creationTimestamp": (
1420 metadata.creation_timestamp.isoformat() if metadata.creation_timestamp else None
1421 ),
1422 "labels": metadata.labels or {},
1423 "uid": metadata.uid,
1424 },
1425 "spec": {
1426 "parallelism": spec.parallelism,
1427 "completions": spec.completions,
1428 "backoffLimit": spec.backoff_limit,
1429 },
1430 "status": {
1431 "active": status.active or 0,
1432 "succeeded": status.succeeded or 0,
1433 "failed": status.failed or 0,
1434 "startTime": status.start_time.isoformat() if status.start_time else None,
1435 "completionTime": (
1436 status.completion_time.isoformat() if status.completion_time else None
1437 ),
1438 "conditions": [
1439 {
1440 "type": c.type,
1441 "status": c.status,
1442 "reason": c.reason,
1443 "message": c.message,
1444 }
1445 for c in (status.conditions or [])
1446 ],
1447 },
1448 }
1450 def _get_job_status(self, job: V1Job) -> str:
1451 """Determine the status of a job: running, completed, or failed."""
1452 status = job.status
1453 conditions = status.conditions or []
1455 for condition in conditions:
1456 if condition.type == "Complete" and condition.status == "True":
1457 return "completed"
1458 if condition.type == "Failed" and condition.status == "True":
1459 return "failed"
1461 if (status.active or 0) > 0:
1462 return "running"
1464 return "pending"
1466 async def get_resource_status(
1467 self, api_version: str, kind: str, name: str, namespace: str
1468 ) -> dict[str, Any] | None:
1469 """
1470 Get the status of a specific resource
1471 """
1472 access_error = self._resource_access_error(api_version, kind, namespace)
1473 if access_error:
1474 return {
1475 "api_version": api_version,
1476 "kind": kind,
1477 "name": name,
1478 "namespace": namespace,
1479 "exists": False,
1480 "forbidden": True,
1481 "error": access_error,
1482 }
1484 try:
1485 api_resource = self._get_api_resource(api_version, kind)
1486 if not api_resource.namespaced:
1487 return {
1488 "api_version": api_version,
1489 "kind": kind,
1490 "name": name,
1491 "namespace": namespace,
1492 "exists": False,
1493 "forbidden": True,
1494 "error": "Cluster-scoped resource operations are not allowed",
1495 }
1496 resource = await self._get_existing_resource(
1497 api_version,
1498 kind,
1499 name,
1500 namespace,
1501 api_resource=api_resource,
1502 )
1503 if resource:
1504 return {
1505 "api_version": api_version,
1506 "kind": kind,
1507 "name": name,
1508 "namespace": namespace,
1509 "exists": True,
1510 "status": resource.get("status", {}),
1511 "metadata": resource.get("metadata", {}),
1512 "spec": resource.get("spec", {}),
1513 }
1514 return {
1515 "api_version": api_version,
1516 "kind": kind,
1517 "name": name,
1518 "namespace": namespace,
1519 "exists": False,
1520 }
1521 except Exception as e:
1522 logger.error(f"Error getting resource status: {e}")
1523 return None
1526def _manifest_security_policy_from_env() -> dict[str, bool]:
1527 return {
1528 key: parse_boolean_environment(key.upper(), default)
1529 for key, default in MANIFEST_SECURITY_POLICY_DEFAULTS.items()
1530 }
1533def create_manifest_processor_from_env() -> ManifestProcessor:
1534 """
1535 Create ManifestProcessor instance from environment variables
1536 """
1537 cluster_id = os.getenv("CLUSTER_NAME", "unknown-cluster")
1538 region = os.getenv("REGION", "unknown-region")
1540 # Enable structured JSON logging for CloudWatch Insights
1541 configure_structured_logging(
1542 service_name="manifest-processor",
1543 cluster_id=cluster_id,
1544 region=region,
1545 )
1547 # Load configuration from environment
1548 config_dict = {
1549 "max_cpu_per_manifest": os.getenv(
1550 "MAX_CPU_PER_MANIFEST",
1551 str(DEFAULT_MANIFEST_RESOURCE_CAPS["max_cpu_per_manifest"]),
1552 ),
1553 "max_memory_per_manifest": os.getenv(
1554 "MAX_MEMORY_PER_MANIFEST",
1555 str(DEFAULT_MANIFEST_RESOURCE_CAPS["max_memory_per_manifest"]),
1556 ),
1557 "max_gpu_per_manifest": int(
1558 os.getenv(
1559 "MAX_GPU_PER_MANIFEST",
1560 str(DEFAULT_MANIFEST_RESOURCE_CAPS["max_gpu_per_manifest"]),
1561 )
1562 ),
1563 "require_accelerator_toleration": parse_boolean_environment(
1564 "REQUIRE_ACCELERATOR_TOLERATION", True
1565 ),
1566 "allowed_namespaces": (
1567 ["gco-jobs"]
1568 if os.getenv("ALLOWED_NAMESPACES") is None
1569 else [
1570 namespace.strip()
1571 for namespace in os.environ["ALLOWED_NAMESPACES"].split(",")
1572 if namespace.strip()
1573 ]
1574 ),
1575 "validation_enabled": parse_boolean_environment("VALIDATION_ENABLED", True),
1576 "yaml_max_depth": int(os.getenv("YAML_MAX_DEPTH", "50")),
1577 "manifest_security_policy": _manifest_security_policy_from_env(),
1578 }
1580 allowed_kinds_env = os.getenv("ALLOWED_KINDS")
1581 if allowed_kinds_env is not None:
1582 # An absent variable uses the authoritative defaults; an explicitly
1583 # empty value is a deliberate deny-all policy and must stay empty.
1584 config_dict["allowed_kinds"] = [
1585 kind.strip() for kind in allowed_kinds_env.split(",") if kind.strip()
1586 ]
1588 # Image registry allowlist — sourced from the same CDK env vars the
1589 # queue_processor reads, so an attacker who holds sqs:SendMessage on
1590 # the regional queue can't reach an image source the REST path
1591 # rejects. When unset (or empty) the ManifestProcessor falls back
1592 # to its hardcoded default. Empty/missing values are dropped to
1593 # match the queue_processor's parsing rules.
1594 trusted_registries_env = os.getenv("TRUSTED_REGISTRIES", "")
1595 trusted_registries = [r.strip() for r in trusted_registries_env.split(",") if r.strip()]
1596 if trusted_registries:
1597 config_dict["trusted_registries"] = trusted_registries
1599 trusted_dockerhub_orgs_env = os.getenv("TRUSTED_DOCKERHUB_ORGS", "")
1600 trusted_dockerhub_orgs = [o.strip() for o in trusted_dockerhub_orgs_env.split(",") if o.strip()]
1601 if trusted_dockerhub_orgs:
1602 config_dict["trusted_dockerhub_orgs"] = trusted_dockerhub_orgs
1604 return ManifestProcessor(cluster_id, region, config_dict)