Implement Job Metrics and its APIs

.gitignore

@@ -2,6 +2,12 @@
 .DS_Store
 .vscode/
 
+
+# local development
+initialize_db.sh
+insert_resource.sql
+prepare_test_db.sql
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

bqp_database_access/__init__.py

@@ -1,4 +1,4 @@
-from . import budgets, feedback, jobs, resources, status, tokens, users
+from . import budgets, feedback, jobs, resources, status, tokens, users, job_metrics
 
 
 def create_app():

bqp_database_access/_database.py

@@ -6,6 +6,7 @@
 from datetime import datetime
 from pony.orm import Database, Optional, PrimaryKey, Required, Set  # type: ignore
 
+
 # if os.getenv("QUANTUM_DB_TESTING") is not None:
 #     db_config.set_test_env()
 
@@ -86,8 +87,7 @@ class UserGroup(database.Entity):
 
         name = PrimaryKey(str, auto=False)
         note = Optional(str)
-        users = Set("User", table="users_in_user_groups",
-                    reverse="user_groups")
+        users = Set("User", table="users_in_user_groups", reverse="user_groups")
         budgets = Set("Budget", table="user_groups_in_budgets")
         time_slots = Set("TimeSlots", table="user_groups_in_time_slots")
         owner = Required("User")
@@ -151,6 +151,87 @@ class CircuitJob(database.Entity):
 
         queued = Required(bool, default=False)
 
+        timestamp_data = Optional("TimestampData", reverse="circuit_job")
+
+    class TimestampData(database.Entity):
+        """Table for storing job metrics.
+        For each metric, there are two corresponding columns in the table reffering to start and finist times as timestamps.
+        Coloumns:
+            api_entry: timestamp when the job entered the MQP API
+            api_exit: timestamp when the job exited the MQP API
+            qdb_entry: timestamp when the job entered the QuantumDB
+            qdb_exit: timestamp when the job exited the QuantumDB
+            qjr_entry: timestamp when the job entered the MQSS Quantum Job Runner
+            qjr_exit: timestamp when the job exited the MQSS Quantum Job Runner
+            isv_jr_entry: timestamp when the job entered the ISV Job Runner
+            isv_jr_exit: timestamp when the job exited the ISV Job Runner
+            quantum_daemon_jr_entry: timestamp when the job entered the Quantum Daemon Job Runner
+            quantum_daemon_jr_exit: timestamp when the job exited the Quantum Daemon Job Runner
+            generator_entry: timestamp when the job entered the circuit generator
+            generator_exit: timestamp when the job exited the circuit generator
+            scheduler_entry: timestamp when the job entered the scheduler
+            scheduler_exit: timestamp when the job exited the scheduler
+            pass_runner_entry: timestamp when the job entered the pass runner
+            pass_runner_exit: timestamp when the job exited the pass runner
+            passes_applied: dictionary listing entry and exit-times for each pass applied
+            transpiler_entry: timestamp when the job entered the transpiler
+            transpiler_exit: timestamp when the job exited the transpiler
+            submitter_entry: timestamp when the job entered the submitter
+            submitter_exit: timestamp when the job exited the submitter
+            pass_selection_entry: timestamp when the job entered the pass selection
+            pass_selection_exit: timestamp when the job exited the pass selection
+            knitter_entry: timestamp when the job entered the knitter
+            knitter_exit: timestamp when the job exited the knitter
+            job_execution_start: timestamp when the job started execution on the quantum hardware
+            job_execution_end: timestamp when the job finished execution on the quantum hardware
+        """
+
+        _table_ = "timestamp_data"
+        circuit_job = Required(CircuitJob, reverse="timestamp_data")
+
+        api_entry = Required(datetime)
+        api_exit = Required(datetime)
+
+        qdb_entry = Optional(datetime)
+        qdb_exit = Optional(datetime)
+
+        qjr_entry = Optional(datetime)
+        qjr_exit = Optional(datetime)
+
+        isv_jr_entry = Optional(datetime)
+        isv_jr_exit = Optional(datetime)
+
+        quantum_daemon_jr_entry = Optional(datetime)
+        quantum_daemon_jr_exit = Optional(datetime)
+
+        generator_entry = Optional(datetime)
+        generator_exit = Optional(datetime)
+
+        scheduler_entry = Optional(datetime)
+        scheduler_exit = Optional(datetime)
+
+        pass_runner_entry = Optional(datetime)
+        pass_runner_exit = Optional(datetime)
+
+        passes_applied = Optional(
+            str
+        )  # This is a dictionary listing entry and exit-times for passes
+
+        transpiler_entry = Optional(datetime)
+        transpiler_exit = Optional(datetime)
+
+        submitter_entry = Optional(datetime)
+        submitter_exit = Optional(datetime)
+
+        pass_selection_entry = Optional(datetime)
+        pass_selection_exit = Optional(datetime)
+
+        knitter_entry = Optional(datetime)
+        knitter_exit = Optional(datetime)
+
+        job_execution_start = Optional(datetime)
+        job_execution_end = Optional(datetime)
+
     class HamiltonianJob(database.Entity):
         _table_ = "hamiltonian_job"
 
@@ -286,19 +367,24 @@ class TimeSlots(database.Entity):
 # pylint: enable=unused-variable
 
 
-def _define_database(create_tables: bool = False, **db_params):
+def _define_database(create_tables: bool = False, **db_params) -> Database:
+
     db = Database(**db_params)
 
     _define_entities(db)
 
-    db.generate_mapping(create_tables=create_tables)
+    try:
+        db.generate_mapping(create_tables=create_tables)
+    except Exception as e:
+        print(e)
 
     return db
 
 
-def open_database(create_tables: bool = False):
-    if os.getenv("QUANTUM_DB_TESTING") is not None and [os.getenv("USER_TESTING") == ""
-    or os.getenv("USER_TESTING") is None]:
+def open_database(create_tables: bool = False) -> Database:
+    if os.getenv("QUANTUM_DB_TESTING") is not None and [
+        os.getenv("USER_TESTING") == "" or os.getenv("USER_TESTING") is None
+    ]:
         return _define_database(
             create_tables=create_tables,
             provider="sqlite",

bqp_database_access/job_metrics.py

@@ -0,0 +1,106 @@
+"""This module contains all helpers related to job metrics for resources in the database."""
+
+from datetime import datetime
+
+from pony.orm import db_session  # type: ignore
+
+from ._database import open_database
+
+
+@db_session
+def fill_job_metrics(job_id: int, metrics: dict) -> None:
+    """Given the job_id and metrics, fill the new TimestampData table.
+    This is being called by the
+
+    Args:
+        job_id (int): uuid of the circuit jobs
+        metrics (dict): Dictionary of the runtime metrics
+    """
+    quantum_db = open_database()
+
+    job = quantum_db.TimestampData.get(id=job_id)
+
+    if job is None:
+        return
+
+    # list of timestamp attribute names on the TimestampData entity
+    timestamp_attrs = [
+        "api_entry",
+        "api_exit",
+        "qdb_entry",
+        "qdb_exit",
+        "qjr_entry",
+        "qjr_exit",
+        "isv_jr_entry",
+        "isv_jr_exit",
+        "quantum_daemon_jr_entry",
+        "quantum_daemon_jr_exit",
+        "generator_entry",
+        "generator_exit",
+        "scheduler_entry",
+        "scheduler_exit",
+        "pass_runner_entry",
+        "pass_runner_exit",
+        "passes_applied",
+        "transpiler_entry",
+        "transpiler_exit",
+        "submitter_entry",
+        "submitter_exit",
+        "pass_selection_entry",
+        "pass_selection_exit",
+        "knitter_entry",
+        "knitter_exit",
+        "job_execution_start",
+        "job_execution_end",
+    ]
+
+    for attr in timestamp_attrs:
+        match_key = next((k for k in metrics.keys() if k.lower() == attr.lower()), None)
+        if match_key is None:
+            continue
+        value = metrics.get(match_key)
+        if value is None:
+            continue
+        try:
+            setattr(job, attr, value)
+        except Exception:
+            continue
+
+    quantum_db.commit()
+
+
+@db_session
+def update_timestamp_data_submitted(id: str) -> None:
+    """Fills the timestamp_completed field of the TimestampData table, since
+    that information is already filled on the CircuitJob table.
+
+    Args:
+        id (str): job id of a CircuitJob.
+    """
+    quantum_db = open_database()
+
+    jobs = list(
+        quantum_db.CircuitJob.select(status="COMPLETED").filter(
+            lambda job: (job.id == id)
+        )
+    )
+    timestamp_job = quantum_db.TimestampData.get(id=id)
+    timestamp_job.job_execution_end = jobs.timestamp_completed
+
+
+@db_session
+def fetch_timestamp_data_by_job_id(
+    id: str,
+) -> list["TimestampData"] | None:
+    quantum_db = open_database()
+
+    try:
+        query = quantum_db.TimestampData.select(id=id)
+        job = list(query)
+    except Exception as e:
+        print("Job metrics cannot be found (or not specifed) for this job")
+        return None
+    if not job:
+        print(f"Job metrics cannot be found (or not specified) for this job")
+        return None
+    return job

bqp_database_access/jobs.py

@@ -25,16 +25,21 @@ def fetch_by_identity(identity: str) -> list["CircuitJob"]:  # type: ignore
 
 
 @db_session
-def fetch_by_identity_pages(identity: str, page: int, jobs_per_page: int,
-                             order: str, order_by: str, filter_query: str) -> \
-    dict[str, list["CircuitJob"] | int]:  # type: ignore
+def fetch_by_identity_pages(
+    identity: str,
+    page: int,
+    jobs_per_page: int,
+    order: str,
+    order_by: str,
+    filter_query: str,
+) -> dict[str, list["CircuitJob"] | int]:  # type: ignore
     """
     fetch all Circuit jobs belonging to a user with particular identity and pagintate the
     results (instead of fetchign all db entries, only a certain range is fetched which is
     expected to reduce fetch time and ease query load) for the case where jobs have to be
     displayed on MQP website page
     """
-    start_list = page*jobs_per_page
+    start_list = page * jobs_per_page
     quantum_db = open_database()
     user = quantum_db.User.get(identity=identity)
     table = quantum_db.CircuitJob
@@ -43,20 +48,24 @@ def fetch_by_identity_pages(identity: str, page: int, jobs_per_page: int,
     query = f"SELECT * FROM circuit_job WHERE owner = '{identity}'"
 
     if filter_query:
-        n_total = quantum_db.select(f"SELECT COUNT(*) FROM circuit_job WHERE owner = '{identity}' AND status = '{filter_query}'")[0]
+        n_total = quantum_db.select(
+            f"SELECT COUNT(*) FROM circuit_job WHERE owner = '{identity}' AND status = '{filter_query}'"
+        )[0]
         query += f"AND status = '{filter_query}'"
     else:
-        n_total = quantum_db.select(f"SELECT COUNT(*) FROM circuit_job WHERE OWNER = '{identity}'")[0]
-
+        n_total = quantum_db.select(
+            f"SELECT COUNT(*) FROM circuit_job WHERE OWNER = '{identity}'"
+        )[0]
+
     if order_by:
-        query+=f" ORDER BY {order_by}"
+        query += f" ORDER BY {order_by}"
 
-    query+=f" {order}"
-    query+=f" LIMIT {jobs_per_page}"
-    query+=f" OFFSET {start_list}"
+    query += f" {order}"
+    query += f" LIMIT {jobs_per_page}"
+    query += f" OFFSET {start_list}"
     query_res = table.select_by_sql(query)
-    return {"jobs": query_res,
-            "totaljob_nr": int(n_total)}
+    return {"jobs": query_res, "totaljob_nr": int(n_total)}
+
 
 @db_session
 def fetch_hamiltonian_job_by_identity(identity: str) -> list["HamiltonianJob"]:  # type: ignore
@@ -299,14 +308,14 @@ def fetch_all_pending_hamiltonian_jobs() -> list["HamiltonianJob"]:  # type: ign
 
 
 @db_session
-def fetch_all_waiting_jobs() -> list["CircuitJob"]: # type: ignore
+def fetch_all_waiting_jobs() -> list["CircuitJob"]:  # type: ignore
     quantum_db = open_database()
 
     return list(quantum_db.CircuitJob.select(status="WAITING"))
 
 
 @db_session
-def fetch_all_waiting_hamiltonian_jobs() -> list["HamiltonianJob"]: # type: ignore
+def fetch_all_waiting_hamiltonian_jobs() -> list["HamiltonianJob"]:  # type: ignore
     quantum_db = open_database()
 
     return list(quantum_db.HamiltonianJob.select(status="WAITING"))
@@ -402,4 +411,4 @@ def cancel_hamiltonian_job(job_id: int, note: str) -> None:
 
     job.note = note
     job.timestamp_cancelled = datetime.now()
-    job.status = "CANCELLED"
+    job.status = "CANCELLED"

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "bqp-database-access"
-version = "0.3.29"
+version = "0.3.30"
 description = ""
 authors = [
     { name = "Martin Ruefenacht", email = "martin.ruefenacht@lrz.de" },