vdk-meta-jobs: exec job with arguments

projects/vdk-plugins/vdk-meta-jobs/README.md

@@ -28,6 +28,7 @@ def run(job_input):
         "job_name": "name-of-job",
         "team_name": "team-of-job",
         "fail_meta_job_on_error": True or False,
+        "arguments": {"key1": value1, "key2": value2},
         "depends_on": [name-of-job1, name-of-job2]
         },
         ...
@@ -38,7 +39,8 @@ def run(job_input):
 When defining a job to be run following attributes are supported:
 * **job_name**: required, the name of the data job
 * **team_name:**: optional, the team of the data job. If omitted , it will use the meta job's team
-* **fail_meta_job_on_error**: optional, default is true. if true, the meta job will abort and fail if the orchestrated job fails, if false, meta job won't fail and continue.
+* **fail_meta_job_on_error**: optional, default is true. If true, the meta job will abort and fail if the orchestrated job fails, if false, meta job won't fail and continue.
+* **arguments**: optional, the arguments that are passed to the underlying orchestrated data job.
 * **depends_on**: required (can be empty), list of other jobs that the orchestrated job depends on. The job will not be started until depends_on job have finished.
 
 
@@ -65,38 +67,44 @@ JOBS_RUN_ORDER = [
         "job_name": "job1",
         "team_name": "team-awesome",
         "fail_meta_job_on_error": True,
+        "arguments": {},
         "depends_on": []
     },
 
     {
         "job_name": "job2",
         "team_name": "team-awesome",
         "fail_meta_job_on_error": True,
+        "arguments": {},
         "depends_on": ["job1"]
     },
     {
         "job_name": "job3",
         "team_name": "team-awesome",
         "fail_meta_job_on_error": True,
+        "arguments": {},
         "depends_on": ["job1"]
     },
     {
         "job_name": "job4",
         "team_name": "team-awesome",
         "fail_meta_job_on_error": True,
+        "arguments": {},
         "depends_on": ["job1"]
     },
 
     {
         "job_name": "job5",
         "team_name": "team-awesome",
         "fail_meta_job_on_error": True,
+        "arguments": {},
         "depends_on": ["job3"]
     },
     {
         "job_name": "job6",
         "team_name": "team-awesome",
         "fail_meta_job_on_error": True,
+        "arguments": {},
         "depends_on": ["job3"]
     },
 ]

projects/vdk-plugins/vdk-meta-jobs/src/vdk/plugin/meta_jobs/api/meta_job.py

@@ -16,6 +16,7 @@ class SingleJob:
     job_name: str
     team_name: str = None
     fail_meta_job_on_error: bool = True
+    arguments: dict = None
     depends_on: List[str] = field(default_factory=list)
 
 

projects/vdk-plugins/vdk-meta-jobs/src/vdk/plugin/meta_jobs/cached_data_job_executor.py

@@ -2,12 +2,12 @@
 # SPDX-License-Identifier: Apache-2.0
 import json
 import logging
-import os
 import time
 from typing import Dict
 from typing import Optional
 
 import urllib3.exceptions as url_exception
+from vdk.api.job_input import IJobArguments
 from vdk.internal.core.errors import ErrorMessage
 from vdk.internal.core.errors import UserCodeError
 from vdk.plugin.meta_jobs.meta import IDataJobExecutor
@@ -44,7 +44,7 @@ def start_job(self, job_name: str) -> None:
         job = self.__get_job(job_name)
         job.start_attempt += 1
         execution_id = self.start_new_job_execution(
-            job_name=job.job_name, team_name=job.team_name
+            job_name=job.job_name, team_name=job.team_name, arguments=job.arguments
         )
         log.info(f"Starting new data job execution with id {execution_id}")
         job.execution_id = execution_id
@@ -135,7 +135,9 @@ def get_all_jobs(self):
     def get_currently_running_jobs(self):
         return [j for j in self._jobs_cache.values() if j.status in ACTIVE_JOB_STATUSES]
 
-    def start_new_job_execution(self, job_name: str, team_name: str) -> str:
+    def start_new_job_execution(
+        self, job_name: str, team_name: str, arguments: IJobArguments = None
+    ) -> str:
         """
         Start a new data job execution.
         The stages of the process are:
@@ -152,6 +154,7 @@ def start_new_job_execution(self, job_name: str, team_name: str) -> str:
 
         :param job_name: name of the data job to be executed
         :param team_name: name of the owning team
+        :param arguments: arguments of the data job
         :return: id of the started job execution
         """
         current_retries = 0
@@ -163,7 +166,7 @@ def start_new_job_execution(self, job_name: str, team_name: str) -> str:
 
         while current_retries < ALLOWED_RETRIES:
             try:
-                execution_id = self._executor.start_job(job_name, team_name)
+                execution_id = self._executor.start_job(job_name, team_name, arguments)
                 return execution_id
             except url_exception.TimeoutError as e:
                 log.info(

projects/vdk-plugins/vdk-meta-jobs/src/vdk/plugin/meta_jobs/dag_validator.py

@@ -1,6 +1,7 @@
 # Copyright 2023-2023 VMware, Inc.
 # SPDX-License-Identifier: Apache-2.0
 import graphlib
+import json
 import logging
 from collections import namedtuple
 from typing import Dict
@@ -14,7 +15,13 @@
 ERROR = Error(
     TYPE="type", PERMISSION="permission", REQUIREMENT="requirement", CONFLICT="conflict"
 )
-allowed_job_keys = {"job_name", "team_name", "fail_meta_job_on_error", "depends_on"}
+allowed_job_keys = {
+    "job_name",
+    "team_name",
+    "fail_meta_job_on_error",
+    "depends_on",
+    "arguments",
+}
 required_job_keys = {"job_name", "depends_on"}
 
 
@@ -66,6 +73,7 @@ def _validate_job(self, job: Dict):
         self._validate_dependencies(job)
         self._validate_team_name(job)
         self._validate_fail_meta_job_on_error(job)
+        self._validate_arguments(job)
         log.info(f"Successfully validated job: {job['job_name']}")
 
     def _validate_allowed_and_required_keys(self, job: Dict):
@@ -141,6 +149,27 @@ def _validate_fail_meta_job_on_error(self, job: Dict):
                 f" is {type(job['fail_meta_job_on_error'])}. Expected type is bool.",
             )
 
+    def _validate_arguments(self, job: Dict):
+        if "arguments" in job:
+            if not isinstance(job["arguments"], dict):
+                self._raise_error(
+                    job["job_name"],
+                    ERROR.TYPE,
+                    "The type of the job dict key arguments is not dict.",
+                    f"Change the Data Job Dict value of arguments. "
+                    f"Current type is {type(job['arguments'])}. Expected type is dict.",
+                )
+            try:
+                json.dumps(job["arguments"])
+            except TypeError as e:
+                self._raise_error(
+                    list(job),
+                    ERROR.TYPE,
+                    str(e),
+                    f"Change the Data Job Dict value of arguments. "
+                    f"Current type is {type(job['arguments'])} but not serializable as JSON.",
+                )
+
     def _check_dag_cycles(self, jobs: List[Dict]):
         topological_sorter = graphlib.TopologicalSorter()
         for job in jobs:

projects/vdk-plugins/vdk-meta-jobs/src/vdk/plugin/meta_jobs/meta.py

@@ -6,14 +6,16 @@
 from typing import Optional
 
 from taurus_datajob_api import DataJobExecution
+from vdk.api.job_input import IJobArguments
 from vdk.plugin.meta_jobs.api import meta_job
 
 
 class IDataJobExecutor(abc.ABC):
     @abc.abstractmethod
-    def start_job(self, job_name: str, team_name: str):
+    def start_job(self, job_name: str, team_name: str, arguments: IJobArguments = None):
         """
         Start an execution of a data job and returns its execution id
+        :param arguments:
         :param job_name:
         :param team_name:
         :return: execution id

projects/vdk-plugins/vdk-meta-jobs/src/vdk/plugin/meta_jobs/meta_dag.py

@@ -22,7 +22,11 @@
 
 
 class MetaJobsDag:
-    def __init__(self, team_name: str, meta_config: MetaPluginConfiguration):
+    def __init__(
+        self,
+        team_name: str,
+        meta_config: MetaPluginConfiguration,
+    ):
         self._team_name = team_name
         self._topological_sorter = TopologicalSorter()
         self._delayed_starting_jobs = TimeBasedQueue(
@@ -49,6 +53,7 @@ def build_dag(self, jobs: List[Dict]):
                 job["job_name"],
                 job.get("team_name", self._team_name),
                 job.get("fail_meta_job_on_error", True),
+                job.get("arguments", None),
             )
             self._job_executor.register_job(trackable_job)
             self._topological_sorter.add(trackable_job.job_name, *job["depends_on"])

projects/vdk-plugins/vdk-meta-jobs/src/vdk/plugin/meta_jobs/remote_data_job.py

@@ -14,6 +14,7 @@
 from taurus_datajob_api import DataJobExecutionRequest
 from taurus_datajob_api import DataJobsExecutionApi
 from urllib3 import Retry
+from vdk.api.job_input import IJobArguments
 from vdk.plugin.control_api_auth.authentication import Authentication
 
 log = logging.getLogger(__name__)
@@ -49,12 +50,14 @@ def __init__(
         job_name: str,
         team_name: str,
         rest_api_url: str,
+        arguments: IJobArguments = None,
         timeout: int = 5,  # TODO: Set reasonable default
         **kwargs,
     ) -> None:
-        self.job_name = job_name
-        self.team_name = team_name
-        self._rest_api_url = rest_api_url
+        self.__job_name = job_name
+        self.__team_name = team_name
+        self.__rest_api_url = rest_api_url
+        self.__arguments = arguments
         self.timeout = timeout
         self.deployment_id = "production"  # currently multiple deployments are not supported so this remains hardcoded
         self.auth: Optional[Authentication] = kwargs.pop("auth", None)
@@ -78,19 +81,19 @@ def __init__(
 
         self.__execution_api = self._get_execution_api()
 
-    def start_job_execution(self, **request_kwargs) -> str:
+    def start_job_execution(self) -> str:
         """
         Triggers a manual Datajob execution.
 
         :param: request_kwargs: Request arguments to be included with the HTTP request
         """
         execution_request = DataJobExecutionRequest(
             started_by="meta-job",  # TODO: specify name of meta job
-            args=request_kwargs,
+            args=self.__arguments,
         )
         _, _, headers = self.__execution_api.data_job_execution_start_with_http_info(
-            team_name=self.team_name,
-            job_name=self.job_name,
+            team_name=self.__team_name,
+            job_name=self.__job_name,
             deployment_id=self.deployment_id,
             data_job_execution_request=execution_request,
             _request_timeout=self.timeout,
@@ -107,8 +110,8 @@ def cancel_job_execution(self, execution_id: str) -> None:
         :param execution_id: ID of the job execution
         """
         self.__execution_api.data_job_execution_cancel(
-            team_name=self.team_name,
-            job_name=self.job_name,
+            team_name=self.__team_name,
+            job_name=self.__job_name,
             execution_id=execution_id,
             _request_timeout=self.timeout,
         )
@@ -121,7 +124,9 @@ def get_job_execution_log(self, execution_id: str) -> str:
         :return: job execution logs
         """
         return self.__execution_api.data_job_logs_download(
-            team_name=self.team_name, job_name=self.job_name, execution_id=execution_id
+            team_name=self.__team_name,
+            job_name=self.__job_name,
+            execution_id=execution_id,
         ).logs
 
     def get_job_execution_status(self, execution_id: str) -> DataJobExecution:
@@ -132,7 +137,9 @@ def get_job_execution_status(self, execution_id: str) -> DataJobExecution:
         :return: The execution status object listing details about the status of this particular execution
         """
         job_execution: DataJobExecution = self.__execution_api.data_job_execution_read(
-            team_name=self.team_name, job_name=self.job_name, execution_id=execution_id
+            team_name=self.__team_name,
+            job_name=self.__job_name,
+            execution_id=execution_id,
         )
         return job_execution
 
@@ -144,7 +151,7 @@ def get_job_executions(self) -> Optional[List[DataJobExecution]]:
         :return: A list of DataJobExecution objects for the available executions.
         """
         job_execution_list = self.__execution_api.data_job_execution_list(
-            team_name=self.team_name, job_name=self.job_name
+            team_name=self.__team_name, job_name=self.__job_name
         )
         return job_execution_list
 
@@ -200,7 +207,7 @@ def wait_for_job(
             return job_status
 
     def _get_execution_api(self):
-        rest_api_url = self._rest_api_url
+        rest_api_url = self.__rest_api_url
 
         config = Configuration(host=rest_api_url, api_key=None)
         config.connection_pool_maxsize = self.http_connection_pool_maxsize

projects/vdk-plugins/vdk-meta-jobs/src/vdk/plugin/meta_jobs/remote_data_job_executor.py

@@ -4,15 +4,18 @@
 from typing import Optional
 
 from taurus_datajob_api import DataJobExecution
+from vdk.api.job_input import IJobArguments
 from vdk.internal.control.configuration.vdk_config import VDKConfig
 from vdk.plugin.meta_jobs.meta import IDataJobExecutor
 from vdk.plugin.meta_jobs.remote_data_job import RemoteDataJob
 
 
 class RemoteDataJobExecutor(IDataJobExecutor):
-    def start_job(self, job_name: str, team_name: str):
+    def start_job(self, job_name: str, team_name: str, arguments: IJobArguments = None):
         vdk_cfg = VDKConfig()
-        job = RemoteDataJob(job_name, team_name, vdk_cfg.control_service_rest_api_url)
+        job = RemoteDataJob(
+            job_name, team_name, vdk_cfg.control_service_rest_api_url, arguments
+        )
         return job.start_job_execution()
         # catch error on 409
 

projects/vdk-plugins/vdk-meta-jobs/tests/jobs/meta-job-arguments/config.ini

@@ -0,0 +1,2 @@
+[owner]
+team = team-awesome

projects/vdk-plugins/vdk-meta-jobs/tests/jobs/meta-job-arguments/meta.py

@@ -0,0 +1,23 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+import logging
+
+from vdk.api.job_input import IJobInput
+from vdk.plugin.meta_jobs.meta_job_runner import MetaJobInput
+
+log = logging.getLogger(__name__)
+
+
+def run(job_input: IJobInput):
+    log.info(f"Dummy arguments {job_input.get_arguments()}")
+
+    job1 = dict(job_name="job1", depends_on=[])
+    job2 = dict(
+        job_name="job2",
+        depends_on=["job1"],
+        fail_meta_job_on_error=False,
+        arguments={"table_name": "test_table", "counter": 123},
+    )
+    job3 = dict(job_name="job3", depends_on=["job1"])
+    job4 = dict(job_name="job4", depends_on=["job2", "job3"])
+    MetaJobInput().run_meta_job([job1, job2, job3, job4])

projects/vdk-plugins/vdk-meta-jobs/tests/jobs/meta-job-empty-arguments/config.ini

@@ -0,0 +1,2 @@
+[owner]
+team = team-awesome

projects/vdk-plugins/vdk-meta-jobs/tests/jobs/meta-job-empty-arguments/meta.py

@@ -0,0 +1,23 @@
+# Copyright 2021-2023 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+import logging
+
+from vdk.api.job_input import IJobInput
+from vdk.plugin.meta_jobs.meta_job_runner import MetaJobInput
+
+log = logging.getLogger(__name__)
+
+
+def run(job_input: IJobInput):
+    log.info(f"Dummy arguments {job_input.get_arguments()}")
+
+    job1 = dict(job_name="job1", depends_on=[])
+    job2 = dict(
+        job_name="job2",
+        depends_on=["job1"],
+        fail_meta_job_on_error=False,
+        arguments={},
+    )
+    job3 = dict(job_name="job3", depends_on=["job1"])
+    job4 = dict(job_name="job4", depends_on=["job2", "job3"])
+    MetaJobInput().run_meta_job([job1, job2, job3, job4])

projects/vdk-plugins/vdk-meta-jobs/tests/jobs/meta-job-wrong-job-arguments-type/config.ini

@@ -0,0 +1,2 @@
+[owner]
+team = team-awesome