Fix per stage PromQL metric queries by using time.time() instead of time.perf_counter() (#177)

PromQL queries require the epoch time as input which requires using
`time.time()` since `time.perf_counter()` should only be used for
relative differences in time. Lost precision associated with using
`time.time()` over `time.perf_counter()` is of negligible consequence.

Minor logging improvements also included.

Author: Bslabe123

inference_perf/client/metricsclient/prometheus_client/base.py

@@ -156,6 +156,7 @@ def collect_metrics_for_stage(self, runtime_parameters: PerfRuntimeParameters, s
         # Get the query evaluation time and duration for the stage
         # The query evaluation time is the end time of the stage plus the scrape interval and a buffer to ensure metrics are collected
         # Duration is calculated as the difference between the eval time and start time of the stage
+        logger.debug(f"runtime parameters for stage {stage_id}: {runtime_parameters}")
         query_eval_time = runtime_parameters.stages[stage_id].end_time + self.scrape_interval + PROMETHEUS_SCRAPE_BUFFER_SEC
         query_duration = query_eval_time - runtime_parameters.stages[stage_id].start_time
         return self.get_model_server_metrics(runtime_parameters.model_server_client, query_duration, query_eval_time)
@@ -230,15 +231,15 @@ def execute_query(self, query: str, eval_time: str) -> float:
         """
         query_result = 0.0
         try:
-            logger.debug(f"Making PromQL query: '{query}'")
+            logger.debug(f"making PromQL query: '{query}'")
             response = requests.get(self.query_url, headers=self.get_headers(), params={"query": query, "time": eval_time})
             if response is None:
-                logger.error("Error executing query: %s" % (query))
+                logger.error("error executing query: %s" % (query))
                 return query_result
 
             response.raise_for_status()
         except Exception as e:
-            logger.error("Error executing query: %s" % (e))
+            logger.error("error executing query: %s" % (e))
             return query_result
 
         # Check if the response is valid
@@ -258,9 +259,11 @@ def execute_query(self, query: str, eval_time: str) -> float:
         #         ]
         #     }
         # }
+        
         response_obj = response.json()
+        logger.debug(f"got result for query '{query}': {response_obj}")
         if response_obj.get("status") != "success":
-            logger.error("Error executing query: %s" % (response_obj))
+            logger.error("error executing query: %s" % (response_obj))
             return query_result
 
         data = response_obj.get("data", {})
@@ -276,8 +279,9 @@ def execute_query(self, query: str, eval_time: str) -> float:
                 try:
                     query_result = round(float(result[0]["value"][1]), 6)
                 except ValueError:
-                    logger.error("Error converting value to float: %s" % (result[0]["value"][1]))
+                    logger.error("error converting value to float: %s" % (result[0]["value"][1]))
                     return query_result
+        logger.debug(f"inferred result from query '{query}': {query_result}")
         return query_result
 
     def get_headers(self) -> dict[str, Any]:

inference_perf/client/modelserver/mock_client.py

@@ -30,7 +30,7 @@ def __init__(self, metrics_collector: RequestDataCollector, api_config: APIConfi
         self.metrics_collector = metrics_collector
 
     async def process_request(self, data: InferenceAPIData, stage_id: int, scheduled_time: float) -> None:
-        start = time.monotonic()
+        start = time.perf_counter()
         logger.debug("Processing mock request for stage %d", stage_id)
         await asyncio.sleep(3)
         self.metrics_collector.record_metric(
@@ -43,7 +43,7 @@ async def process_request(self, data: InferenceAPIData, stage_id: int, scheduled
                 ),
                 error=None,
                 start_time=start,
-                end_time=time.monotonic(),
+                end_time=time.perf_counter(),
                 scheduled_time=scheduled_time,
             )
         )

inference_perf/loadgen/load_generator.py

@@ -137,6 +137,7 @@ async def mp_run(self, client: ModelServerClient) -> None:
 
             # Allow generation a second to begin populating the queue so the workers
             # don't miss the initial scheuled request times
+            start_time_epoch = time.time()
             start_time = time.perf_counter() + 1
             num_requests = stage.rate * stage.duration
 
@@ -167,7 +168,7 @@ async def mp_run(self, client: ModelServerClient) -> None:
             logger.debug("Loadgen joining request queue")
             request_queue.join()
             self.stage_runtime_info[stage_id] = StageRuntimeInfo(
-                stage_id=stage_id, rate=stage.rate, start_time=start_time, end_time=time.perf_counter()
+                stage_id=stage_id, rate=stage.rate, start_time=start_time_epoch, end_time=time.time()
             )
             logger.info("Stage %d - run completed", stage_id)
             if self.stageInterval and stage_id < len(self.stages) - 1:
@@ -182,6 +183,7 @@ async def run(self, client: ModelServerClient) -> None:
 
         for stage_id, stage in enumerate(self.stages):
             timer = self.get_timer(stage.rate)
+            start_time_epoch = time.time()
             start_time = time.perf_counter()
             end_time = start_time + stage.duration
             logger.info("Stage %d - run started", stage_id)
@@ -198,7 +200,7 @@ async def run(self, client: ModelServerClient) -> None:
                     else:
                         break
             self.stage_runtime_info[stage_id] = StageRuntimeInfo(
-                stage_id=stage_id, rate=stage.rate, start_time=start_time, end_time=time.perf_counter()
+                stage_id=stage_id, rate=stage.rate, start_time=start_time_epoch, end_time=time.time()
             )
             logger.info("Stage %d - run completed", stage_id)
             if self.stageInterval and stage_id < len(self.stages) - 1:

inference_perf/loadgen/load_timer.py

@@ -44,7 +44,7 @@ def __init__(self, rate: float) -> None:
 
     def start_timer(self, initial: Optional[float] = None) -> Generator[float, None, None]:
         # Set start time
-        next_time = time.monotonic() if initial is None else initial
+        next_time = time.perf_counter() if initial is None else initial
 
         # Given a rate, yield a time to wait before the next request
         while True:
@@ -59,7 +59,7 @@ def __init__(self, rate: float) -> None:
 
     def start_timer(self, initial: Optional[float] = None) -> Generator[float, None, None]:
         # Set start time
-        next_time = time.monotonic() if initial is None else initial
+        next_time = time.perf_counter() if initial is None else initial
 
         # Given a rate, yield a time to wait before the next request
         while True: