feature(service/planner_service): Trigger jenkins jobs by plans

This commit adds a new feature to the Release Planning API - allowing
API consumers to trigger jobs from within argus in Jenkins. The endpoint
accepts target version (for example "6.2.0") in which case it will
trigger all plans with this version (release independent), release name
(will trigger all plans within that relase) or specific plan id (will
only trigger that plan). The API client provides the parameters in two
parts: as a set of common parameters intended for each job (useful for
things like provision_type) and an additional per job
type/backend/region set that each job will require in order to trigger.
Currently, only longevity is supported.

Fixes #377

argus/backend/controller/planner_api.py

@@ -176,3 +176,16 @@ def resolve_plan_entities(plan_id: str):
         "status": "ok",
         "response": result,
     }
+
+@bp.route("/plan/trigger", methods=["POST"])
+@api_login_required
+def trigger_jobs_for_plans():
+
+    payload = get_payload(request)
+    service = PlanningService()
+    result = service.trigger_jobs(payload)
+
+    return {
+        "status": "ok",
+        "response": result,
+    }

argus/backend/models/plan.py

@@ -12,7 +12,7 @@ class ArgusReleasePlan(Model):
     description = columns.Text()
     owner = columns.UUID(required=True)
     participants = columns.List(value_type=columns.UUID)
-    target_version = columns.Ascii()
+    target_version = columns.Ascii(index=True)
     assignee_mapping = columns.Map(key_type=columns.UUID, value_type=columns.UUID)
     release_id = columns.UUID(index=True)
     tests = columns.List(value_type=columns.UUID)

argus/backend/models/web.py

@@ -178,6 +178,7 @@ class ArgusTest(Model):
     enabled = columns.Boolean(default=lambda: True)
     build_system_url = columns.Text()
     plugin_name = columns.Text()
+    plugin_subtype = columns.Text()
 
     def __eq__(self, other):
         if isinstance(other, ArgusTest):

argus/backend/service/jenkins_service.py

@@ -69,6 +69,17 @@ def retrieve_job_parameters(self, build_id: str, build_number: int) -> list[Para
 
         return params
 
+    def latest_build(self, build_id: str) -> int:
+        try:
+            job_info = self._jenkins.get_job_info(name=build_id)
+            last_build = job_info.get("lastBuild")
+            if not last_build:
+                return -1
+            return last_build["number"]
+        except jenkins.JenkinsException:
+            raise JenkinsServiceError("Job doesn't exist", build_id)
+
+
     def get_releases_for_clone(self, test_id: str):
         test_id = UUID(test_id)
         # TODO: Filtering based on origin location / user preferences

argus/backend/service/planner_service.py

@@ -6,14 +6,17 @@
 from copy import deepcopy
 from dataclasses import dataclass
 from functools import reduce
-from typing import Any, Optional
+from typing import Any, Optional, TypedDict
 from uuid import UUID
+from flask import g
 from slugify import slugify
 
 from argus.backend.models.plan import ArgusReleasePlan
 from argus.backend.models.web import ArgusGroup, ArgusRelease, ArgusTest, ArgusUserView, User
+from argus.backend.service.jenkins_service import JenkinsService
 from argus.backend.service.test_lookup import TestLookup
 from argus.backend.service.views import UserViewService
+from argus.backend.util.common import chunk
 
 
 LOGGER = logging.getLogger(__name__)
@@ -60,6 +63,13 @@ class CopyPlanPayload:
     targetReleaseId: str
     targetReleaseName: str
 
+class PlanTriggerPayload(TypedDict):
+    plan_id: str | None
+    release: str | None
+    version: str | None
+    common_params: dict[str, str]
+    params: list[dict[str, str]]
+
 
 class PlannerServiceException(Exception):
     pass
@@ -438,3 +448,92 @@ def resolve_plan(self, plan_id: str | UUID) -> list[dict[str, Any]]:
                 ent["group"] = group.pretty_name or group.name
 
         return mapped
+
+    def trigger_jobs(self, payload: PlanTriggerPayload) -> bool:
+
+        release_name = payload.get("release")
+        plan_id = payload.get("plan_id")
+        version = payload.get("version")
+
+        condition_set = (bool(release_name), bool(plan_id), bool(version))
+
+        match condition_set:
+            case (True, False, False):
+                release = ArgusRelease.get(name=release_name)
+                filter_expr = { "release_id__eq": release.id }
+            case (False, True, False):
+                filter_expr = { "id__eq": plan_id }
+            case (False, False, True):
+                filter_expr = { "target_version__eq": version }
+            case (True, False, True):
+                release = ArgusRelease.get(name=release_name)
+                filter_expr = { "target_version__eq": version, "release_id__eq": release.id }
+            case _:
+                raise PlannerServiceException("No version, release name or plan id specified.", payload)
+
+        plans: list[ArgusReleasePlan] = list(ArgusReleasePlan.filter(**filter_expr).allow_filtering().all())
+
+        if len(plans) == 0:
+            return False, "No plans to trigger"
+
+        common_params = payload.get("common_params", {})
+        params = payload.get("params", [])
+        test_ids = [test_id for plan in plans for test_id in plan.tests]
+        group_ids = [group_id for plan in plans for group_id in plan.groups]
+
+        tests = []
+        for batch in chunk(test_ids):
+            tests.extend(ArgusTest.filter(id__in=batch).all())
+
+        for batch in (chunk(group_ids)):
+            tests.extend(ArgusTest.filter(group_id__in=batch).allow_filtering().all())
+
+        tests = list({test for test in tests})
+
+        LOGGER.info("Will trigger %s tests...", len(tests))
+
+        service = JenkinsService()
+        failures = []
+        successes = []
+        for test in tests:
+            try:
+                latest_build_number = service.latest_build(test.build_system_id)
+                if latest_build_number == -1:
+                    failures.append(test.build_system_id)
+                    continue
+                raw_params = service.retrieve_job_parameters(test.build_system_id, latest_build_number)
+                job_params = { param["name"]: param["value"] for param in raw_params if param.get("value") }
+                backend = job_params.get("backend")
+                match backend.split("-"):
+                    case ["aws", *_]:
+                        region_key = "region"
+                    case ["gce", *_]:
+                        region_key = "gce_datacenter"
+                    case ["azure", *_]:
+                        region_key = "azure_region_name"
+                    case _:
+                        raise PlannerServiceException(f"Unknown backend encountered: {backend}", backend)
+
+                job_params = None
+                for param_set in params:
+                    if param_set["test"] == "longevity" and backend == param_set["backend"]:
+                        job_params = dict(param_set)
+                        job_params.pop("type", None)
+                        region = job_params.pop("region", None)
+                        job_params[region_key] = region
+                        break
+                if not job_params:
+                    raise PlannerServiceException(f"Parameters not found for job {test.build_system_id}", test.build_system_id)
+                final_params = { **job_params, **common_params, **job_params}
+                queue_item = service.build_job(test.build_system_id, final_params, g.user.username)
+                info = service.get_queue_info(queue_item)
+                url = info.get("url", info.get("taskUrl", ""))
+                successes.append(url)
+            except Exception:
+                LOGGER.error("Failed to trigger %s", test.build_system_id, exc_info=True)
+                failures.append(test.build_system_id)
+
+        return {
+            "jobs": successes,
+            "failed_to_execute": failures,
+        }

argus/client/generic/cli.py

@@ -1,3 +1,5 @@
+import json
+from pathlib import Path
 import click
 import logging
 
@@ -39,8 +41,26 @@ def finish_run(api_key: str, base_url: str, id: str, status: str, scylla_version
     client.finalize_generic_run(run_id=id, status=status, scylla_version=scylla_version)
 
 
+@click.command("trigger-jobs")
+@click.option("--api-key", help="Argus API key for authorization", required=True)
+@click.option("--base-url", default="https://argus.scylladb.com", help="Base URL for argus instance")
+@click.option("--version", help="Scylla version to filter plans by", default=None, required=False)
+@click.option("--plan-id", help="Specific plan id for filtering", default=None, required=False)
+@click.option("--release", help="Release name to filter plans by", default=None, required=False)
+@click.option("--job-info-file", required=True, help="JSON file with trigger information (see detailed docs)")
+def trigger_jobs(api_key: str, base_url: str, job_info_file: str, version: str, plan_id: str, release: str):
+    client = ArgusGenericClient(auth_token=api_key, base_url=base_url)
+    path = Path(job_info_file)
+    if not path.exists():
+        LOGGER.error("File not found: %s", job_info_file)
+        exit(128)
+    payload = json.load(path.open("rt", encoding="utf-8"))
+    client.trigger_jobs({ "release": release, "version": version, "plan_id": plan_id, **payload })
+
+
 cli.add_command(submit_run)
 cli.add_command(finish_run)
+cli.add_command(trigger_jobs)
 
 
 if __name__ == "__main__":

argus/client/generic/client.py

@@ -7,6 +7,10 @@
 class ArgusGenericClient(ArgusClient):
     test_type = "generic"
     schema_version: None = "v1"
+
+    class Routes(ArgusClient.Routes):
+        TRIGGER_JOBS = "/planning/plan/trigger"
+
     def __init__(self, auth_token: str, base_url: str, api_version="v1") -> None:
         super().__init__(auth_token, base_url, api_version)
 
@@ -22,6 +26,24 @@ def submit_generic_run(self, build_id: str, run_id: str, started_by: str, build_
         response = self.submit_run(run_type=self.test_type, run_body=request_body)
         self.check_response(response)
 
+    def trigger_jobs(self, common_params: dict[str, str], params: list[dict[str, str]], version: str = None, release: str = None, plan_id: str = None):
+        request_body = {
+            "common_params": common_params,
+            "params": params,
+            "version": version,
+            "release": release,
+            "plan_id": plan_id,
+        }
+        response = self.post(
+            endpoint=self.Routes.TRIGGER_JOBS,
+            location_params={},
+            body={
+                **self.generic_body,
+                **request_body,
+            }
+        )
+        self.check_response(response)
+        return response.json()
 
     def finalize_generic_run(self, run_id: str, status: str, scylla_version: str | None = None):
         response = self.finalize_run(run_type=self.test_type, run_id=run_id, body={

docs/api_usage.md

@@ -44,3 +44,77 @@ Accepts following parameters:
   "status": "ok"
 }
 ```
+
+```http
+POST /api/v1/planning/plan/trigger
+```
+
+Accepts following payload:
+
+Type: application/json
+
+| Parameter | Type | Description |
+| --------- | ---- | ------------|
+| release          | string     | Release name to trigger plans in (can be mixed vith version to narrow filter)            |
+| plan_id          | string     | Specific plan id to trigger            |
+| version          | string     | Target scylla version of plans to trigger             |
+| common_params          | object{ param_name: value }     | Common parameters, such as backend            |
+| params          | object{ param_name: value }[]     | specific job parameters            |
+
+Example payload:
+
+```json
+{
+  {
+  "version": "2024.3.1~rc0",
+  "release": "scylla-master",
+  "plan_id": "some-plan-uuid",
+  "common_params": {
+      "instance_provision": "spot",
+  },
+  "params": [
+      {
+          "test": "longevity",
+          "backend": "aws",
+          "region": "eu-west-1",
+          "scylla_ami_id": "ami-abcd"
+      },
+      {
+          "test": "longevity",
+          "backend": "azure",
+          "region": "us-east1",
+          "azure_image_id": "/subscriptions…",
+      }
+  ]
+  }
+
+}
+```
+
+```json
+{
+  "response": {
+    "jobs": [
+      "http://jenkins/path/job/one/1",
+      "http://jenkins/path/job/two/3",
+      "http://jenkins/path/job/three/5",
+      "http://jenkins/path/job/four/9"
+    ],
+    "failed_to_execute": [
+      "path/job/one",
+      "path/job/two",
+      "path/job/three",
+      "path/job/four"
+    ]
+  }, 
+  "status": "ok"
+}
+```
+
+Additionally, this endpoint is available inside `argus-client-generic` executable, as follows:
+
+```bash
+argus-client-generic trigger-jobs --api-key $key --version $version --plan_id $id --release $release --job-info-file $file_path
+```
+
+`--job-info-file` is a .json file containing `common_params` and `params` parts of the payload, everything else is specified on the command line.