Switch to Azure Monitor Library

.github/workflows/_containerTemplate.yml

@@ -56,7 +56,7 @@ jobs:
       - name: Install cosign
         uses: sigstore/[email protected]
         id: install_cosign
-        if: github.event_name != 'pull_request'
+        if: github.event_name == 'release'
         with:
           cosign-release: 'v2.2.0'
 
@@ -74,7 +74,7 @@ jobs:
       - name: Login Container Registry
         uses: docker/[email protected]
         id: registry_login
-        if: github.event_name != 'pull_request'
+        # if: github.event_name != 'pull_request'
         with:
           registry: ${{ inputs.registry_uri }}
           username: ${{ secrets.USER_NAME }}
@@ -101,7 +101,7 @@ jobs:
         with:
           context: ${{ inputs.working_directory }}
           file: ${{ inputs.working_directory }}/Dockerfile
-          push: ${{ github.event_name != 'pull_request' }}
+          push: true # ${{ github.event_name != 'pull_request' }}
           tags: ${{ steps.metadata.outputs.tags }}
           labels: ${{ steps.metadata.outputs.labels }}
           cache-from: type=gha

code/function/fastapp/api/v1/endpoints/sample.py

@@ -1,17 +1,38 @@
-from typing import Any
+from typing import Annotated
 
-from fastapi import APIRouter
+import httpx
+from fastapi import APIRouter, Header
 from fastapp.models.sample import SampleRequest, SampleResponse
-from fastapp.utils import setup_logging
+from fastapp.utils import setup_logging, setup_tracer
+from opentelemetry.trace import SpanKind
 
 logger = setup_logging(__name__)
+tracer = setup_tracer(__name__)
 
 router = APIRouter()
 
 
 @router.post("/sample", response_model=SampleResponse, name="sample")
 async def post_predict(
-    data: SampleRequest,
+    data: SampleRequest, x_forwarded_for: Annotated[str, Header()] = ""
 ) -> SampleResponse:
     logger.info(f"Received request: {data}")
+
+    # Sample request
+    async with httpx.AsyncClient() as client:
+        response = await client.get("https://www.bing.com")
+    logger.info(f"Received response status code: {response.status_code}")
+
+    # tracer_attributes = {"http.client_ip": x_forwarded_for}
+    # with tracer.start_as_current_span(
+    #     "dependency_span", attributes=tracer_attributes, kind=SpanKind.CLIENT
+    # ) as span:
+    #     try:
+    #         async with httpx.AsyncClient() as client:
+    #             response = await client.get("https://www.bing.com")
+    #         logger.info(f"Received response status code: {response.status_code}")
+    #     except Exception as ex:
+    #         span.set_attribute("status", "exception")
+    #         span.record_exception(ex)
+
     return SampleResponse(output=f"Hello {data.input}")

code/function/fastapp/core/config.py

@@ -14,6 +14,8 @@ class Settings(BaseSettings):
     APPLICATIONINSIGHTS_CONNECTION_STRING: str = Field(
         default="", env="APPLICATIONINSIGHTS_CONNECTION_STRING"
     )
+    WEBSITE_NAME: str = Field(default="", env="WEBSITE_SITE_NAME")
+    WEBSITE_INSTANCE_ID: str = Field(default="", env="WEBSITE_INSTANCE_ID")
     MY_SECRET_CONFIG: str = Field(default="", env="MY_SECRET_CONFIG")
 
 

code/function/fastapp/main.py

@@ -1,7 +1,7 @@
 from fastapi import FastAPI
 from fastapp.api.v1.api_v1 import api_v1_router
 from fastapp.core.config import settings
-from fastapp.utils import setup_tracer
+from fastapp.utils import setup_opentelemetry
 
 
 def get_app() -> FastAPI:
@@ -25,7 +25,7 @@ def get_app() -> FastAPI:
 @app.on_event("startup")
 async def startup_event():
     """Gracefully start the application before the server reports readiness."""
-    setup_tracer(app=app)
+    setup_opentelemetry(app=app)
 
 
 @app.on_event("shutdown")

code/function/fastapp/utils.py

@@ -1,14 +1,15 @@
 import logging
 from logging import Logger
 
+from azure.monitor.opentelemetry import configure_azure_monitor
+
 # from azure.identity import ManagedIdentityCredential
-from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
 from fastapi import FastAPI
 from fastapp.core.config import settings
-from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
-from opentelemetry.sdk.resources import SERVICE_NAME, Resource
-from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry import trace
+from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
+from opentelemetry.instrumentation.system_metrics import SystemMetricsInstrumentor
+from opentelemetry.trace import Tracer
 
 
 def setup_logging(module) -> Logger:
@@ -21,26 +22,48 @@ def setup_logging(module) -> Logger:
     logger.propagate = False
 
     # Create stream handler
-    logger_stream_handler = logging.StreamHandler()
-    logger_stream_handler.setFormatter(
+    stream_handler = logging.StreamHandler()
+    stream_handler.setFormatter(
         logging.Formatter("[%(asctime)s] [%(levelname)s] [%(module)-8.8s] %(message)s")
     )
-    logger.addHandler(logger_stream_handler)
+    logger.addHandler(stream_handler)
     return logger
 
 
-def setup_tracer(app: FastAPI):
+def setup_tracer(module) -> Tracer:
+    """Setup tracer and event handler.
+
+    RETURNS (Tracer): The tracer object to create spans.
+    """
+    tracer = trace.get_tracer(module)
+    return tracer
+
+
+def setup_opentelemetry(app: FastAPI):
     """Setup tracer for Open Telemetry.
 
     app (FastAPI): The app to be instrumented by Open Telemetry.
     RETURNS (None): Nothing is being returned.
     """
     if settings.APPLICATIONINSIGHTS_CONNECTION_STRING:
         # credential = ManagedIdentityCredential()
-        exporter = AzureMonitorTraceExporter.from_connection_string(
-            settings.APPLICATIONINSIGHTS_CONNECTION_STRING,
-            # credential=credential
+
+        # Configure azure monitor exporter
+        configure_azure_monitor(
+            connection_string=settings.APPLICATIONINSIGHTS_CONNECTION_STRING,
+            disable_offline_storage=False,
+            # credential=credential,
         )
-        tracer = TracerProvider(resource=Resource({SERVICE_NAME: "api"}))
-        tracer.add_span_processor(BatchSpanProcessor(exporter))
-        FastAPIInstrumentor.instrument_app(app, tracer_provider=tracer)
+
+        # Configure custom metrics
+        system_metrics_config = {
+            "system.memory.usage": ["used", "free", "cached"],
+            "system.cpu.time": ["idle", "user", "system", "irq"],
+            "system.network.io": ["transmit", "receive"],
+            "process.runtime.memory": ["rss", "vms"],
+            "process.runtime.cpu.time": ["user", "system"],
+        }
+
+        # Create instrumenter
+        HTTPXClientInstrumentor().instrument()
+        SystemMetricsInstrumentor(config=system_metrics_config).instrument()

code/function/requirements.txt

@@ -2,10 +2,11 @@
 # The Python Worker is managed by Azure Functions platform
 # Manually managing azure-functions-worker may cause unexpected issues
 
-# azure-identity~=1.13.0
+# azure-identity~=1.15.0
 azure-functions~=1.17.0
 fastapi~=0.106.0
 pydantic-settings~=2.1.0
-aiohttp~=3.9.1
-opentelemetry-instrumentation-fastapi==0.43b0
-azure-monitor-opentelemetry-exporter==1.0.0b19
+httpx~=0.26.0
+azure-monitor-opentelemetry~=1.1.1
+opentelemetry-instrumentation-httpx~=0.43b0
+opentelemetry-instrumentation-system-metrics~=0.43b0

code/function/wrapper/__init__.py

@@ -1,6 +1,29 @@
 import azure.functions as func
 from fastapp.main import app
+from fastapp.utils import setup_tracer
+from opentelemetry.context import attach, detach
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
+
+tracer = setup_tracer(__name__)
 
 
 async def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
-    return await func.AsgiMiddleware(app).handle_async(req, context)
+    # Start distributed tracing
+    functions_current_context = {
+        "traceparent": context.trace_context.Traceparent,
+        "tracestate": context.trace_context.Tracestate,
+    }
+    parent_context = TraceContextTextMapPropagator().extract(
+        carrier=functions_current_context
+    )
+    token = attach(parent_context)
+
+    # Function logic
+    try:
+        with tracer.start_as_current_span("wrapper") as span:
+            response = await func.AsgiMiddleware(app).handle_async(req, context)
+    finally:
+        # End distributed tracing
+        detach(token)
+
+    return response