Add benchmark

Author: sitaowang1998

components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py

@@ -911,7 +911,12 @@ def handle_pending_query_jobs(
                 orchestration_type,
                 spider_driver,
             )
-            logger.info("Dispatched search job.")
+            logger.info(
+                "Dispatched search job %s at %s with %d tasks.",
+                job.id,
+                job.start_time.isoformat(),
+                len(archives_for_search),
+            )
 
     return reducer_acquisition_tasks
 
@@ -1037,20 +1042,37 @@ async def handle_finished_search_job(
 
     # We set the status regardless of the job's previous status to handle the case where the
     # job is cancelled (status = CANCELLING) while we're in this method.
+    end_time = datetime.datetime.now()
+    duration = (end_time - job.start_time).total_seconds()
     if set_job_or_task_status(
         db_conn,
         QUERY_JOBS_TABLE_NAME,
         job_id,
         new_job_status,
         num_tasks_completed=job.num_archives_searched,
-        duration=(datetime.datetime.now() - job.start_time).total_seconds(),
+        duration=duration,
     ):
         if new_job_status == QueryJobStatus.SUCCEEDED:
-            logger.info("Completed job.")
+            logger.info(
+                "Completed job %s at %s, total duration=%.2fs.",
+                job_id,
+                end_time.isoformat(),
+                duration,
+            )
         elif reducer_failed:
-            logger.error("Completed job with failing reducer.")
+            logger.error(
+                "Completed job %s at %s with failing reducer, total duration=%.2fs.",
+                job_id,
+                end_time.isoformat(),
+                duration,
+            )
         else:
-            logger.info("Completed job with failing tasks.")
+            logger.info(
+                "Completed job %s at %s with failing tasks, total duration=%.2fs.",
+                job_id,
+                end_time.isoformat(),
+                duration,
+            )
     del active_jobs[job_id]
 
 

components/job-orchestration/job_orchestration/scheduler/query/spider_query_scheduler.py

@@ -15,9 +15,10 @@
 
 import datetime
 import json
-from logging import getLogger
 from typing import Any, TYPE_CHECKING
 
+from clp_py_utils.clp_logging import get_logger
+
 from spider_py import Int64
 from spider_py.client import Channel, channel_task, Driver, group, Job, TaskGraph
 from spider_py.core import JobStatus
@@ -36,7 +37,7 @@
 if TYPE_CHECKING:
     from job_orchestration.scheduler.job_config import SearchJobConfig
 
-logger = getLogger(__name__)
+logger = get_logger("search-job-handler")
 
 
 def build_search_task_graph(num_archives: int, has_aggregation: bool) -> TaskGraph:
@@ -245,11 +246,27 @@ def dispatch_search_job(  # noqa: PLR0913
         has_aggregation=has_aggregation,
     )
 
-    logger.info("Submitting Spider job.")
+    submit_start_time = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
+    logger.info(
+        "Submitting Spider job %s at %s with %d search tasks, aggregation=%s",
+        job_id,
+        submit_start_time.isoformat(),
+        len(archives),
+        has_aggregation,
+    )
 
     # Submit job to Spider
     jobs = driver.submit_jobs([task_graph], [tuple(job_inputs)])
 
+    submit_end_time = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
+    submit_duration = (submit_end_time - submit_start_time).total_seconds()
+    logger.info(
+        "Submitted Spider job %s at %s, submission took %.3fs",
+        job_id,
+        submit_end_time.isoformat(),
+        submit_duration,
+    )
+
     if not jobs:
         raise SpiderJobSubmitError(job_id)