Add otel metrics support as well

Author: rbiseck3

requirements/cli.txt

@@ -20,11 +20,16 @@ dataclasses-json==0.6.7
 deprecated==1.2.14
     # via
     #   opentelemetry-api
+    #   opentelemetry-exporter-otlp-proto-grpc
     #   opentelemetry-semantic-conventions
 exceptiongroup==1.2.2
     # via anyio
 fastapi==0.112.2
     # via -r requirements//cli.in
+googleapis-common-protos==1.65.0
+    # via opentelemetry-exporter-otlp-proto-grpc
+grpcio==1.66.0
+    # via opentelemetry-exporter-otlp-proto-grpc
 h11==0.14.0
     # via uvicorn
 idna==3.8
@@ -39,10 +44,16 @@ numpy==2.1.0
     # via pandas
 opentelemetry-api==1.26.0
     # via
+    #   opentelemetry-exporter-otlp-proto-grpc
     #   opentelemetry-instrumentation
     #   opentelemetry-instrumentation-asgi
     #   opentelemetry-instrumentation-fastapi
+    #   opentelemetry-sdk
     #   opentelemetry-semantic-conventions
+opentelemetry-exporter-otlp-proto-common==1.26.0
+    # via opentelemetry-exporter-otlp-proto-grpc
+opentelemetry-exporter-otlp-proto-grpc==1.26.0
+    # via -r requirements//cli.in
 opentelemetry-instrumentation==0.47b0
     # via
     #   opentelemetry-instrumentation-asgi
@@ -51,10 +62,19 @@ opentelemetry-instrumentation-asgi==0.47b0
     # via opentelemetry-instrumentation-fastapi
 opentelemetry-instrumentation-fastapi==0.47b0
     # via -r requirements//cli.in
+opentelemetry-proto==1.26.0
+    # via
+    #   opentelemetry-exporter-otlp-proto-common
+    #   opentelemetry-exporter-otlp-proto-grpc
+opentelemetry-sdk==1.26.0
+    # via
+    #   opentelemetry-exporter-otlp-proto-grpc
+    #   unstructured-ingest
 opentelemetry-semantic-conventions==0.47b0
     # via
     #   opentelemetry-instrumentation-asgi
     #   opentelemetry-instrumentation-fastapi
+    #   opentelemetry-sdk
 opentelemetry-util-http==0.47b0
     # via
     #   opentelemetry-instrumentation-asgi
@@ -63,6 +83,10 @@ packaging==24.1
     # via marshmallow
 pandas==2.2.2
     # via unstructured-ingest
+protobuf==4.25.4
+    # via
+    #   googleapis-common-protos
+    #   opentelemetry-proto
 pydantic==2.8.2
     # via
     #   fastapi
@@ -88,6 +112,7 @@ typing-extensions==4.12.2
     #   anyio
     #   asgiref
     #   fastapi
+    #   opentelemetry-sdk
     #   pydantic
     #   pydantic-core
     #   typing-inspect
@@ -96,7 +121,7 @@ typing-inspect==0.9.0
     # via dataclasses-json
 tzdata==2024.1
     # via pandas
-unstructured-ingest==0.0.6
+unstructured-ingest==0.0.8
     # via -r requirements//cli.in
 uvicorn==0.30.6
     # via -r requirements//cli.in

requirements/constraints.txt

@@ -1 +1 @@
-unstructured-ingest==0.0.0
+unstructured-ingest==0.0.8

test/test_schema.py

@@ -554,6 +554,10 @@ def fn(a: FileData) -> list[FileData]:
                                 ],
                                 "default": None,
                             },
+                            "filesize_bytes": {
+                                "anyOf": [{"type": "integer"}, {"type": "null"}],
+                                "default": None,
+                            },
                         },
                         "required": [],
                     },
@@ -643,6 +647,10 @@ def fn(a: FileData) -> list[FileData]:
                             ],
                             "default": None,
                         },
+                        "filesize_bytes": {
+                            "anyOf": [{"type": "integer"}, {"type": "null"}],
+                            "default": None,
+                        },
                     },
                     "required": [],
                 },

unstructured_platform_plugins/etl_uvicorn/api_generator.py

@@ -11,7 +11,7 @@
 from starlette.responses import RedirectResponse
 from uvicorn.importer import import_from_string
 
-from unstructured_platform_plugins.etl_uvicorn.otel import get_provider
+from unstructured_platform_plugins.etl_uvicorn.otel import get_metric_provider, get_trace_provider
 from unstructured_platform_plugins.etl_uvicorn.utils import (
     get_func,
     get_input_schema,
@@ -172,6 +172,8 @@ async def get_id() -> str:
     except TypeError as e:
         raise TypeError(f"failed to validate function schema: {e}") from e
 
-    FastAPIInstrumentor.instrument_app(fastapi_app, tracer_provider=get_provider())
+    FastAPIInstrumentor.instrument_app(
+        fastapi_app, tracer_provider=get_trace_provider(), meter_provider=get_metric_provider()
+    )
 
     return fastapi_app

unstructured_platform_plugins/etl_uvicorn/otel.py

@@ -1,41 +1,67 @@
 import os
 from typing import Literal, TypedDict
 
-from opentelemetry.environment_variables import OTEL_TRACES_EXPORTER
+from opentelemetry.environment_variables import OTEL_METRICS_EXPORTER, OTEL_TRACES_EXPORTER
 from opentelemetry.sdk.environment_variables import OTEL_SERVICE_NAME
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import (
+    ConsoleMetricExporter,
+    MetricReader,
+    PeriodicExportingMetricReader,
+)
 from opentelemetry.sdk.resources import SERVICE_NAME, Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import (
     ConsoleSpanExporter,
     SimpleSpanProcessor,
 )
 
-ExporterType = Literal["otlp", "jaeger", "zipkin", "console"]
+TraceExporterType = Literal["otlp", "jaeger", "zipkin", "console"]
+MetricExporterType = Literal["otlp", "prometheus", "none"]
 
 
 class OtelSettings(TypedDict):
     service_name: str
-    exporters: list[ExporterType]
+    trace_exporters: list[TraceExporterType]
+    metric_exporters: list[MetricExporterType]
 
 
 def get_settings() -> OtelSettings:
     service_name = os.environ.get(OTEL_SERVICE_NAME, "unknown_service")
-    exporters = os.environ.get(OTEL_TRACES_EXPORTER)
-    exporters = exporters.split(",") if exporters else []
-    return OtelSettings(service_name=service_name, exporters=exporters)
+    trace_exporters = os.environ.get(OTEL_TRACES_EXPORTER)
+    trace_exporters = trace_exporters.split(",") if trace_exporters else []
+
+    metric_exporters = os.environ.get(OTEL_METRICS_EXPORTER)
+    metric_exporters = metric_exporters.split(",") if metric_exporters else []
+    return OtelSettings(
+        service_name=service_name,
+        trace_exporters=trace_exporters,
+        metric_exporters=metric_exporters,
+    )
 
 
-def get_provider() -> TracerProvider:
+def get_trace_provider() -> TracerProvider:
     settings = get_settings()
+    print(settings)
     provider = TracerProvider(resource=Resource({SERVICE_NAME: settings["service_name"]}))
 
-    for exporter_type in settings["exporters"]:
-        _add_exporter(exporter_type=exporter_type, provider=provider)
+    for trace_exporter_type in settings["trace_exporters"]:
+        _add_trace_exporter(exporter_type=trace_exporter_type, provider=provider)
 
     return provider
 
 
-def _add_exporter(exporter_type: ExporterType, provider: TracerProvider):
+def get_metric_provider() -> MeterProvider:
+    settings = get_settings()
+    readers = []
+    for metric_exporter_type in settings["metric_exporters"]:
+        readers.append(_get_metrics_reader(exporter_type=metric_exporter_type))
+    return MeterProvider(
+        resource=Resource({SERVICE_NAME: settings["service_name"]}), metric_readers=readers
+    )
+
+
+def _add_trace_exporter(exporter_type: TraceExporterType, provider: TracerProvider):
     if exporter_type == "otlp":
         _add_traces_otlp_exporter(
             provider,
@@ -47,6 +73,14 @@ def _add_exporter(exporter_type: ExporterType, provider: TracerProvider):
         raise NotImplementedError(f"{exporter_type} implementation not supported yet")
 
 
+def _get_metrics_reader(exporter_type: MetricExporterType) -> MetricReader:
+    if exporter_type == "otlp":
+        return _get_metric_otlp_reader()
+    if exporter_type == "console":
+        return _get_metric_console_reader()
+    raise NotImplementedError(f"{exporter_type} implementation not supported yet")
+
+
 def _add_traces_console_exporter(provider: TracerProvider) -> None:
     exporter = ConsoleSpanExporter()
     processor = SimpleSpanProcessor(exporter)
@@ -59,3 +93,15 @@ def _add_traces_otlp_exporter(provider: TracerProvider) -> None:
     exporter = OTLPSpanExporter()
     processor = SimpleSpanProcessor(exporter)
     provider.add_span_processor(processor)
+
+
+def _get_metric_otlp_reader() -> MetricReader:
+    from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
+
+    exporter = OTLPMetricExporter()
+    return PeriodicExportingMetricReader(exporter)
+
+
+def _get_metric_console_reader() -> MetricReader:
+    exporter = ConsoleMetricExporter()
+    return PeriodicExportingMetricReader(exporter)