Merge pull request #34 from TranslatorSRI/compression

Enable compression and add Jaeger

Dockerfile

@@ -1,4 +1,4 @@
-FROM renciorg/renci-python-image:v0.0.1
+FROM ghcr.io/translatorsri/renci-python-image:3.12.4
 
 # Add image info
 LABEL org.opencontainers.image.source /github.com/TranslatorSRI/answer-appraiser

Dockerfile.test

@@ -1,4 +1,4 @@
-FROM renciorg/renci-python-image:v0.0.1
+FROM ghcr.io/translatorsri/renci-python-image:3.12.4
 
 # Add image info
 LABEL org.opencontainers.image.source /github.com/TranslatorSRI/answer-appraiser

app/config.py

@@ -12,6 +12,10 @@ class Settings(BaseSettings):
     redis_port: int = 6379
     redis_password: str = "supersecretpassword"
 
+    jaeger_enabled: str = "True"
+    jaeger_host: str = "jaeger"
+    jaeger_port: int = 6831
+
     class Config:
         env_file = ".env"
 

app/server.py

@@ -1,15 +1,16 @@
+import gzip
+import json
 import logging
+import os
 import redis
 import traceback
+import warnings
 
-from fastapi import Body, BackgroundTasks, HTTPException, status
-from fastapi.responses import JSONResponse
-import httpx
+from fastapi import HTTPException, status, Request
+from fastapi.responses import JSONResponse, Response
 from starlette.middleware.cors import CORSMiddleware
 from uuid import uuid4
 
-from reasoner_pydantic import AsyncQuery, AsyncQueryResponse, Response, Query
-
 from .config import settings
 from .logger import setup_logger, get_logger
 from .trapi import TRAPI
@@ -21,7 +22,7 @@
 
 openapi_args = dict(
     title="SRI Answer Appraiser",
-    version="0.6.0",
+    version="0.6.1",
     terms_of_service="",
     description="SRI service that provides metrics for scoring and ordering of results",
     trapi="1.5.0",
@@ -53,6 +54,40 @@
     allow_headers=["*"],
 )
 
+if settings.jaeger_enabled == "True":
+    LOGGER.info("Starting up Jaeger")
+
+    from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+    from opentelemetry import trace
+    from opentelemetry.exporter.jaeger.thrift import JaegerExporter
+    from opentelemetry.sdk.resources import (
+        SERVICE_NAME,
+        Resource,
+    )
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import BatchSpanProcessor
+    from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
+
+    # httpx connections need to be open a little longer by the otel decorators
+    # but some libs display warnings of resource being unclosed.
+    # these supresses such warnings.
+    logging.captureWarnings(capture=True)
+    warnings.filterwarnings("ignore", category=ResourceWarning)
+    service_name = os.environ.get("OTEL_SERVICE_NAME", "ANSWER-APPRAISER")
+    jaeger_exporter = JaegerExporter(
+        agent_host_name=settings.jaeger_host,
+        agent_port=int(settings.jaeger_port),
+    )
+    resource = Resource(attributes={SERVICE_NAME: service_name})
+    provider = TracerProvider(resource=resource)
+    processor = BatchSpanProcessor(jaeger_exporter)
+    provider.add_span_processor(processor)
+    trace.set_tracer_provider(provider)
+    FastAPIInstrumentor.instrument_app(
+        APP, tracer_provider=provider, excluded_urls="docs,openapi.json"
+    )
+    HTTPXClientInstrumentor().instrument()
+
 EXAMPLE = {
     "message": {
         "query_graph": {
@@ -113,64 +148,27 @@
     }
 }
 
-ASYNC_EXAMPLE = {
-    "callback": "http://test",
-    **EXAMPLE,
-}
-
-
-async def async_appraise(message, callback, logger: logging.Logger):
-    try:
-        await get_ordering_components(message, logger)
-    except Exception:
-        logger.error(f"Something went wrong while appraising: {traceback.format_exc()}")
-    logger.info("Done appraising")
-    try:
-        logger.info(f"Posting to callback {callback}")
-        async with httpx.AsyncClient(timeout=httpx.Timeout(timeout=600.0)) as client:
-            res = await client.post(callback, json=message)
-            logger.info(f"Posted to {callback} with code {res.status_code}")
-    except Exception as e:
-        logger.error(f"Unable to post to callback {callback}.")
-
-
-@APP.post("/async_get_appraisal", response_model=AsyncQueryResponse)
-async def get_appraisal(
-    background_tasks: BackgroundTasks,
-    query: AsyncQuery = Body(..., example=ASYNC_EXAMPLE),
-):
-    """Appraise Answers"""
-    qid = str(uuid4())[:8]
-    query_dict = query.dict()
-    log_level = query_dict.get("log_level") or "INFO"
-    logger = get_logger(qid, log_level)
-    logger.info("Starting async appraisal")
-    message = query_dict["message"]
-    if not message.get("results"):
-        logger.warning("No results given.")
-        return JSONResponse(
-            content={"status": "Rejected", "description": "No Results.", "job_id": qid},
-            status_code=400,
-        )
-    callback = query_dict["callback"]
-    background_tasks.add_task(async_appraise, message, callback, logger)
-    return JSONResponse(
-        content={
-            "status": "Accepted",
-            "description": f"Appraising answers. Will send response to {callback}",
-            "job_id": qid,
-        },
-        status_code=200,
-    )
-
 
 @APP.post("/get_appraisal")
-async def sync_get_appraisal(query=Body(..., example=EXAMPLE)):
+async def sync_get_appraisal(request: Request):
     qid = str(uuid4())[:8]
-    # query_dict = query.dict()
-    log_level = query.get("log_level") or "INFO"
-    logger = get_logger(qid, log_level)
+    logger = get_logger(qid, "INFO")
     logger.info("Starting sync appraisal")
+    compressed = False
+    if request.headers.get("content-encoding") == "gzip":
+        try:
+            raw_body = await request.body()
+            query = json.loads(gzip.decompress(raw_body))
+            compressed = True
+        except Exception:
+            return Response("Invalid request. Failed to decompress and ingest.", 400)
+    else:
+        try:
+            query = await request.json()
+        except json.JSONDecodeError:
+            return Response("Invalid request. Failed to parse JSON.", 400)
+    log_level = query.get("log_level") or "INFO"
+    logger.setLevel(log_level)
     message = query["message"]
     if not message.get("results"):
         return JSONResponse(
@@ -181,8 +179,12 @@ async def sync_get_appraisal(query=Body(..., example=EXAMPLE)):
         await get_ordering_components(message, logger)
     except Exception:
         logger.error(f"Something went wrong while appraising: {traceback.format_exc()}")
+    if compressed:
+        query = gzip.compress(json.dumps(query).encode())
+    else:
+        query = json.dumps(query)
     logger.info("Done appraising")
-    return query
+    return Response(query)
 
 
 @APP.get("/redis_ready")

manage.py

@@ -53,7 +53,7 @@ def lock(extra_args):
     """
     for src, locked in REQUIREMENTS_FILES.items():
         command = f"""\
-        docker run -v $(pwd):/app python:3.9 \
+        docker run -v $(pwd):/app python:3.12 \
             /bin/bash -c "
                 # Install lockfile first so that we get the
                 # currently installed versions of dependencies
@@ -74,7 +74,7 @@ def verify_locked(extra_args):
     for src, locked in REQUIREMENTS_FILES.items():
         dependencies = get_command_output(
             f"""\
-        docker run -v $(pwd):/app python:3.9 \
+        docker run -v $(pwd):/app python:3.12 \
             /bin/bash -c "
                 pip install -qqq -r /app/{locked} &&
                 pip install -qqq -r /app/{src}    &&
@@ -84,7 +84,7 @@ def verify_locked(extra_args):
         )
         lock_dependencies = get_command_output(
             f"""\
-        docker run -v $(pwd):/app python:3.9 \
+        docker run -v $(pwd):/app python:3.12 \
             /bin/bash -c "
                 pip install -qqq -r /app/{locked} &&
                 pip freeze
@@ -102,7 +102,7 @@ def upgrade(extra_args):
     """
     for src, locked in REQUIREMENTS_FILES.items():
         command = f"""\
-        docker run -v $(pwd):/app python:3.9 \
+        docker run -v $(pwd):/app python:3.12 \
             /bin/bash -c "
                 # Install dependencies, getting latest version
                 pip install -r /app/{src} &&

requirements-lock.txt

@@ -1,34 +1,48 @@
-anyio==3.7.0
-asgiref==3.7.2
-async-timeout==4.0.2
-certifi==2023.5.7
-click==7.1.2
-exceptiongroup==1.1.1
-fastapi==0.75.0
-gunicorn==20.1.0
+anyio==4.6.0
+asgiref==3.8.1
+certifi==2024.8.30
+click==8.1.7
+Deprecated==1.2.14
+fastapi==0.115.0
+googleapis-common-protos==1.59.1
+grpcio==1.66.2
+gunicorn==23.0.0
 h11==0.14.0
-httpcore==0.17.2
-httptools==0.2.0
-httpx==0.24.1
-idna==3.4
-numpy==1.25.1
-pandas==2.0.3
-Pillow==10.0.0
-pydantic==1.10.9
-python-dateutil==2.8.2
-python-dotenv==1.0.0
-pytz==2023.3
-PyYAML==6.0
-rdkit==2023.3.2
-reasoner-pydantic==5.0.3
-redis==4.6.0
+httpcore==1.0.6
+httpx==0.27.2
+idna==3.10
+importlib_metadata==8.4.0
+numpy==2.1.1
+opentelemetry-api==1.27.0
+opentelemetry-exporter-jaeger==1.21.0
+opentelemetry-exporter-jaeger-proto-grpc==1.21.0
+opentelemetry-exporter-jaeger-thrift==1.21.0
+opentelemetry-instrumentation==0.48b0
+opentelemetry-instrumentation-asgi==0.48b0
+opentelemetry-instrumentation-fastapi==0.48b0
+opentelemetry-instrumentation-httpx==0.48b0
+opentelemetry-sdk==1.27.0
+opentelemetry-semantic-conventions==0.48b0
+opentelemetry-util-http==0.48b0
+packaging==24.1
+pandas==2.2.3
+pillow==10.4.0
+protobuf==4.25.5
+pydantic==1.10.18
+python-dateutil==2.9.0.post0
+pytz==2024.2
+rdkit==2024.3.5
+reasoner-pydantic==5.0.6
+redis==5.1.1
+setuptools==70.0.0
 six==1.16.0
-sniffio==1.3.0
-starlette==0.17.1
-tqdm==4.65.0
-typing_extensions==4.6.3
-tzdata==2023.3
-uvicorn==0.13.3
-uvloop==0.17.0
-watchgod==0.8.2
-websockets==11.0.3
+sniffio==1.3.1
+starlette==0.38.6
+thrift==0.21.0
+tqdm==4.66.5
+typing_extensions==4.12.2
+tzdata==2024.2
+uvicorn==0.31.0
+wheel==0.43.0
+wrapt==1.16.0
+zipp==3.20.2

requirements-test-lock.txt

@@ -1,19 +1,16 @@
-async-timeout==4.0.2
-attrs==23.1.0
-certifi==2023.5.7
-chardet==4.0.0
-coverage==7.2.7
-fakeredis==2.17.0
-idna==2.10
+certifi==2024.8.30
+charset-normalizer==3.3.2
+coverage==7.6.1
+fakeredis==2.25.1
+idna==3.10
 iniconfig==2.0.0
-packaging==23.1
-pluggy==0.13.1
-py==1.11.0
-pytest==6.2.2
-pytest-asyncio==0.16.0
-pytest-cov==2.11.1
-redis==4.6.0
-requests==2.25.1
+packaging==24.1
+pluggy==1.5.0
+pytest==8.3.3
+pytest-cov==5.0.0
+redis==5.1.1
+requests==2.32.3
+setuptools==70.0.0
 sortedcontainers==2.4.0
-toml==0.10.2
-urllib3==1.26.16
+urllib3==2.2.3
+wheel==0.43.0

requirements-test.txt

@@ -1,4 +1,4 @@
-fakeredis==2.17.0
-pytest==6.2.2
-pytest-cov==2.11.1
-requests==2.25.1
+fakeredis
+pytest
+pytest-cov
+requests

requirements.txt

@@ -1,10 +1,14 @@
-fastapi==0.75.0
-gunicorn==20.1.0
-httpx==0.24.1
-numpy==1.25.1
-pandas==2.0.3
-rdkit==2023.3.2
-reasoner-pydantic==5.0.3
-redis==4.6.0
-tqdm==4.65.0
-uvicorn==0.13.3
+fastapi
+gunicorn
+httpx
+numpy
+opentelemetry-sdk
+opentelemetry-instrumentation-fastapi
+opentelemetry-exporter-jaeger
+opentelemetry-instrumentation-httpx
+pandas
+rdkit
+reasoner-pydantic
+redis
+tqdm
+uvicorn