Create Runtime Submission Context (#39)

The submission context infers submitter and submitter's
system information at runtime and attaches these to the
`k8s_annotations`. 

We dropped submission of environment information due
to security concerns.

src/jobs/cli.py

@@ -139,6 +139,6 @@ def main():
     logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO)
 
     args = _make_argparser().parse_args()
-
     job = discover_job(args)
+
     submit_job(job, args)

src/jobs/job.py

@@ -17,6 +17,7 @@
 from jobs.assembler import config
 from jobs.assembler.renderers import RENDERERS
 from jobs.image import Image
+from jobs.submission_context import SubmissionContext
 from jobs.types import AnyPath, K8sResourceKind
 from jobs.utils.helpers import remove_none_values
 from jobs.utils.math import to_rational
@@ -188,11 +189,16 @@ class JobOptions:
 
 class Job(Generic[P, T]):
     def __init__(
-        self, func: Callable[P, T], *, options: JobOptions | None = None
+        self,
+        func: Callable[P, T],
+        *,
+        options: JobOptions | None = None,
+        context: SubmissionContext | None = None,
     ) -> None:
         functools.update_wrapper(self, func)
         self._func = func
         self.options = options
+        self.context = context or SubmissionContext()
 
         if (module := inspect.getmodule(self._func)) is None:
             raise ValueError("Cannot derive module for Job function.")

src/jobs/submission_context.py

@@ -0,0 +1,81 @@
+import logging
+import platform
+import subprocess
+from dataclasses import asdict, dataclass, field
+from typing import Any, Callable, TypeVar, overload
+
+T = TypeVar("T")
+TDefault = TypeVar("TDefault")
+
+
+@overload
+def _maybe(fn: Callable[..., T]) -> T | None: ...
+
+
+@overload
+def _maybe(fn: Callable[..., T], default_value: TDefault) -> T | TDefault: ...
+
+
+def _maybe(
+    fn: Callable[..., T], default_value: TDefault | None = None
+) -> T | TDefault | None:
+    try:
+        return fn()
+    except:  # noqa E722
+        logging.debug(
+            f"Failed to execute {fn.__name__}, using fallback {default_value}"
+        )
+        return default_value
+
+
+def get_git_config_info(item: str) -> str:
+    return _maybe(
+        lambda: subprocess.check_output(
+            ["git", "config", item], encoding="utf-8"
+        ).strip(),
+        "Unknown",
+    )
+
+
+def get_git_username() -> str:
+    return get_git_config_info("user.name")
+
+
+def get_git_user_email() -> str:
+    return get_git_config_info("user.email")
+
+
+def determine_platform_info() -> dict[str, Any]:
+    skip = ["system_alias"]
+    return {
+        name: _maybe(fn)
+        for name, fn in platform.__dict__.items()
+        if callable(fn) and not name.startswith("_") and name not in skip
+    }
+
+
+@dataclass
+class SubmitterInformation:
+    username: str = field(default_factory=lambda: get_git_config_info("user.name"))
+    email: str = field(default_factory=lambda: get_git_config_info("user.email"))
+
+
+@dataclass
+class SubmissionContext:
+    submitter: SubmitterInformation = field(default_factory=SubmitterInformation)
+    platform_info: dict[str, Any] = field(default_factory=determine_platform_info)
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "submitter": asdict(self.submitter),
+            "platform": self.platform_info,
+        }
+
+    def resolve(self) -> dict[str, str]:
+        result = {}
+        for key, value in self.to_dict().items():
+            if isinstance(value, dict):
+                result.update({f"{key}.{k}": str(v) for k, v in value.items()})
+            else:
+                result[key] = str(value)
+        return result

src/jobs/utils/kubernetes.py

@@ -16,10 +16,10 @@ def sanitize_rfc1123_domain_name(s: str) -> str:
 
 def k8s_annotations(job: Job) -> dict[str, str]:
     """Determine the Kubernetes annotations for a Job"""
-    if not job.options:
-        return {}
     # Store as annotations since labels have restrictive value formats
-    return job.options.labels
+    options = job.options.labels if job.options else {}
+    context = job.context.resolve() if job.context else {}
+    return options | context
 
 
 class KubernetesNamespaceMixin: