diff --git a/apps/metrics/backend/services/starrocks_metrics_service.py b/apps/metrics/backend/services/starrocks_metrics_service.py index fbf3080..29cb07f 100644 --- a/apps/metrics/backend/services/starrocks_metrics_service.py +++ b/apps/metrics/backend/services/starrocks_metrics_service.py @@ -41,6 +41,70 @@ class StarRocksMetricsService: AND product_id = %s ORDER BY date ASC """, + "dcr": """ + SELECT + date, + product_id, + value, + updated_date + FROM dws_dcr + WHERE date >= %s + AND date < %s + AND product_id = %s + ORDER BY date ASC + """, + "mrar": """ + SELECT + date, + product_id, + CASE + WHEN monthly_requests = 0 THEN 0.0 + ELSE (monthly_accepted_requests * 1.0) / monthly_requests + END AS value, + updated_date + FROM dws_mrar + WHERE date >= %s + AND date < %s + AND product_id = %s + ORDER BY date ASC + """, + "trar": """ + SELECT + product_id, + CASE + WHEN total_requests = 0 THEN 0.0 + ELSE (total_accepted_requests * 1.0) / total_requests + END AS value, + updated_date + FROM dws_trar + WHERE product_id = %s + """, + "mrqr": """ + SELECT + date, + product_id, + CASE + WHEN monthly_requests = 0 THEN 0.0 + ELSE (monthly_quoted_requests * 1.0) / monthly_requests + END AS value, + updated_date + FROM dws_mrqr + WHERE date >= %s + AND date < %s + AND product_id = %s + ORDER BY date ASC + """, + "trqr": """ + SELECT + product_id, + CASE + WHEN total_requests = 0 THEN 0.0 + ELSE (total_quoted_requests * 1.0) / total_requests + END AS value, + updated_date + FROM dws_trqr + WHERE product_id = %s + """, }, "magicleaps": { @@ -92,9 +156,9 @@ class StarRocksMetricsService: self, product_id: str, metric_name: str, - step: str, - start_date: Union[str, date], - end_date: Union[str, date] + step: Optional[str], + start_date: Optional[Union[str, date]], + end_date: Optional[Union[str, date]] ) -> List[Dict[str, Any]]: """ Query metric data for a specific date range. @@ -140,6 +204,17 @@ class StarRocksMetricsService: await self.module_logger.log_error(error_msg) raise HTTPException(status_code=404, detail=error_msg) + # Check if metric need time params + # Starting with "t" indicates a query for the total count since the very first record. + # NOTE: This determination logic is subject to future changes. + if (start_date is None) or (end_date is None) or (step is None): + if metric_name.startswith('t'): + return await self._query_metric_by_product_id(product_id, metric_name) + else: + error_msg = f"Metric '{metric_name}' should be queried by start date, end date and step." + await self.module_logger.log_error(error_msg) + raise HTTPException(status_code=404, detail=error_msg) + # Parse date strings if they are strings if isinstance(start_date, str): try: @@ -284,7 +359,7 @@ class StarRocksMetricsService: result_dict[date_str] = { "date": date_str, - "value": int(value) if value is not None else 0, + "value": value if value is not None else 0, "metric": metric_name, "labels": labels } @@ -373,4 +448,67 @@ class StarRocksMetricsService: "metric_name": metric_name, "sql_query": self.METRIC_SQL_MAP[product_id][metric_name].strip(), "description": f"{metric_name} count from StarRocks table dws_{metric_name}" - } \ No newline at end of file + } + + async def _query_metric_by_product_id(self, product_id: str, metric_name: str) -> List[Dict[str, Any]]: + """ + Query metric not suitable for date range (e.g. data related to calculating total records). + + Args: + product_id: Product ID to identify which product's metrics to query + metric_name: Name of the metric to query + + Returns: + List of dictionaries with 'product_id' key. + + Raises: + Exception: If StarRocks query fails + + Example: + result = await service.query_metric_by_time_range( + "freeleaps", + "total_request_quoted_rate", + ) + # Returns: [{"date": "freeleaps", "value": 45, "labels": {...}},] + """ + + # Get the SQL query for the metric + sql_query = self.METRIC_SQL_MAP[product_id][metric_name] + + try: + await self.module_logger.log_info( + f"Querying metric '{metric_name}' from product '{product_id}'") + + # Execute the query + result = await self.starrocks_client.execute_query( + query=sql_query, + params=(product_id) + ) + + # Parse the result and format it + for row in result: + # Get the value + value = row.get("value", 0) + if value is None: + value = 0 + + # Create labels dictionary + labels = { + "product_id": row.get("product_id", product_id), + "metric_type": metric_name, + } + result_dict = [] + result_dict.append({ + "date": None, + "value": value if value is not None else 0, + "metric": metric_name, + "labels": labels + }) + + await self.module_logger.log_info( + f"Successfully queried metric '{metric_name}'") + return result_dict + + except Exception as e: + await self.module_logger.log_error(f"Failed to query metric '{metric_name}': {e}") + raise \ No newline at end of file diff --git a/apps/metrics/webapi/routes/starrocks_metrics/metrics_query.py b/apps/metrics/webapi/routes/starrocks_metrics/metrics_query.py index a035a05..0e2eb66 100644 --- a/apps/metrics/webapi/routes/starrocks_metrics/metrics_query.py +++ b/apps/metrics/webapi/routes/starrocks_metrics/metrics_query.py @@ -9,7 +9,7 @@ from backend.services.starrocks_metrics_service import StarRocksMetricsService class MetricDataPoint(BaseModel): """Single data point in metric time series.""" - date: str = Field(..., description="Date in YYYY-MM-DD format") + date: Optional[str] = Field(None, description="Date in YYYY-MM-DD format") value: Union[int, float] = Field(..., description="Metric value") labels: Dict[str, Any] = Field(default_factory=dict, description="Metric labels") @@ -19,17 +19,16 @@ class MetricTimeSeriesResponse(BaseModel): metric_name: str = Field(..., description="Name of the queried metric") data_points: List[MetricDataPoint] = Field(..., description="List of data points") total_points: int = Field(..., description="Total number of data points") - time_range: Dict[str, str] = Field(..., description="Start and end date of the query") + time_range: Dict[Optional[str], Optional[str]] = Field([None, None], description="Start and end date of the query") class MetricQueryRequest(BaseModel): - """Request model for metric query.""" + """Request model for metric query by time range.""" product_id: str = Field(..., description="Product ID to identify which product's data to query") metric_name: str = Field(..., description="Name of the metric to query") - step: str = Field(..., description="Aggregation step, e.g., 1d or 1m") - start_date: str = Field(..., description="Start date in YYYY-MM-DD HH:MM:SS format") - end_date: str = Field(..., description="End date in YYYY-MM-DD HH:MM:SS format") - + step: Optional[str] = Field(None, description="Aggregation step, e.g., 1d or 1m") + start_date: Optional[str] = Field(None, description="Start date in YYYY-MM-DD HH:MM:SS format") + end_date: Optional[str] = Field(None, description="End date in YYYY-MM-DD HH:MM:SS format") router = APIRouter() @@ -72,11 +71,11 @@ async def metrics_query( ], total_points=len(data_points), time_range={ - "start": request.start_date, - "end": request.end_date + "start": request.start_date if request.start_date else None, + "end": request.end_date if request.end_date else None } ) await module_logger.log_info( f"Successfully queried metric '{request.metric_name}' with {len(data_points)} data points") - return response + return response \ No newline at end of file