just support project SA volumes from config

tests/core/actionrun_test.py

@@ -1861,10 +1861,9 @@ def test_handler_exiting_failstart_failed(self, mock_k8s_action_run):
         )
         assert mock_k8s_action_run.is_failed
 
-    @mock.patch("tron.core.actionrun.get_projected_sa_volumes", autospec=True)
     @mock.patch("tron.core.actionrun.filehandler", autospec=True)
     @mock.patch("tron.core.actionrun.KubernetesClusterRepository", autospec=True)
-    def test_recover(self, mock_cluster_repo, mock_filehandler, mock_get_projected_sa_volumes, mock_k8s_action_run):
+    def test_recover(self, mock_cluster_repo, mock_filehandler, mock_k8s_action_run):
         mock_k8s_action_run.machine.state = ActionRun.UNKNOWN
         mock_k8s_action_run.end_time = 1000
         mock_k8s_action_run.exit_status = 0
@@ -1895,7 +1894,7 @@ def test_recover(self, mock_cluster_repo, mock_filehandler, mock_get_projected_s
                 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_get_projected_sa_volumes.return_value,
+                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,
@@ -1918,7 +1917,6 @@ def test_recover(self, mock_cluster_repo, mock_filehandler, mock_get_projected_s
         mock_filehandler.OutputStreamSerializer.assert_called_with(
             mock_k8s_action_run.output_path,
         )
-        mock_get_projected_sa_volumes.assert_called_once_with("mock_namespace")
 
     @mock.patch("tron.core.actionrun.filehandler", autospec=True)
     @mock.patch("tron.core.actionrun.MesosClusterRepository", autospec=True)
@@ -1953,12 +1951,9 @@ def test_recover_no_k8s_task_id(
         assert mock_k8s_action_run.is_unknown
         assert mock_k8s_action_run.end_time is not None
 
-    @mock.patch("tron.core.actionrun.get_projected_sa_volumes", autospec=True)
     @mock.patch("tron.core.actionrun.filehandler", autospec=True)
     @mock.patch("tron.core.actionrun.KubernetesClusterRepository", autospec=True)
-    def test_recover_task_none(
-        self, mock_cluster_repo, mock_filehandler, mock_get_projected_sa_volumes, mock_k8s_action_run
-    ):
+    def test_recover_task_none(self, mock_cluster_repo, mock_filehandler, mock_k8s_action_run):
         mock_k8s_action_run.machine.state = ActionRun.UNKNOWN
         last_attempt = mock_k8s_action_run.create_attempt()
         last_attempt.kubernetes_task_id = "test-kubernetes-task-id"
@@ -1971,7 +1966,6 @@ def test_recover_task_none(
         assert mock_k8s_action_run.is_unknown
         assert mock_get_cluster.return_value.recover.call_count == 0
         assert mock_k8s_action_run.end_time is not None
-        mock_get_projected_sa_volumes.assert_called_once_with("mock_namespace")
 
     @mock.patch("tron.core.actionrun.KubernetesClusterRepository", autospec=True)
     def test_kill_task_k8s(self, mock_cluster_repo, mock_k8s_action_run):

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
@@ -594,7 +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")}
-    arg_sa_volumes = [{"audience": "for.bar.com", "container_path": "/var/run/secrets/whatever"}]
+    config_sa_volumes = [ConfigProjectedSAVolume(audience="for.bar.com", container_path="/var/run/secrets/whatever")]
 
     expected_args = {
         "name": mock.ANY,
@@ -606,7 +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": arg_sa_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"],
@@ -631,7 +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=arg_sa_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

@@ -33,7 +33,6 @@
 from tron.utils import maybe_decode
 from tron.utils import proxy
 from tron.utils import timeutils
-from tron.utils.authentication import get_projected_sa_volumes
 from tron.utils.observer import Observable
 from tron.utils.observer import Observer
 from tron.utils.state import Machine
@@ -348,10 +347,6 @@ def id(self):
     def name(self):
         return self.action_name
 
-    @property
-    def service_name(self):
-        return self.job_run_id.split(".", 1)[0]
-
     @property
     def last_attempt(self):
         if self.attempts:
@@ -1169,7 +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=get_projected_sa_volumes(self.service_name),
+                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,
@@ -1245,7 +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=get_projected_sa_volumes(self.service_name),
+            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

@@ -9,7 +9,6 @@
 
 from task_processing.interfaces.event import Event
 from task_processing.plugins.kubernetes.task_config import KubernetesTaskConfig
-from task_processing.plugins.kubernetes.types import ProjectedSAVolume
 from task_processing.runners.subscription import Subscription
 from task_processing.task_processor import TaskProcessor
 from twisted.internet.defer import Deferred
@@ -22,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
@@ -478,7 +478,7 @@ def create_task(
         env: Dict[str, str],
         secret_env: Dict[str, ConfigSecretSource],
         secret_volumes: Collection[ConfigSecretVolume],
-        projected_sa_volumes: List[ProjectedSAVolume],
+        projected_sa_volumes: Collection[ConfigProjectedSAVolume],
         field_selector_env: Dict[str, ConfigFieldSelectorSource],
         volumes: Collection[ConfigVolume],
         cap_add: Collection[str],
@@ -514,7 +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=projected_sa_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,