Sync turing folder from GitHub

turing/api/app.py

@@ -1,12 +1,14 @@
 import base64
 import os
+from typing import Literal
 
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Query
 from fastapi.responses import JSONResponse
 import gradio as gr
 from loguru import logger
 
 from turing.api.demo import create_demo
+from turing.api.resource_monitoring import PrometheusBodyMiddleware, instrumentator
 from turing.api.schemas import PredictionRequest, PredictionResponse
 from turing.modeling.predict import ModelInference
 
@@ -53,6 +55,9 @@ app = FastAPI(
     version="1.0.0"
 )
 
+## Add Prometheus middleware
+app.add_middleware(PrometheusBodyMiddleware)
+
 @app.get("/manifest.json")
 def get_manifest():
     return JSONResponse(content={
@@ -67,8 +72,10 @@ def get_manifest():
 
 # Global inference engine instance
 inference_engine = ModelInference()
-
 demo = create_demo(inference_engine)
+
+# Instrument the app with Prometheus metrics
+instrumentator.instrument(app).expose(app,include_in_schema=False, should_gzip=True)
 app = gr.mount_gradio_app(app, demo, path="/gradio")
 
 @app.get("/")
@@ -80,17 +87,19 @@ def health_check():
 
 
 @app.post("/predict", response_model=PredictionResponse)
-def predict(request: PredictionRequest):
+async def predict(request: PredictionRequest, language: Literal["java", "python", "pharo"] = Query(
+        ...
+    )):
     """
     Endpoint to classify a list of code comments.
     Dynamically loads the model from MLflow based on the request parameters.
     """
     try:
-        logger.info(f"Received prediction request for language: {request.language}")
-
+        logger.info(f"Received prediction request for language: {language}")
+        
         # Perform prediction using the inference engine
         raw, predictions, run_id, artifact = inference_engine.predict_payload(
-            texts=request.texts, language=request.language
+            texts=request.texts, language=language
         )
 
         # Ensure predictions are serializable (convert numpy arrays to lists)
@@ -100,7 +109,7 @@ def predict(request: PredictionRequest):
         return PredictionResponse(
             predictions=raw.tolist(),
             labels=predictions,
-            model_info={"artifact": artifact, "language": request.language},
+            model_info={"artifact": artifact, "language": language},
         )
 
     except Exception as e:

turing/api/resource_monitoring.py

@@ -0,0 +1,148 @@
+import json
+from typing import Callable
+
+from fastapi import Request
+from prometheus_client import Counter, Gauge
+from prometheus_fastapi_instrumentator import Instrumentator, metrics
+from prometheus_fastapi_instrumentator.metrics import Info
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.types import Message
+
+SUBSYSTEM = "model"
+NAMESPACE = "turing_api"
+
+# Define Prometheus metrics
+instrumentator = Instrumentator(
+    should_group_status_codes=False,
+    should_ignore_untemplated=True,
+    should_respect_env_var=False,
+    should_instrument_requests_inprogress=True,
+    excluded_handlers=["/metrics"],
+    inprogress_name="fastapi_inprogress",
+    inprogress_labels=True
+)
+
+## Define custom metric for tracking requested languages
+def http_requested_languages_total(
+        metric_name: str = "Total HTTP requested languages",
+        metric_description: str = "Total number of HTTP requests per programming language", 
+        metric_namespace: str = NAMESPACE,
+        metric_subsystem: str = SUBSYSTEM  ) -> Callable[[Info],None]:
+    METRIC = Counter(
+        metric_name,
+        metric_description,
+        namespace=metric_namespace,
+        subsystem=metric_subsystem,
+        labelnames=["language"]
+    )
+    async def instrumentation(info: Info) -> None:
+        try:
+            if info.modified_handler != "/predict":
+                return
+            lang = info.request.query_params.get("language")
+        except Exception:
+            print("Failed to get language from request")
+            lang = "other"
+
+        METRIC.labels(language=lang).inc()
+
+    return instrumentation
+
+
+## Define custom metrics for tracking code comments in requests
+http_request_code_comments_total = Counter (
+        "Total HTTP request code comments",
+        "Total number of comments in HTTP requests", 
+        namespace=NAMESPACE,
+        subsystem=SUBSYSTEM,
+        labelnames=["language"]
+    )
+
+## Define custom metrics for tracking characters in code comments
+http_request_comment_characters_total = Counter(
+        "Total HTTP request code comment characters",
+        "Total number of characters in the HTTP requests", 
+        namespace=NAMESPACE,
+        subsystem=SUBSYSTEM,
+        labelnames=["endpoint","language"]
+    )
+   
+
+## Define custom metric for tracking maximum characters per comment
+http_request_maximum_characters_per_comment = Gauge(
+        "Maximum characters per comment",
+        "Maximum number of characters in a single comment from HTTP requests", 
+        namespace=NAMESPACE,
+        subsystem=SUBSYSTEM,
+        labelnames=["language"]
+    )
+
+
+## Middleware to extract and record metrics from request body
+class PrometheusBodyMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next):
+        if request.url.path != "/predict":
+            return await call_next(request)
+        
+        body_bytes = await request.body()
+        query_params = request.query_params
+        try:
+            if body_bytes:
+                language = query_params.get("language", "unknown")
+                body_json = json.loads(body_bytes)
+                print(f"Request body JSON: {body_json}")
+                texts = body_json.get("texts")
+                if texts:
+                    total_characters = sum(len(example) for example in texts if example)
+                    max_characters = max((len(example) for example in texts if example), default=0)
+                    http_request_comment_characters_total.labels(endpoint="/predict", language=language).inc(total_characters)
+                    http_request_maximum_characters_per_comment.labels(language=language).set(max_characters)
+                    http_request_code_comments_total.labels(language=language).inc(len(texts))
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            pass
+
+        async def receive() -> Message:
+            return {"type": "http.request", "body": body_bytes}
+        
+        request._receive = receive
+
+        response = await call_next(request)
+        return response
+
+## Register metrics with the instrumentator
+instrumentator.add(
+    metrics.request_size(
+        should_include_handler=True,
+        should_include_method=False,
+        should_include_status=True,
+        metric_namespace=NAMESPACE,
+        metric_subsystem=SUBSYSTEM,
+    )
+).add(
+    metrics.response_size(
+        should_include_handler=True,
+        should_include_method=False,
+        should_include_status=True,
+        metric_namespace=NAMESPACE,
+        metric_subsystem=SUBSYSTEM
+    )
+).add(
+    http_requested_languages_total()
+).add(
+    metrics.requests(
+        should_include_handler=True,
+        should_include_method=True,
+        should_include_status=True,
+        metric_namespace=NAMESPACE,
+        metric_subsystem=SUBSYSTEM
+    )
+).add(
+    metrics.latency(
+        should_include_handler=True,        
+        should_include_method=False,
+        should_include_status=True,
+        metric_namespace=NAMESPACE,
+        metric_subsystem=SUBSYSTEM
+    )
+)
+

turing/api/schemas.py

@@ -10,9 +10,6 @@ class PredictionRequest(BaseModel):
         description="List of code comments to classify",
         example=["public void main", "def init self"],
     )
-    language: str = Field(
-        ..., description="Programming language (java, python, pharo)", example="java"
-    )
 
 
 # Output Schema