opea-project
diff --git a/‎app-backend/Dockerfile‎
Lines changed: 7 additions & 0 deletions b/‎app-backend/Dockerfile‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎app-backend/micro_service.py‎
Lines changed: 196 additions & 0 deletions b/‎app-backend/micro_service.py‎
Lines changed: 196 additions & 0 deletions
diff --git a/‎app-backend/opea_telemetry.py‎
Lines changed: 72 additions & 0 deletions b/‎app-backend/opea_telemetry.py‎
Lines changed: 72 additions & 0 deletions
@@ -20,9 +20,16 @@ RUN pip install --no-cache-dir --upgrade pip==24.3.1 setuptools==75.3.0 && \
 
 COPY ./templates/microservices/* /home/user/templates/microservices/
 COPY ./megaservice.py /home/user/megaservice.py
+COPY ./orchestrator.py /home/user/GenAIComps/comps/cores/mega/orchestrator.py
 COPY config/* /home/user/config/
 
 ENV PYTHONPATH=/home/user/GenAIComps
+ENV TELEMETRY_ENDPOINT="http://my-otel-collector-opentelemetry-collector.opentelemetry.svc.cluster.local:4318/v1/traces"
+# ENV embedding_tei_langchain_0_port=9007 \
+#     llm_tgi_0_port=9008 \
+#     prepare_doc_redis_prep_0_port=9009 \
+#     reranking_tei_0_port=9010 \
+#     retriever_redis_0_port=9011
 
 USER user
 
 
@@ -0,0 +1,196 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+import os
+from collections import defaultdict, deque
+from enum import Enum
+from typing import Any, List, Optional, Type
+
+from ..proto.docarray import TextDoc
+from .constants import ServiceRoleType, ServiceType
+from .http_service import HTTPService
+from .logger import CustomLogger
+from .utils import check_ports_availability
+
+opea_microservices = {}
+
+logger = CustomLogger("micro_service")
+logflag = os.getenv("LOGFLAG", False)
+
+
+class MicroService(HTTPService):
+    """MicroService class to create a microservice."""
+
+    def __init__(
+        self,
+        name: str = "",
+        service_role: ServiceRoleType = ServiceRoleType.MICROSERVICE,
+        service_type: ServiceType = ServiceType.LLM,
+        protocol: str = "http",
+        host: str = "localhost",
+        port: int = 8080,
+        api_key: str = None,
+        ssl_keyfile: Optional[str] = None,
+        ssl_certfile: Optional[str] = None,
+        endpoint: Optional[str] = "/",
+        input_datatype: Type[Any] = TextDoc,
+        output_datatype: Type[Any] = TextDoc,
+        provider: Optional[str] = None,
+        provider_endpoint: Optional[str] = None,
+        use_remote_service: Optional[bool] = False,
+        description: Optional[str] = None,
+        dynamic_batching: bool = False,
+        dynamic_batching_timeout: int = 1,
+        dynamic_batching_max_batch_size: int = 32,
+    ):
+        """Init the microservice."""
+        self.service_role = service_role
+        self.service_type = service_type
+        self.protocol = protocol
+        self.host = host
+        self.port = port
+        self.api_key = api_key
+        self.endpoint = endpoint
+        self.input_datatype = input_datatype
+        self.output_datatype = output_datatype
+        self.use_remote_service = use_remote_service
+        self.description = description
+        self.dynamic_batching = dynamic_batching
+        self.dynamic_batching_timeout = dynamic_batching_timeout
+        self.dynamic_batching_max_batch_size = dynamic_batching_max_batch_size
+        self.uvicorn_kwargs = {}
+
+        if ssl_keyfile:
+            self.uvicorn_kwargs["ssl_keyfile"] = ssl_keyfile
+
+        if ssl_certfile:
+            self.uvicorn_kwargs["ssl_certfile"] = ssl_certfile
+
+        if not use_remote_service:
+
+            if self.protocol.lower() == "http":
+                if not (check_ports_availability(self.host, self.port)):
+                    raise RuntimeError(f"port:{self.port}")
+
+            self.provider = provider
+            self.provider_endpoint = provider_endpoint
+            self.endpoints = []
+
+            runtime_args = {
+                "protocol": self.protocol,
+                "host": self.host,
+                "port": self.port,
+                "title": name,
+                "description": "OPEA Microservice Infrastructure",
+            }
+
+            super().__init__(uvicorn_kwargs=self.uvicorn_kwargs, runtime_args=runtime_args)
+
+            # create a batch request processor loop if using dynamic batching
+            if self.dynamic_batching:
+                self.buffer_lock = asyncio.Lock()
+                self.request_buffer = defaultdict(deque)
+                self.add_startup_event(self._dynamic_batch_processor())
+
+            self._async_setup()
+
+        # overwrite name
+        self.name = f"{name}/{self.__class__.__name__}" if name else self.__class__.__name__
+
+    async def _dynamic_batch_processor(self):
+        if logflag:
+            logger.info("dynamic batch processor looping...")
+        while True:
+            await asyncio.sleep(self.dynamic_batching_timeout)
+            runtime_batch: dict[Enum, list[dict]] = {}  # {ServiceType.Embedding: [{"request": xx, "response": yy}, {}]}
+
+            async with self.buffer_lock:
+                # prepare the runtime batch, access to buffer is locked
+                if self.request_buffer:
+                    for service_type, request_lst in self.request_buffer.items():
+                        batch = []
+                        # grab min(MAX_BATCH_SIZE, REQUEST_SIZE) requests from buffer
+                        for _ in range(min(self.dynamic_batching_max_batch_size, len(request_lst))):
+                            batch.append(request_lst.popleft())
+
+                        runtime_batch[service_type] = batch
+
+            # Run batched inference on the batch and set results
+            for service_type, batch in runtime_batch.items():
+                if not batch:
+                    continue
+                results = await self.dynamic_batching_infer(service_type, batch)
+
+                for req, result in zip(batch, results):
+                    req["response"].set_result(result)
+
+    async def dynamic_batching_infer(self, service_type: Enum, batch: list[dict]):
+        """Need to implement."""
+        raise NotImplementedError("Unimplemented dynamic batching inference!")
+
+    def _validate_env(self):
+        """Check whether to use the microservice locally."""
+        if self.use_remote_service:
+            raise Exception(
+                "Method not allowed for a remote service, please "
+                "set use_remote_service to False if you want to use a local micro service!"
+            )
+
+    @property
+    def endpoint_path(self):
+        if self.api_key:
+            return f"{self.host}{self.endpoint}"
+        else:
+            return f"{self.protocol}://{self.host}:{self.port}{self.endpoint}"
+
+    @property
+    def api_key_value(self):
+        return self.api_key
+
+
+def register_microservice(
+    name: str,
+    service_role: ServiceRoleType = ServiceRoleType.MICROSERVICE,
+    service_type: ServiceType = ServiceType.UNDEFINED,
+    protocol: str = "http",
+    host: str = "localhost",
+    port: int = 8080,
+    ssl_keyfile: Optional[str] = None,
+    ssl_certfile: Optional[str] = None,
+    endpoint: Optional[str] = "/",
+    input_datatype: Type[Any] = TextDoc,
+    output_datatype: Type[Any] = TextDoc,
+    provider: Optional[str] = None,
+    provider_endpoint: Optional[str] = None,
+    methods: List[str] = ["POST"],
+    dynamic_batching: bool = False,
+    dynamic_batching_timeout: int = 1,
+    dynamic_batching_max_batch_size: int = 32,
+):
+    def decorator(func):
+        if name not in opea_microservices:
+            micro_service = MicroService(
+                name=name,
+                service_role=service_role,
+                service_type=service_type,
+                protocol=protocol,
+                host=host,
+                port=port,
+                ssl_keyfile=ssl_keyfile,
+                ssl_certfile=ssl_certfile,
+                endpoint=endpoint,
+                input_datatype=input_datatype,
+                output_datatype=output_datatype,
+                provider=provider,
+                provider_endpoint=provider_endpoint,
+                dynamic_batching=dynamic_batching,
+                dynamic_batching_timeout=dynamic_batching_timeout,
+                dynamic_batching_max_batch_size=dynamic_batching_max_batch_size,
+            )
+            opea_microservices[name] = micro_service
+        opea_microservices[name].app.router.add_api_route(endpoint, func, methods=methods)
+
+        return func
+
+    return decorator
@@ -0,0 +1,72 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import contextlib
+import inspect
+import os
+from functools import wraps
+
+from opentelemetry import trace
+from opentelemetry.context.contextvars_context import ContextVarsRuntimeContext
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HTTPSpanExporter
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+from ..mega.logger import CustomLogger
+
+logger = CustomLogger("OpeaComponent")
+
+
+def detach_ignore_err(self, token: object) -> None:
+    """Resets Context to a previous value.
+
+    Args:
+        token: A reference to a previous Context.
+    """
+    try:
+        self._current_context.reset(token)  # type: ignore
+    except Exception as e:
+        pass
+
+
+# bypass the ValueError that ContextVar context was created in a different Context from StreamingResponse
+ContextVarsRuntimeContext.detach = detach_ignore_err
+
+resource = Resource.create({SERVICE_NAME: "opea"})
+traceProvider = TracerProvider(resource=resource)
+
+ENABLE_OPEA_TELEMETRY = False
+telemetry_endpoint = os.environ.get("TELEMETRY_ENDPOINT")
+if telemetry_endpoint is not None:
+
+    ENABLE_OPEA_TELEMETRY = True
+    logger.info(f" Has Telemetry Endpoint :  {telemetry_endpoint}")
+    traceProvider.add_span_processor(BatchSpanProcessor(HTTPSpanExporter(endpoint=telemetry_endpoint)))
+
+in_memory_exporter = InMemorySpanExporter()
+traceProvider.add_span_processor(BatchSpanProcessor(in_memory_exporter))
+trace.set_tracer_provider(traceProvider)
+
+tracer = trace.get_tracer(__name__)
+
+
+def opea_telemetry(func):
+    if inspect.iscoroutinefunction(func):
+
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            with tracer.start_as_current_span(func.__name__) if ENABLE_OPEA_TELEMETRY else contextlib.nullcontext():
+                res = await func(*args, **kwargs)
+            return res
+
+    else:
+
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            with tracer.start_as_current_span(func.__name__) if ENABLE_OPEA_TELEMETRY else contextlib.nullcontext():
+                res = func(*args, **kwargs)
+            return res
+
+    return wrapper