spark_on_k8s.api.submission_policy

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

AdmissionError

Bases: ValueError

A job is outside the administrator's remote submission policy.

Source code in spark_on_k8s/api/submission_policy.py

class AdmissionError(ValueError):
    """A job is outside the administrator's remote submission policy."""

SettingsSource

Serve the last valid settings, re-reading changed files at most once per reload interval.

Source code in spark_on_k8s/api/submission_policy.py

class SettingsSource:
    """Serve the last valid settings, re-reading changed files at most once per reload interval."""

    def __init__(self, clock=time.monotonic):
        self._clock = clock
        self._lock = threading.Lock()
        self._settings: SubmissionSettings | None = None
        self._next_check = 0.0
        self._rejected = None

    def get(self) -> SubmissionSettings:
        settings = self._settings
        if settings is None:
            with self._lock:
                if self._settings is None:
                    # Startup still fails closed on missing or invalid configuration.
                    self._settings = SubmissionSettings.from_env()
                    self._next_check = self._clock() + self._settings.reload_interval
                return self._settings
        # Requests never wait for another request's check; they use the settings already loaded.
        if settings.reload_interval and self._clock() >= self._next_check and self._lock.acquire(False):
            try:
                settings = self._settings = self._reload(settings)
                self._next_check = self._clock() + settings.reload_interval
            finally:
                self._lock.release()
        return settings

    def _reload(self, settings):
        paths = [path for path, _ in settings.sources]
        current = _current_identities(paths)
        if current == tuple(identity for _, identity in settings.sources) or current == self._rejected:
            return settings
        try:
            updated = SubmissionSettings.read(*paths, settings.reload_interval)
            # A changed policy must pass the same namespace-scope check as at startup.
            from spark_on_k8s.api.namespace_scope import namespace_scope

            namespace_scope().require_configured(updated.policy.namespaces, "The submission policy")
        except Exception as error:
            # Report a changed file once; never include token values in the message.
            self._rejected = current
            logger.error(
                "Kept the previous remote submission token and policy; the changed files are invalid: %s",
                error,
            )
            return settings
        self._rejected = None
        # Credential and policy changes are security relevant, so they are visible at the default level.
        logger.warning(
            "Reloaded the remote submission token and policy files; accepted tokens: %d.",
            1 + len(updated.extra_tokens),
        )
        return updated

SubmissionPolicy dataclass

Source code in spark_on_k8s/api/submission_policy.py

@dataclass(frozen=True)
class SubmissionPolicy:
    namespaces: dict[str, list[str]]
    applications: list[dict[str, Any]]
    limits: dict[str, int]
    spark_conf: dict[str, list[Any]]
    # Label key -> (allowed values or None, full-match pattern or None).
    required_labels: dict[str, tuple[list[str] | None, re.Pattern | None]] = field(default_factory=dict)
    fixed_labels: dict[str, str] = field(default_factory=dict)

    @classmethod
    def from_mapping(cls, document):
        doc = _object(
            document,
            {"schema_version", "namespaces", "applications", "limits", "spark_conf", "labels"},
            "policy",
        )
        if type(doc.get("schema_version")) is not int or doc["schema_version"] != 1:
            raise ValueError("Submission policy schema_version must be 1.")
        namespaces = doc.get("namespaces")
        if not isinstance(namespaces, dict) or not namespaces:
            raise ValueError("Policy namespaces must map allowed namespaces to service-account lists.")
        for namespace, accounts in namespaces.items():
            if not isinstance(namespace, str) or not DNS_LABEL.fullmatch(namespace):
                raise ValueError("Policy namespace must be a Kubernetes DNS label.")
            if not all(DNS_LABEL.fullmatch(v) for v in _strings(accounts, "Service accounts")):
                raise ValueError("Policy service accounts must be Kubernetes DNS labels.")
        applications = doc.get("applications")
        if not isinstance(applications, list) or not applications:
            raise ValueError("Policy applications must be a nonempty list.")
        for app in applications:
            _object(
                app,
                {"image", "app_path", "class_name", "arguments", "namespaces", "service_accounts"},
                "approved application",
            )
            _validate_placement(app, namespaces)
            image, path = app.get("image"), app.get("app_path")
            if not isinstance(image, str) or not image or any(c.isspace() for c in image):
                raise ValueError("Approved image must be an exact image reference.")
            if not isinstance(path, str) or not path.startswith("local:///"):
                raise ValueError("Approved app_path must be a local:/// path inside the image.")
            if ".." in PurePosixPath(path[8:]).parts or any(c in path for c in "\x00\r\n?#"):
                raise ValueError("Approved app_path must be a plain absolute local path.")
            if app.get("class_name") is not None and (
                not isinstance(app["class_name"], str) or not app["class_name"].strip()
            ):
                raise ValueError("Approved class_name must be a nonempty string or null.")
            arguments = app.get("arguments", [[]])
            if (
                not isinstance(arguments, list)
                or not arguments
                or any(
                    not isinstance(argv, list) or not all(isinstance(arg, str) for arg in argv)
                    for argv in arguments
                )
            ):
                raise ValueError("Approved arguments must be a nonempty list of exact argument lists.")
        limits = _object(doc.get("limits"), LIMIT_FIELDS | OPTIONAL_LIMIT_FIELDS, "resource limits")
        if not LIMIT_FIELDS <= set(limits) or any(type(v) is not int or v < 1 for v in limits.values()):
            raise ValueError("Every policy resource limit must be a positive integer.")
        validate_max_runtime(limits.get("max_runtime"), "Policy max_runtime limit")
        conf = _object(doc.get("spark_conf", {}), SAFE_SPARK_CONF, "Spark configuration allowlist")
        for values in conf.values():
            if (
                not isinstance(values, list)
                or not values
                or any(type(v) not in {str, int, bool} for v in values)
            ):
                raise ValueError("Allowed Spark configuration values must be explicit scalar lists.")
        required_labels, fixed_labels = _label_policy(doc["labels"]) if "labels" in doc else ({}, {})
        return cls(namespaces, applications, limits, conf, required_labels, fixed_labels)

    def admit(self, document):
        # Import only when enabled. The shared JobSpec owns schema/type validation.
        from spark_on_k8s.jobs import JobSpec

        if not isinstance(document, dict) or set(document) - {"apiVersion", "kind", "metadata", "spec"}:
            raise AdmissionError("Remote jobs cannot select profiles or Kubernetes credentials.")
        spec = document.get("spec")
        # Driver labels are admitted only to supply the policy's required labels.
        label_fields = {"driver_labels"} if self.required_labels else set()
        if not isinstance(spec, dict) or set(spec) - ALLOWED_SPEC - label_fields:
            raise AdmissionError("Remote jobs contain a setting that is not admitted by the submission API.")
        job = JobSpec.from_mapping(document)
        options = job.to_submit_kwargs()
        namespace, account = options["namespace"], options["service_account"]
        if account not in self.namespaces.get(namespace, []):
            raise AdmissionError("The namespace and service-account pair is not approved.")
        matches = [
            app
            for app in self.applications
            if options["image"] == app["image"]
            and options["app_path"] == app["app_path"]
            and options["class_name"] == app.get("class_name")
            and options["app_arguments"] in app.get("arguments", [[]])
        ]
        if not matches:
            raise AdmissionError(
                "The image, application entrypoint, class and argument list are not approved."
            )
        if not any(
            namespace in app.get("namespaces", [namespace])
            and account in app.get("service_accounts", [account])
            for app in matches
        ):
            raise AdmissionError(
                "The application is not approved for this namespace and service-account pair."
            )
        self._admit_labels(options)
        for key, value in options["spark_conf"].items():
            if key not in self.spark_conf or not any(
                type(value) is type(allowed) and value == allowed for allowed in self.spark_conf[key]
            ):
                raise AdmissionError("The Spark configuration key or value is not approved.")
        for role in ("driver", "executor"):
            resources = options[f"{role}_resources"]
            if (
                resources.cpu > self.limits[f"{role}_cpu"]
                or resources.memory + resources.memory_overhead > self.limits[f"{role}_memory_mb"]
            ):
                raise AdmissionError("Requested CPU or total memory exceeds the submission policy.")
        instances = options["executor_instances"]
        if instances.min is not None and instances.max is None:
            raise AdmissionError("Dynamic allocation requires an explicit maximum executor count.")
        if max(instances.initial or 0, instances.min or 0, instances.max or 0) > self.limits["executors"]:
            raise AdmissionError("Requested executor count exceeds the submission policy.")
        runtime_limit = self.limits.get("max_runtime")
        if runtime_limit is not None and (
            options.get("max_runtime") is None or options["max_runtime"] > runtime_limit
        ):
            raise AdmissionError("The submission policy requires max_runtime within its runtime limit.")
        # No operation in the API waits for a job. All defaults are supplied by JobSpec, not ambient CLI env.
        options.update(app_waiter="no_wait", should_print=False)
        return options

    def _admit_labels(self, options):
        """Check required cost labels, then apply them with the fixed labels to driver and executors."""
        labels = options["driver_labels"]
        missing = sorted(set(self.required_labels) - set(labels))
        if missing:
            raise AdmissionError(f"The submission policy requires the driver labels: {', '.join(missing)}.")
        if set(labels) - set(self.required_labels):
            raise AdmissionError("Remote jobs can set only the driver labels the submission policy requires.")
        for key, (values, pattern) in self.required_labels.items():
            value = labels[key]
            if (
                not LABEL_TEXT.fullmatch(value)
                or (values is not None and value not in values)
                or (pattern is not None and not pattern.fullmatch(value))
            ):
                raise AdmissionError(
                    f"The submission policy does not approve the value of driver label {key}."
                )
        labels = {**labels, **self.fixed_labels}
        options.update(driver_labels=labels, executor_labels=dict(labels))