Merge pull request #245 from opensafely-core/jobrunner-v2.73.0

Update to jobrunner v2.73.0, including pipeline

opensafely/_vendor/jobrunner/actions.py

@@ -4,6 +4,7 @@
 from typing import Dict, List
 
 from opensafely._vendor.pipeline.exceptions import ProjectValidationError
+from opensafely._vendor.pipeline.models import Action
 from opensafely._vendor.pipeline.outputs import get_output_dirs
 
 from opensafely._vendor.jobrunner.lib.path_utils import ensure_unix_path
@@ -19,6 +20,7 @@ class ActionSpecification:
  run: str
  needs: List[str]
  outputs: Dict[str, Dict[str, str]]
+ action: Action
 
 
 def get_action_specification(config, action_id, using_dummy_data_backend=False):
@@ -80,6 +82,7 @@ def get_action_specification(config, action_id, using_dummy_data_backend=False):
  run=run_command,
  needs=action_spec.needs,
  outputs=action_spec.outputs.dict(exclude_unset=True),
+ action=action_spec,
  )
 
 

opensafely/_vendor/jobrunner/cli/add_job.py

@@ -5,6 +5,7 @@
 import argparse
 import dataclasses
 import pprint
+import sys
 import textwrap
 from pathlib import Path
 from urllib.parse import urlparse
@@ -36,6 +37,7 @@ def main(
  requested_actions=actions,
  force_run_dependencies=force_run_dependencies,
  cancelled_actions=[],
+ codelists_ok=True,
  )
  )
  print("Submitting JobRequest:\n")
@@ -46,6 +48,8 @@ def main(
  for job in jobs:
  display_obj(job)
 
+ return job_request, jobs
+
 
 def display_obj(obj):
  if hasattr(obj, "asdict"):
@@ -57,7 +61,10 @@ def display_obj(obj):
  print()
 
 
-def run():
+def run(argv=None):
+ if argv is None:
+ argv = sys.argv[1:]
+
  configure_logging()
  parser = argparse.ArgumentParser(description=__doc__.partition("\n\n")[0])
  parser.add_argument("repo_url", help="URL (or local path) of git repository")
@@ -82,8 +89,8 @@ def run():
  )
  parser.add_argument("-f", "--force-run-dependencies", action="store_true")
 
- args = parser.parse_args()
- main(**vars(args))
+ args = parser.parse_args(argv)
+ return main(**vars(args))
 
 
 if __name__ == "__main__":

opensafely/_vendor/jobrunner/cli/kill_job.py

@@ -4,6 +4,7 @@
 import argparse
 
 from opensafely._vendor.jobrunner.executors import local
+from opensafely._vendor.jobrunner.job_executor import JobResults
 from opensafely._vendor.jobrunner.lib import database, docker
 from opensafely._vendor.jobrunner.models import Job, State, StatusCode
 from opensafely._vendor.jobrunner.run import job_to_job_definition, mark_job_as_failed
@@ -32,7 +33,16 @@ def main(partial_job_ids, cleanup=False):
  )
  if container_metadata:
  job = job_to_job_definition(job)
- metadata = local.get_job_metadata(job, {}, container_metadata)
+ # create a dummy JobResults with just the message we want
+ results = JobResults(
+ outputs=None,
+ unmatched_patterns=None,
+ unmatched_outputs=None,
+ exit_code=container_metadata["State"]["ExitCode"],
+ image_id=container_metadata["Image"],
+ message="job killed by OpenSAFELY administrator",
+ )
+ metadata = local.get_job_metadata(job, {}, container_metadata, results)
  local.write_job_logs(job, metadata, copy_log_to_workspace=False)
 
  if cleanup:

opensafely/_vendor/jobrunner/config.py

@@ -6,6 +6,8 @@
 from multiprocessing import cpu_count
 from pathlib import Path
 
+from opensafely._vendor import pipeline
+
 
 class ConfigException(Exception):
  pass
@@ -46,6 +48,7 @@ def _is_valid_backend_name(name):
 
 WORKDIR = Path(os.environ.get("WORKDIR", default_work_dir)).resolve()
 DATABASE_FILE = WORKDIR / "db.sqlite"
+METRICS_FILE = WORKDIR / "metrics.sqlite"
 GIT_REPO_DIR = WORKDIR / "repos"
 
 # valid archive formats
@@ -143,25 +146,7 @@ def database_urls_from_env(env):
 ) # 16mb
 
 
-# TODO: we might want to take this list from pipeline if we implement it there.
-LEVEL4_FILE_TYPES = [
- # tables
- ".csv",
- ".tsv",
- # images
- ".jpg",
- ".jpeg",
- ".png",
- ".svg",
- ".svgz",
- # reports
- ".html",
- ".pdf",
- ".txt",
- ".log",
- ".json",
- ".md",
-]
+LEVEL4_FILE_TYPES = pipeline.constants.LEVEL4_FILE_TYPES
 
 STATA_LICENSE = os.environ.get("STATA_LICENSE")
 STATA_LICENSE_REPO = os.environ.get(

opensafely/_vendor/jobrunner/create_or_update_jobs.py

@@ -36,6 +36,10 @@ class JobRequestError(Exception):
  pass
 
 
+class StaleCodelistError(JobRequestError):
+ pass
+
+
 class NothingToDoError(JobRequestError):
  pass
 
@@ -62,10 +66,10 @@ def create_or_update_jobs(job_request):
  JobRequestError,
  ) as e:
  log.info(f"JobRequest failed:\n{e}")
- create_failed_job(job_request, e)
+ create_job_from_exception(job_request, e)
  except Exception:
  log.exception("Uncaught error while creating jobs")
- create_failed_job(job_request, JobRequestError("Internal error"))
+ create_job_from_exception(job_request, JobRequestError("Internal error"))
  else:
  if job_request.cancelled_actions:
  log.debug("Cancelling actions: %s", job_request.cancelled_actions)
@@ -114,7 +118,8 @@ def create_jobs(job_request):
 
 
 def validate_job_request(job_request):
- if config.ALLOWED_GITHUB_ORGS:
+ # http prefix allows local git repos, useful for tests
+ if job_request.repo_url.startswith("http") and config.ALLOWED_GITHUB_ORGS:
  validate_repo_url(job_request.repo_url, config.ALLOWED_GITHUB_ORGS)
  if not job_request.requested_actions:
  raise JobRequestError("At least one action must be supplied")
@@ -238,6 +243,7 @@ def recursively_build_jobs(jobs_by_action, job_request, pipeline_config, action)
  commit=job_request.commit,
  workspace=job_request.workspace,
  database_name=job_request.database_name,
+ requires_db=action_spec.action.is_database_action,
  action=action,
  wait_for_job_ids=wait_for_job_ids,
  requires_outputs_from=action_spec.needs,
@@ -311,12 +317,12 @@ def assert_codelists_ok(job_request, new_jobs):
  # Codelists are out of date; fail the entire job request if any job
  # requires database access
  if job.requires_db:
- raise JobRequestError(
+ raise StaleCodelistError(
  f"Codelists are out of date (required by action {job.action})"
  )
 
 
-def create_failed_job(job_request, exception):
+def create_job_from_exception(job_request, exception):
  """
  Sometimes we want to say to the job-server (and the user): your JobRequest
  was broken so we weren't able to create any jobs for it. But the only way
@@ -327,19 +333,25 @@ def create_failed_job(job_request, exception):
 
  This is a bit of a hack, but it keeps the sync protocol simple.
  """
+ action = "__error__"
+ error = exception
+ state = State.FAILED
+ status_message = str(exception)
+
  # Special case for the NothingToDoError which we treat as a success
  if isinstance(exception, NothingToDoError):
  state = State.SUCCEEDED
  code = StatusCode.SUCCEEDED
- status_message = "All actions have already run"
  action = job_request.requested_actions[0]
  error = None
+ # StaleCodelistError is a failure but not an INTERNAL_ERROR
+ elif isinstance(exception, StaleCodelistError):
+ code = StatusCode.STALE_CODELISTS
  else:
- state = State.FAILED
  code = StatusCode.INTERNAL_ERROR
+ # include exception name in message to aid debugging
  status_message = f"{type(exception).__name__}: {exception}"
- action = "__error__"
- error = exception
+
  now = time.time()
  job = Job(
  job_request_id=job_request.id,
@@ -379,7 +391,10 @@ def set_cancelled_flag_for_actions(job_request_id, actions):
  # working.
  update_where(
  Job,
- {"cancelled": True},
+ {
+ "cancelled": True,
+ "completed_at": int(time.time()),
+ },
  job_request_id=job_request_id,
  action__in=actions,
  )

opensafely/_vendor/jobrunner/executors/local.py

@@ -11,7 +11,7 @@
 
 from opensafely._vendor.pipeline.legacy import get_all_output_patterns_from_project_file
 
-from opensafely._vendor.jobrunner import config
+from opensafely._vendor.jobrunner import config, record_stats
 from opensafely._vendor.jobrunner.executors import volumes
 from opensafely._vendor.jobrunner.job_executor import (
  ExecutorAPI,
@@ -241,16 +241,24 @@ def get_status(self, job_definition, timeout=15):
  f"docker timed out after {timeout}s inspecting container {name}"
  )
 
+ metrics = record_stats.read_job_metrics(job_definition.id)
+
  if container is None: # container doesn't exist
  if job_definition.cancelled:
  if volumes.get_volume_api(job_definition).volume_exists(job_definition):
  # jobs prepared but not running do not need to finalize, so we
  # proceed directly to the FINALIZED state here
  return JobStatus(
- ExecutorState.FINALIZED, "Prepared job was cancelled"
+ ExecutorState.FINALIZED,
+ "Prepared job was cancelled",
+ metrics=metrics,
  )
  else:
- return JobStatus(ExecutorState.UNKNOWN, "Pending job was cancelled")
+ return JobStatus(
+ ExecutorState.UNKNOWN,
+ "Pending job was cancelled",
+ metrics=metrics,
+ )
 
  # timestamp file presence means we have finished preparing
  timestamp_ns = volumes.get_volume_api(job_definition).read_timestamp(
@@ -261,24 +269,31 @@ def get_status(self, job_definition, timeout=15):
  # re-prepare it anyway.
  if timestamp_ns is None:
  # we are Jon Snow
- return JobStatus(ExecutorState.UNKNOWN)
+ return JobStatus(ExecutorState.UNKNOWN, metrics={})
  else:
  # we've finish preparing
- return JobStatus(ExecutorState.PREPARED, timestamp_ns=timestamp_ns)
+ return JobStatus(
+ ExecutorState.PREPARED, timestamp_ns=timestamp_ns, metrics=metrics
+ )
 
  if container["State"]["Running"]:
  timestamp_ns = datestr_to_ns_timestamp(container["State"]["StartedAt"])
- return JobStatus(ExecutorState.EXECUTING, timestamp_ns=timestamp_ns)
+ return JobStatus(
+ ExecutorState.EXECUTING, timestamp_ns=timestamp_ns, metrics=metrics
+ )
  elif job_definition.id in RESULTS:
  return JobStatus(
  ExecutorState.FINALIZED,
  timestamp_ns=RESULTS[job_definition.id].timestamp_ns,
+ metrics=metrics,
  )
  else:
  # container present but not running, i.e. finished
  # Nb. this does not include prepared jobs, as they have a volume but not a container
  timestamp_ns = datestr_to_ns_timestamp(container["State"]["FinishedAt"])
- return JobStatus(ExecutorState.EXECUTED, timestamp_ns=timestamp_ns)
+ return JobStatus(
+ ExecutorState.EXECUTED, timestamp_ns=timestamp_ns, metrics=metrics
+ )
 
  def get_results(self, job_definition):
  if job_definition.id not in RESULTS:
@@ -409,7 +424,9 @@ def finalize_job(job_definition):
  base_revision=labels.get("org.opensafely.base.vcs-ref", "unknown"),
  base_created=labels.get("org.opencontainers.base.build-date", "unknown"),
  )
- job_metadata = get_job_metadata(job_definition, outputs, container_metadata)
+ job_metadata = get_job_metadata(
+ job_definition, outputs, container_metadata, results
+ )
 
  if job_definition.cancelled:
  write_job_logs(job_definition, job_metadata, copy_log_to_workspace=False)
@@ -426,7 +443,7 @@ def finalize_job(job_definition):
  return results
 
 
-def get_job_metadata(job_definition, outputs, container_metadata):
+def get_job_metadata(job_definition, outputs, container_metadata, results):
  # job_metadata is a big dict capturing everything we know about the state
  # of the job
  job_metadata = dict()
@@ -437,6 +454,7 @@ def get_job_metadata(job_definition, outputs, container_metadata):
  job_metadata["docker_image_id"] = container_metadata["Image"]
  # convert exit code to str so 0 exit codes get logged
  job_metadata["exit_code"] = str(container_metadata["State"]["ExitCode"])
+ job_metadata["status_message"] = results.message
  job_metadata["container_metadata"] = container_metadata
  job_metadata["outputs"] = outputs
  job_metadata["commit"] = job_definition.study.commit
@@ -679,6 +697,7 @@ def write_log_file(job_definition, job_metadata, filename, excluded):
  "commit",
  "docker_image_id",
  "exit_code",
+ "status_message",
  "created_at",
  "completed_at",
  "database_name",

opensafely/_vendor/jobrunner/job_executor.py

@@ -67,6 +67,7 @@ class JobStatus:
  timestamp_ns: int = (
  None # timestamp this JobStatus occurred, in integer nanoseconds
  )
+ metrics: dict = field(default_factory=dict)
 
 
 @dataclass

opensafely/_vendor/jobrunner/lib/docker_stats.py

@@ -30,6 +30,7 @@ def get_container_stats(timeout=DEFAULT_TIMEOUT):
  removeprefix(row["Name"], "os-job-"): {
  "cpu_percentage": float(row["CPUPerc"].rstrip("%")),
  "memory_used": _parse_size(row["MemUsage"].split()[0]),
+ "container_id": row["Container"],
  }
  for row in data
  if row["Name"].startswith("os-job-")