support projected service account volumes

requirements-minimal.txt

@@ -21,7 +21,7 @@ pytimeparse
 pytz
 PyYAML>=5.1
 requests
-task_processing[mesos_executor,k8s]
+task_processing[mesos_executor,k8s]>=1.1.0
 Twisted>=19.7.0
 urllib3>=1.24.2
 Werkzeug>=0.15.3

requirements.txt

@@ -80,7 +80,7 @@ setuptools==65.5.1
 six==1.15.0
 sshpubkeys==3.1.0
 stack-data==0.6.2
-task-processing==0.14.0
+task-processing==1.1.0
 traitlets==5.0.0
 Twisted==22.10.0
 typing-extensions==4.5.0

tests/core/actionrun_test.py

@@ -1894,6 +1894,7 @@ def test_recover(self, mock_cluster_repo, mock_filehandler, mock_k8s_action_run)
                 serializer=serializer,
                 volumes=mock_k8s_action_run.command_config.extra_volumes,
                 secret_volumes=mock_k8s_action_run.command_config.secret_volumes,
+                projected_sa_volumes=mock_k8s_action_run.command_config.projected_sa_volumes,
                 cap_add=mock_k8s_action_run.command_config.cap_add,
                 cap_drop=mock_k8s_action_run.command_config.cap_drop,
                 task_id=last_attempt.kubernetes_task_id,

tests/kubernetes_test.py

@@ -7,6 +7,7 @@
 from task_processing.plugins.kubernetes.task_config import KubernetesTaskConfig
 
 from tron.config.schema import ConfigFieldSelectorSource
+from tron.config.schema import ConfigProjectedSAVolume
 from tron.config.schema import ConfigSecretSource
 from tron.config.schema import ConfigSecretVolume
 from tron.config.schema import ConfigSecretVolumeItem
@@ -464,6 +465,7 @@ def test_create_task_disabled():
         env={},
         secret_env={},
         secret_volumes=[],
+        projected_sa_volumes=[],
         field_selector_env={},
         volumes=[],
         cap_add=[],
@@ -493,6 +495,7 @@ def test_create_task(mock_kubernetes_cluster):
         env={},
         secret_env={},
         secret_volumes=[],
+        projected_sa_volumes=[],
         field_selector_env={},
         volumes=[],
         cap_add=[],
@@ -523,6 +526,7 @@ def test_create_task_with_task_id(mock_kubernetes_cluster):
         env={},
         secret_env={},
         secret_volumes=[],
+        projected_sa_volumes=[],
         field_selector_env={},
         volumes=[],
         cap_add=[],
@@ -556,6 +560,7 @@ def test_create_task_with_invalid_task_id(mock_kubernetes_cluster):
             env={},
             secret_env={},
             secret_volumes=[],
+            projected_sa_volumes=[],
             field_selector_env={},
             volumes=[],
             cap_add=[],
@@ -590,6 +595,7 @@ def test_create_task_with_config(mock_kubernetes_cluster):
     ]
     config_secrets = {"TEST_SECRET": ConfigSecretSource(secret_name="tron-secret-test-secret--A", key="secret_A")}
     config_field_selector = {"POD_IP": ConfigFieldSelectorSource(field_path="status.podIP")}
+    config_sa_volumes = [ConfigProjectedSAVolume(audience="for.bar.com", container_path="/var/run/secrets/whatever")]
 
     expected_args = {
         "name": mock.ANY,
@@ -601,6 +607,7 @@ def test_create_task_with_config(mock_kubernetes_cluster):
         "environment": {"TEST_ENV": "foo"},
         "secret_environment": {k: v._asdict() for k, v in config_secrets.items()},
         "secret_volumes": [v._asdict() for v in config_secret_volumes],
+        "projected_sa_volumes": [v._asdict() for v in config_sa_volumes],
         "field_selector_environment": {k: v._asdict() for k, v in config_field_selector.items()},
         "volumes": [v._asdict() for v in default_volumes + config_volumes],
         "cap_add": ["KILL"],
@@ -625,6 +632,7 @@ def test_create_task_with_config(mock_kubernetes_cluster):
         env=expected_args["environment"],
         secret_env=config_secrets,
         secret_volumes=config_secret_volumes,
+        projected_sa_volumes=config_sa_volumes,
         field_selector_env=config_field_selector,
         volumes=config_volumes,
         cap_add=["KILL"],

tron/config/schema.py

@@ -169,6 +169,7 @@ def config_object_factory(name, required=None, optional=None):
         "env",  # dict
         "secret_env",  # dict of str, ConfigSecretSource
         "secret_volumes",  # List of ConfigSecretVolume
+        "projected_sa_volumes",  # List of ConfigProjectedSAVolume
         "field_selector_env",  # dict of str, ConfigFieldSelectorSource
         "extra_volumes",  # List of ConfigVolume
         "expected_runtime",  # datetime.Timedelta
@@ -208,6 +209,7 @@ def config_object_factory(name, required=None, optional=None):
         "env",  # dict
         "secret_env",  # dict of str, ConfigSecretSource
         "secret_volumes",  # List of ConfigSecretVolume
+        "projected_sa_volumes",  # List of ConfigProjectedSAVolume
         "field_selector_env",  # dict of str, ConfigFieldSelectorSource
         "extra_volumes",  # List of ConfigVolume
         "trigger_downstreams",  # None, bool or dict
@@ -252,7 +254,6 @@ def config_object_factory(name, required=None, optional=None):
     optional=["mode"],
 )
 
-
 _ConfigSecretVolume = config_object_factory(
     name="ConfigSecretVolume",
     required=["secret_volume_name", "secret_name", "container_path"],
@@ -283,6 +284,12 @@ def _asdict(self) -> dict:
     optional=[],
 )
 
+ConfigProjectedSAVolume = config_object_factory(
+    name="ConfigProjectedSAVolume",
+    required=["container_path", "audience"],
+    optional=["expiration_seconds"],
+)
+
 ConfigFieldSelectorSource = config_object_factory(
     name="ConfigFieldSelectorSource",
     required=["field_path"],

tron/core/action.py

@@ -11,6 +11,7 @@
 from tron.config.schema import CLEANUP_ACTION_NAME
 from tron.config.schema import ConfigAction
 from tron.config.schema import ConfigNodeAffinity
+from tron.config.schema import ConfigProjectedSAVolume
 from tron.config.schema import ConfigSecretVolume
 
 log = logging.getLogger(__name__)
@@ -33,6 +34,7 @@ class ActionCommandConfig:
     env: dict = field(default_factory=dict)
     secret_env: dict = field(default_factory=dict)
     secret_volumes: List[ConfigSecretVolume] = field(default_factory=list)
+    projected_sa_volumes: List[ConfigProjectedSAVolume] = field(default_factory=list)
     field_selector_env: dict = field(default_factory=dict)
     extra_volumes: set = field(default_factory=set)
     node_selectors: dict = field(default_factory=dict)

tron/core/actionrun.py

@@ -1164,6 +1164,7 @@ def submit_command(self, attempt: ActionRunAttempt) -> Optional[KubernetesTask]:
                 env=build_environment(original_env=attempt.command_config.env, run_id=self.id),
                 secret_env=attempt.command_config.secret_env,
                 secret_volumes=attempt.command_config.secret_volumes,
+                projected_sa_volumes=attempt.command_config.projected_sa_volumes,
                 field_selector_env=attempt.command_config.field_selector_env,
                 serializer=filehandler.OutputStreamSerializer(self.output_path),
                 volumes=attempt.command_config.extra_volumes,
@@ -1239,6 +1240,7 @@ def recover(self) -> Optional[KubernetesTask]:
             field_selector_env=last_attempt.command_config.field_selector_env,
             serializer=filehandler.OutputStreamSerializer(self.output_path),
             secret_volumes=last_attempt.command_config.secret_volumes,
+            projected_sa_volumes=last_attempt.command_config.projected_sa_volumes,
             volumes=last_attempt.command_config.extra_volumes,
             cap_add=last_attempt.command_config.cap_add,
             cap_drop=last_attempt.command_config.cap_drop,

tron/kubernetes.py

@@ -21,6 +21,7 @@
 from tron.config.schema import ConfigFieldSelectorSource
 from tron.config.schema import ConfigKubernetes
 from tron.config.schema import ConfigNodeAffinity
+from tron.config.schema import ConfigProjectedSAVolume
 from tron.config.schema import ConfigSecretSource
 from tron.config.schema import ConfigSecretVolume
 from tron.config.schema import ConfigVolume
@@ -477,6 +478,7 @@ def create_task(
         env: Dict[str, str],
         secret_env: Dict[str, ConfigSecretSource],
         secret_volumes: Collection[ConfigSecretVolume],
+        projected_sa_volumes: Collection[ConfigProjectedSAVolume],
         field_selector_env: Dict[str, ConfigFieldSelectorSource],
         volumes: Collection[ConfigVolume],
         cap_add: Collection[str],
@@ -512,6 +514,7 @@ def create_task(
                 environment=env,
                 secret_environment={k: v._asdict() for k, v in secret_env.items()},
                 secret_volumes=[volume._asdict() for volume in secret_volumes],
+                projected_sa_volumes=[volume._asdict() for volume in projected_sa_volumes],
                 field_selector_environment={k: v._asdict() for k, v in field_selector_env.items()},
                 cap_add=cap_add,
                 cap_drop=cap_drop,