Merge of arcade up and dev (#77)
Combination of these PRs: https://github.com/ArcadeAI/arcade-ai/pull/74 https://github.com/ArcadeAI/arcade-ai/pull/76
This commit is contained in:
Sam Partee
GitHub
parent
83cf070c82
commit
ef72e6c5aa
14 changed files with 418 additions and 141 deletions
8
Makefile
8
Makefile
|
|
@ -52,14 +52,6 @@ publish: ## publish a release to pypi.
|
|||
.PHONY: build-and-publish
|
||||
build-and-publish: build publish ## Build and publish.
|
||||
|
||||
.PHONY: docs-test
|
||||
docs-test: ## Test if documentation can be built without warnings or errors
|
||||
@cd arcade && poetry run mkdocs build -s
|
||||
|
||||
.PHONY: docs
|
||||
docs: ## Build and serve the documentation
|
||||
@cd arcade && poetry run mkdocs serve -a localhost:8777
|
||||
|
||||
.PHONY: docker
|
||||
docker: ## Build and run the Docker container
|
||||
@cd docker && make docker-build
|
||||
|
|
|
|||
|
|
@ -4,6 +4,9 @@ import time
|
|||
from datetime import datetime
|
||||
from typing import Any, Callable, ClassVar
|
||||
|
||||
from opentelemetry import trace
|
||||
from opentelemetry.metrics import Meter
|
||||
|
||||
from arcade.actor.core.common import Actor, Router
|
||||
from arcade.actor.core.components import (
|
||||
ActorComponent,
|
||||
|
|
@ -36,7 +39,9 @@ class BaseActor(Actor):
|
|||
HealthCheckComponent,
|
||||
)
|
||||
|
||||
def __init__(self, secret: str | None = None, disable_auth: bool = False) -> None:
|
||||
def __init__(
|
||||
self, secret: str | None = None, disable_auth: bool = False, otel_meter: Meter | None = None
|
||||
) -> None:
|
||||
"""
|
||||
Initialize the BaseActor with an empty ToolCatalog.
|
||||
If no secret is provided, the actor will use the ARCADE_ACTOR_SECRET environment variable.
|
||||
|
|
@ -49,6 +54,11 @@ class BaseActor(Actor):
|
|||
)
|
||||
|
||||
self.secret = self._set_secret(secret, disable_auth)
|
||||
self.environment = os.environ.get("ARCADE_ENVIRONMENT", "local")
|
||||
if otel_meter:
|
||||
self.tool_counter = otel_meter.create_counter(
|
||||
"tool_call", "requests", "Total number of tools called"
|
||||
)
|
||||
|
||||
def _set_secret(self, secret: str | None, disable_auth: bool) -> str:
|
||||
if disable_auth:
|
||||
|
|
@ -100,20 +110,52 @@ class BaseActor(Actor):
|
|||
|
||||
start_time = time.time()
|
||||
|
||||
output = await ToolExecutor.run(
|
||||
func=materialized_tool.tool,
|
||||
definition=materialized_tool.definition,
|
||||
input_model=materialized_tool.input_model,
|
||||
output_model=materialized_tool.output_model,
|
||||
context=tool_request.context,
|
||||
**tool_request.inputs or {},
|
||||
if self.tool_counter:
|
||||
self.tool_counter.add(
|
||||
1,
|
||||
{
|
||||
"tool_name": tool_fqname.name,
|
||||
"toolkit_version": str(tool_fqname.toolkit_version),
|
||||
"toolkit_name": tool_fqname.toolkit_name,
|
||||
"environment": self.environment,
|
||||
},
|
||||
)
|
||||
invocation_id = tool_request.invocation_id or ""
|
||||
logger.info(
|
||||
f"{invocation_id} | Calling tool: {tool_fqname} version: {tool_request.tool.version}"
|
||||
)
|
||||
logger.debug(f"{invocation_id} | Tool inputs: {tool_request.inputs}")
|
||||
|
||||
tracer = trace.get_tracer(__name__)
|
||||
with tracer.start_as_current_span("RunTool"):
|
||||
output = await ToolExecutor.run(
|
||||
func=materialized_tool.tool,
|
||||
definition=materialized_tool.definition,
|
||||
input_model=materialized_tool.input_model,
|
||||
output_model=materialized_tool.output_model,
|
||||
context=tool_request.context,
|
||||
**tool_request.inputs or {},
|
||||
)
|
||||
|
||||
if output.error:
|
||||
logger.warning(
|
||||
f"{invocation_id} | Tool {tool_fqname} version {tool_request.tool.version} failed"
|
||||
)
|
||||
logger.warning(f"{invocation_id} | Tool error: {output.error.message}")
|
||||
logger.debug(
|
||||
f"{invocation_id} | Tool developer message: {output.error.developer_message}"
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
f"{invocation_id} | Tool {tool_fqname} version {tool_request.tool.version} success"
|
||||
)
|
||||
logger.debug(f"{invocation_id} | Tool output: {output}")
|
||||
|
||||
end_time = time.time() # End time in seconds
|
||||
duration_ms = (end_time - start_time) * 1000 # Convert to milliseconds
|
||||
|
||||
return ToolCallResponse(
|
||||
invocation_id=tool_request.invocation_id or "",
|
||||
invocation_id=invocation_id,
|
||||
duration=duration_ms,
|
||||
finished_at=datetime.now().isoformat(),
|
||||
success=not output.error,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,7 @@
|
|||
from typing import Any
|
||||
|
||||
from opentelemetry import trace
|
||||
|
||||
from arcade.actor.core.common import Actor, ActorComponent, RequestData, Router
|
||||
from arcade.core.schema import ToolCallRequest, ToolCallResponse, ToolDefinition
|
||||
|
||||
|
|
@ -18,7 +20,9 @@ class CatalogComponent(ActorComponent):
|
|||
"""
|
||||
Handle the request to get the catalog.
|
||||
"""
|
||||
return self.actor.get_catalog()
|
||||
tracer = trace.get_tracer(__name__)
|
||||
with tracer.start_as_current_span("Catalog"):
|
||||
return self.actor.get_catalog()
|
||||
|
||||
|
||||
class CallToolComponent(ActorComponent):
|
||||
|
|
@ -35,9 +39,11 @@ class CallToolComponent(ActorComponent):
|
|||
"""
|
||||
Handle the request to call (invoke) a tool.
|
||||
"""
|
||||
call_tool_request_data = request.body_json
|
||||
call_tool_request = ToolCallRequest.model_validate(call_tool_request_data)
|
||||
return await self.actor.call_tool(call_tool_request)
|
||||
tracer = trace.get_tracer(__name__)
|
||||
with tracer.start_as_current_span("CallTool"):
|
||||
call_tool_request_data = request.body_json
|
||||
call_tool_request = ToolCallRequest.model_validate(call_tool_request_data)
|
||||
return await self.actor.call_tool(call_tool_request)
|
||||
|
||||
|
||||
class HealthCheckComponent(ActorComponent):
|
||||
|
|
@ -54,4 +60,6 @@ class HealthCheckComponent(ActorComponent):
|
|||
"""
|
||||
Handle the request for a health check.
|
||||
"""
|
||||
return self.actor.health_check()
|
||||
tracer = trace.get_tracer(__name__)
|
||||
with tracer.start_as_current_span("HealthCheck"):
|
||||
return self.actor.health_check()
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ from typing import Any, Callable
|
|||
|
||||
from fastapi import Depends, FastAPI, Request
|
||||
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
||||
from opentelemetry.metrics import Meter
|
||||
|
||||
from arcade.actor.core.base import (
|
||||
BaseActor,
|
||||
|
|
@ -19,13 +20,18 @@ class FastAPIActor(BaseActor):
|
|||
"""
|
||||
|
||||
def __init__(
|
||||
self, app: FastAPI, secret: str | None = None, *, disable_auth: bool = False
|
||||
self,
|
||||
app: FastAPI,
|
||||
secret: str | None = None,
|
||||
*,
|
||||
disable_auth: bool = False,
|
||||
otel_meter: Meter | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Initialize the FastAPIActor with a FastAPI app instance.
|
||||
If no secret is provided, the actor will use the ARCADE_ACTOR_SECRET environment variable.
|
||||
"""
|
||||
super().__init__(secret, disable_auth)
|
||||
super().__init__(secret, disable_auth, otel_meter)
|
||||
self.app = app
|
||||
self.router = FastAPIRouter(app, self)
|
||||
self.register_routes(self.router)
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ def start_servers(
|
|||
port: int,
|
||||
engine_config: str | None,
|
||||
engine_env: dict[str, str] | None = None,
|
||||
debug: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Start the actor and engine servers.
|
||||
|
|
@ -39,11 +40,11 @@ def start_servers(
|
|||
engine_config = _get_engine_config(engine_config)
|
||||
|
||||
# Prepare command-line arguments for the actor server and engine
|
||||
actor_cmd = _build_actor_command(host, port)
|
||||
actor_cmd = _build_actor_command(host, port, debug)
|
||||
engine_cmd = _build_engine_command(engine_config)
|
||||
|
||||
# Start and manage the processes
|
||||
_manage_processes(actor_cmd, engine_cmd, engine_env)
|
||||
_manage_processes(actor_cmd, engine_cmd, engine_env, debug)
|
||||
|
||||
|
||||
def _validate_host(host: str) -> str:
|
||||
|
|
@ -109,6 +110,12 @@ def _get_engine_config(engine_config: str | None) -> str:
|
|||
f"❌ Engine config file not found at {engine_config_path}", style="bold red"
|
||||
)
|
||||
raise RuntimeError("Engine config file not found.")
|
||||
|
||||
elif Path(os.path.expanduser("~/.arcade/engine.yaml")).is_file():
|
||||
engine_config_path = Path(os.path.expanduser("~/.arcade/engine.yaml"))
|
||||
console.print(
|
||||
f"Using default engine config file at {engine_config_path}", style="bold green"
|
||||
)
|
||||
else:
|
||||
# Look for engine.yaml in the current directory
|
||||
engine_config_path = Path(os.getcwd()) / "engine.yaml"
|
||||
|
|
@ -121,13 +128,14 @@ def _get_engine_config(engine_config: str | None) -> str:
|
|||
return str(engine_config_path)
|
||||
|
||||
|
||||
def _build_actor_command(host: str, port: int) -> list[str]:
|
||||
def _build_actor_command(host: str, port: int, debug: bool) -> list[str]:
|
||||
"""
|
||||
Builds the command to start the actor server.
|
||||
|
||||
Args:
|
||||
host: Host for the actor server.
|
||||
port: Port for the actor server.
|
||||
debug: Whether to run in debug mode.
|
||||
|
||||
Returns:
|
||||
The command as a list.
|
||||
|
|
@ -142,12 +150,14 @@ def _build_actor_command(host: str, port: int) -> list[str]:
|
|||
sys.exit(1)
|
||||
cmd = [
|
||||
arcade_bin,
|
||||
"dev",
|
||||
"actorup",
|
||||
"--host",
|
||||
host,
|
||||
"--port",
|
||||
str(port),
|
||||
]
|
||||
if debug:
|
||||
cmd.append("--debug")
|
||||
return cmd
|
||||
|
||||
|
||||
|
|
@ -165,13 +175,12 @@ def _build_engine_command(engine_config: str) -> list[str]:
|
|||
if not engine_bin:
|
||||
console.print(
|
||||
"❌ Engine binary not found, refer to the installation guide at "
|
||||
"https://docs.arcade-ai.com/docs/home/deployment for how to install the engine",
|
||||
"https://docs.arcade-ai.com/guides/installation for how to install the engine",
|
||||
style="bold red",
|
||||
)
|
||||
sys.exit(1)
|
||||
cmd = [
|
||||
engine_bin,
|
||||
"dev",
|
||||
"-c",
|
||||
engine_config,
|
||||
]
|
||||
|
|
@ -179,7 +188,10 @@ def _build_engine_command(engine_config: str) -> list[str]:
|
|||
|
||||
|
||||
def _manage_processes(
|
||||
actor_cmd: list[str], engine_cmd: list[str], engine_env: dict[str, str] | None = None
|
||||
actor_cmd: list[str],
|
||||
engine_cmd: list[str],
|
||||
engine_env: dict[str, str] | None = None,
|
||||
debug: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Manages the lifecycle of the actor and engine processes.
|
||||
|
|
@ -188,6 +200,7 @@ def _manage_processes(
|
|||
actor_cmd: The command to start the actor server.
|
||||
engine_cmd: The command to start the engine.
|
||||
engine_env: Environment variables to set for the engine.
|
||||
debug: Whether to run in debug mode.
|
||||
"""
|
||||
actor_process: subprocess.Popen | None = None
|
||||
engine_process: subprocess.Popen | None = None
|
||||
|
|
@ -208,14 +221,14 @@ def _manage_processes(
|
|||
try:
|
||||
# Start the actor server
|
||||
console.print("Starting actor server...", style="bold green")
|
||||
actor_process = _start_process("Actor", actor_cmd)
|
||||
actor_process = _start_process("Actor", actor_cmd, debug=debug)
|
||||
|
||||
# Wait a bit to ensure actor is up
|
||||
time.sleep(2)
|
||||
|
||||
# Start the engine
|
||||
console.print("Starting engine...", style="bold green")
|
||||
engine_process = _start_process("Engine", engine_cmd, engine_env)
|
||||
engine_process = _start_process("Engine", engine_cmd, env=engine_env, debug=debug)
|
||||
|
||||
# Monitor processes
|
||||
_monitor_processes(actor_process, engine_process)
|
||||
|
|
@ -248,7 +261,7 @@ def _manage_processes(
|
|||
|
||||
|
||||
def _start_process(
|
||||
name: str, cmd: list[str], env: dict[str, str] | None = None
|
||||
name: str, cmd: list[str], env: dict[str, str] | None = None, debug: bool = False
|
||||
) -> subprocess.Popen:
|
||||
"""
|
||||
Starts a subprocess and begins streaming its output.
|
||||
|
|
@ -257,7 +270,7 @@ def _start_process(
|
|||
name: Name of the process.
|
||||
cmd: Command to execute.
|
||||
env: Environment variables to set for the process.
|
||||
|
||||
debug: Whether to run in debug mode.
|
||||
Returns:
|
||||
The subprocess.Popen object.
|
||||
|
||||
|
|
@ -268,8 +281,13 @@ def _start_process(
|
|||
if env:
|
||||
_env.update(env)
|
||||
|
||||
# TODO temporary fix for GIN_MODE
|
||||
_env["GIN_MODE"] = "release"
|
||||
if debug:
|
||||
_env["GIN_MODE"] = "debug"
|
||||
else:
|
||||
_env["GIN_MODE"] = "release"
|
||||
|
||||
if name == "Actor":
|
||||
_env["PYTHONUNBUFFERED"] = "1"
|
||||
|
||||
try:
|
||||
process = subprocess.Popen( # noqa: S603, RUF100
|
||||
|
|
@ -303,7 +321,12 @@ def _stream_output(process: subprocess.Popen, name: str) -> None:
|
|||
return
|
||||
with pipe:
|
||||
for line in iter(pipe.readline, ""):
|
||||
console.print(f"[{style}]{name}>[/{style}] {line.rstrip()}")
|
||||
line = line.rstrip()
|
||||
if "WARNING" in line:
|
||||
line = line.replace("WARNING", "[orange]WARNING[/orange]")
|
||||
if "ERROR" in line:
|
||||
line = line.replace("ERROR", "[red]ERROR[/red]")
|
||||
console.print(f"[{style}]{name}>[/{style}] {line}")
|
||||
|
||||
threading.Thread(target=stream, args=(process.stdout, stdout_style), daemon=True).start()
|
||||
threading.Thread(target=stream, args=(process.stderr, "red"), daemon=True).start()
|
||||
|
|
|
|||
|
|
@ -284,37 +284,6 @@ def chat(
|
|||
raise typer.Exit()
|
||||
|
||||
|
||||
@cli.command(help="Start a local Arcade Actor server", rich_help_panel="Launch")
|
||||
def dev(
|
||||
host: str = typer.Option(
|
||||
"127.0.0.1", help="Host for the app, from settings by default.", show_default=True
|
||||
),
|
||||
port: int = typer.Option(
|
||||
"8002", "-p", "--port", help="Port for the app, defaults to ", show_default=True
|
||||
),
|
||||
disable_auth: bool = typer.Option(
|
||||
False,
|
||||
"--no-auth",
|
||||
help="Disable authentication for the actor. Not recommended for production.",
|
||||
show_default=True,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Starts the actor with host, port, and reload options. Uses
|
||||
Uvicorn as ASGI actor. Parameters allow runtime configuration.
|
||||
"""
|
||||
from arcade.cli.serve import serve_default_actor
|
||||
|
||||
try:
|
||||
serve_default_actor(host, port, disable_auth)
|
||||
except KeyboardInterrupt:
|
||||
typer.Exit()
|
||||
except Exception as e:
|
||||
error_message = f"❌ Failed to start Arcade Actor: {escape(str(e))}"
|
||||
console.print(error_message, style="bold red")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
|
||||
@cli.command(help="Show/edit the local Arcade configuration", rich_help_panel="User")
|
||||
def config(
|
||||
action: str = typer.Argument("show", help="The action to take (show/edit)"),
|
||||
|
|
@ -500,8 +469,8 @@ def evals(
|
|||
display_eval_results(results, show_details=show_details)
|
||||
|
||||
|
||||
@cli.command(help="Start an Arcade Cluster instance", rich_help_panel="Launch")
|
||||
def up(
|
||||
@cli.command(help="Launch Arcade AI locally for tool dev", rich_help_panel="Launch")
|
||||
def dev(
|
||||
host: str = typer.Option("127.0.0.1", help="Host for the actor server.", show_default=True),
|
||||
port: int = typer.Option(
|
||||
8002, "-p", "--port", help="Port for the actor server.", show_default=True
|
||||
|
|
@ -509,14 +478,47 @@ def up(
|
|||
engine_config: str = typer.Option(
|
||||
None, "-c", "--config", help="Path to the engine configuration file."
|
||||
),
|
||||
debug: bool = typer.Option(False, "-d", "--debug", help="Show debug information"),
|
||||
) -> None:
|
||||
"""
|
||||
Start both the actor and engine servers.
|
||||
"""
|
||||
try:
|
||||
# TODO: pass Engine env vars from here
|
||||
start_servers(host, port, engine_config)
|
||||
start_servers(host, port, engine_config, engine_env=None, debug=debug)
|
||||
except Exception as e:
|
||||
error_message = f"❌ Failed to start servers: {escape(str(e))}"
|
||||
console.print(error_message, style="bold red")
|
||||
typer.Exit(code=1)
|
||||
|
||||
|
||||
@cli.command(help="Start a local Arcade Actor server", rich_help_panel="Launch", hidden=True)
|
||||
def actorup(
|
||||
host: str = typer.Option(
|
||||
"127.0.0.1", help="Host for the app, from settings by default.", show_default=True
|
||||
),
|
||||
port: int = typer.Option(
|
||||
"8002", "-p", "--port", help="Port for the app, defaults to ", show_default=True
|
||||
),
|
||||
disable_auth: bool = typer.Option(
|
||||
False,
|
||||
"--no-auth",
|
||||
help="Disable authentication for the actor. Not recommended for production.",
|
||||
show_default=True,
|
||||
),
|
||||
debug: bool = typer.Option(False, "--debug", "-d", help="Show debug information"),
|
||||
) -> None:
|
||||
"""
|
||||
Starts the actor with host, port, and reload options. Uses
|
||||
Uvicorn as ASGI actor. Parameters allow runtime configuration.
|
||||
"""
|
||||
from arcade.cli.serve import serve_default_actor
|
||||
|
||||
try:
|
||||
serve_default_actor(host, port, disable_auth=disable_auth, debug=debug)
|
||||
except KeyboardInterrupt:
|
||||
typer.Exit()
|
||||
except Exception as e:
|
||||
error_message = f"❌ Failed to start Arcade Actor: {escape(str(e))}"
|
||||
console.print(error_message, style="bold red")
|
||||
typer.Exit(code=1)
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ from typing import Any
|
|||
|
||||
from loguru import logger
|
||||
|
||||
from arcade.core.telemetry import OTELHandler
|
||||
|
||||
try:
|
||||
import fastapi
|
||||
except ImportError:
|
||||
|
|
@ -60,9 +62,9 @@ def setup_logging(log_level: int = logging.INFO) -> None:
|
|||
"sink": sys.stdout,
|
||||
"serialize": False,
|
||||
"level": log_level,
|
||||
"format": "{time:MM-DD HH:mm:ss} | {level: <8} | {message}"
|
||||
"format": "{level} [{time:HH:mm:ss.SSS}] {message}"
|
||||
+ (" {name}:{function}:{line}" if log_level <= logging.DEBUG else "")
|
||||
+ ("{exception}\n" if "{exception}" in "{message}" else ""),
|
||||
+ ("\n{exception}" if "{exception}" in "{message}" else ""),
|
||||
}
|
||||
]
|
||||
)
|
||||
|
|
@ -84,13 +86,15 @@ def serve_default_actor(
|
|||
disable_auth: bool = False,
|
||||
workers: int = 1,
|
||||
timeout_keep_alive: int = 5,
|
||||
enable_otel: bool = False,
|
||||
debug: bool = False,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
"""
|
||||
Get an instance of a FastAPI server with the Arcade Actor.
|
||||
"""
|
||||
# Setup unified logging
|
||||
setup_logging()
|
||||
setup_logging(log_level=logging.DEBUG if debug else logging.INFO)
|
||||
|
||||
toolkits = Toolkit.find_all_arcade_toolkits()
|
||||
if not toolkits:
|
||||
|
|
@ -114,7 +118,12 @@ def serve_default_actor(
|
|||
version="0.1.0",
|
||||
lifespan=lifespan, # Use custom lifespan to catch errors, notably KeyboardInterrupt (Ctrl+C)
|
||||
)
|
||||
actor = FastAPIActor(app, secret=actor_secret, disable_auth=disable_auth)
|
||||
|
||||
otel_handler = OTELHandler(app, enable=enable_otel)
|
||||
|
||||
actor = FastAPIActor(
|
||||
app, secret=actor_secret, disable_auth=disable_auth, otel_meter=otel_handler.get_meter()
|
||||
)
|
||||
for toolkit in toolkits:
|
||||
actor.register_toolkit(toolkit)
|
||||
|
||||
|
|
@ -143,4 +152,6 @@ def serve_default_actor(
|
|||
except KeyboardInterrupt:
|
||||
logger.info("Server stopped by user.")
|
||||
finally:
|
||||
if enable_otel:
|
||||
otel_handler.shutdown()
|
||||
logger.debug("Server shutdown complete.")
|
||||
|
|
|
|||
113
arcade/arcade/core/telemetry.py
Normal file
113
arcade/arcade/core/telemetry.py
Normal file
|
|
@ -0,0 +1,113 @@
|
|||
import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import FastAPI
|
||||
from opentelemetry import _logs, trace
|
||||
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
|
||||
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
|
||||
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
|
||||
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
|
||||
from opentelemetry.metrics import Meter, get_meter_provider, set_meter_provider
|
||||
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
|
||||
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
|
||||
from opentelemetry.sdk.metrics import MeterProvider
|
||||
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
|
||||
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
|
||||
from opentelemetry.sdk.trace import TracerProvider
|
||||
from opentelemetry.sdk.trace.export import BatchSpanProcessor
|
||||
|
||||
|
||||
class ShutdownError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class OTELHandler:
|
||||
def __init__(self, app: FastAPI, enable: bool = True, log_level: int = logging.INFO):
|
||||
self._tracer_provider: Optional[TracerProvider] = None
|
||||
self._tracer_span_exporter: Optional[OTLPSpanExporter] = None
|
||||
self._meter_provider: Optional[MeterProvider] = None
|
||||
self._meter_reader: Optional[PeriodicExportingMetricReader] = None
|
||||
self._otlp_metric_exporter: Optional[OTLPMetricExporter] = None
|
||||
self._logger_provider: Optional[LoggerProvider] = None
|
||||
self._log_processor: Optional[BatchLogRecordProcessor] = None
|
||||
self.environment = os.environ.get("ARCADE_ENVIRONMENT", "local")
|
||||
|
||||
if enable:
|
||||
logging.info(
|
||||
"🔎 Initializing OpenTelemetry. Use environment variables to configure the connection"
|
||||
)
|
||||
self.resource = Resource(
|
||||
attributes={SERVICE_NAME: "arcade-actor", "environment": self.environment}
|
||||
)
|
||||
|
||||
self._init_tracer()
|
||||
self._init_metrics()
|
||||
self._init_logging(log_level)
|
||||
|
||||
FastAPIInstrumentor().instrument_app(app)
|
||||
|
||||
def _init_tracer(self) -> None:
|
||||
self._tracer_provider = TracerProvider(resource=self.resource)
|
||||
trace.set_tracer_provider(self._tracer_provider)
|
||||
|
||||
# Create an OTLP exporter
|
||||
self._tracer_span_exporter = OTLPSpanExporter()
|
||||
|
||||
try:
|
||||
self._tracer_span_exporter.export([trace.get_tracer(__name__).start_span("ping")])
|
||||
except Exception as e:
|
||||
raise ConnectionError(
|
||||
f"Could not connect to OpenTelemetry Tracer endpoint. Check OpenTelemetry configuration or disable: {e}"
|
||||
)
|
||||
|
||||
# Create a batch span processor and add the exporter
|
||||
span_processor = BatchSpanProcessor(self._tracer_span_exporter)
|
||||
self._tracer_provider.add_span_processor(span_processor)
|
||||
|
||||
def _init_metrics(self) -> None:
|
||||
self._otlp_metric_exporter = OTLPMetricExporter()
|
||||
|
||||
self._meter_reader = PeriodicExportingMetricReader(self._otlp_metric_exporter)
|
||||
|
||||
self._meter_provider = MeterProvider(
|
||||
metric_readers=[self._meter_reader], resource=self.resource
|
||||
)
|
||||
|
||||
set_meter_provider(self._meter_provider)
|
||||
|
||||
def get_meter(self) -> Meter:
|
||||
return get_meter_provider().get_meter(__name__)
|
||||
|
||||
def _init_logging(self, log_level: int) -> None:
|
||||
otlp_log_exporter = OTLPLogExporter()
|
||||
|
||||
self._logger_provider = LoggerProvider(resource=self.resource)
|
||||
_logs.set_logger_provider(self._logger_provider)
|
||||
|
||||
# Create a batch span processor and add the exporter
|
||||
self._log_processor = BatchLogRecordProcessor(otlp_log_exporter)
|
||||
self._logger_provider.add_log_record_processor(self._log_processor)
|
||||
|
||||
handler = LoggingHandler(level=log_level, logger_provider=self._logger_provider)
|
||||
logging.getLogger().addHandler(handler)
|
||||
|
||||
def _shutdown_tracer(self) -> None:
|
||||
if self._tracer_span_exporter is None:
|
||||
raise ShutdownError("Tracer provider not initialized. Failed to shutdown")
|
||||
self._tracer_span_exporter.shutdown()
|
||||
|
||||
def _shutdown_metrics(self) -> None:
|
||||
if self._otlp_metric_exporter is None:
|
||||
raise ShutdownError("Meter provider not initialized. Failed to shutdown")
|
||||
self._otlp_metric_exporter.shutdown()
|
||||
|
||||
def _shutdown_logging(self) -> None:
|
||||
if self._logger_provider is None:
|
||||
raise ShutdownError("Log provider not initialized. Failed to shutdown")
|
||||
self._logger_provider.shutdown()
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self._shutdown_tracer()
|
||||
self._shutdown_metrics()
|
||||
self._shutdown_logging()
|
||||
|
|
@ -1,8 +0,0 @@
|
|||
# arcade-ai
|
||||
|
||||
[](https://img.shields.io/github/v/release/spartee/arcade-ai)
|
||||
[](https://github.com/spartee/arcade-ai/actions/workflows/main.yml?query=branch%3Amain)
|
||||
[](https://img.shields.io/github/commit-activity/m/spartee/arcade-ai)
|
||||
[](https://img.shields.io/github/license/spartee/arcade-ai)
|
||||
|
||||
Arcade AI python
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
site_name: arcade-ai
|
||||
repo_url: https://github.com/spartee/arcade-ai
|
||||
site_url: https://spartee.github.io/arcade-ai
|
||||
site_description: Arcade AI python
|
||||
site_author: Arcade AI
|
||||
edit_uri: edit/main/docs/
|
||||
repo_name: spartee/arcade-ai
|
||||
copyright: Maintained by <a href="https://spartee.com">Florian</a>.
|
||||
|
||||
nav:
|
||||
- Home: index.md
|
||||
- Modules: modules.md
|
||||
plugins:
|
||||
- search
|
||||
- mkdocstrings:
|
||||
handlers:
|
||||
python:
|
||||
setup_commands:
|
||||
- import sys
|
||||
- sys.path.append('../')
|
||||
theme:
|
||||
name: material
|
||||
feature:
|
||||
tabs: true
|
||||
palette:
|
||||
- media: "(prefers-color-scheme: light)"
|
||||
scheme: default
|
||||
primary: white
|
||||
accent: deep orange
|
||||
toggle:
|
||||
icon: material/brightness-7
|
||||
name: Switch to dark mode
|
||||
- media: "(prefers-color-scheme: dark)"
|
||||
scheme: slate
|
||||
primary: black
|
||||
accent: deep orange
|
||||
toggle:
|
||||
icon: material/brightness-4
|
||||
name: Switch to light mode
|
||||
icon:
|
||||
repo: fontawesome/brands/github
|
||||
|
||||
extra:
|
||||
social:
|
||||
- icon: fontawesome/brands/github
|
||||
link: https://github.com/spartee/arcade-ai
|
||||
- icon: fontawesome/brands/python
|
||||
link: https://pypi.org/project/arcade-ai
|
||||
|
||||
markdown_extensions:
|
||||
- toc:
|
||||
permalink: true
|
||||
- pymdownx.arithmatex:
|
||||
generic: true
|
||||
|
|
@ -22,6 +22,9 @@ tomlkit = "^0.12.4"
|
|||
openai = "^1.36.0" # TODO: relax to an earlier version that still has what we need
|
||||
pyjwt = "^2.8.0"
|
||||
loguru = "^0.7.0"
|
||||
opentelemetry-instrumentation-fastapi = {version = "0.48b0", optional = true}
|
||||
opentelemetry-exporter-otlp-proto-http = {version = "1.27.0", optional = true}
|
||||
opentelemetry-exporter-otlp-proto-common = {version = "1.27.0", optional = true}
|
||||
fastapi = {version = "^0.110.0", optional = true}
|
||||
uvicorn = {version = "^0.30.0", optional = true}
|
||||
scipy = {version = "^1.14.0", optional = true}
|
||||
|
|
@ -29,7 +32,7 @@ numpy = {version = "^2.0.0", optional = true}
|
|||
scikit-learn = {version = "^1.5.0", optional = true}
|
||||
|
||||
[tool.poetry.extras]
|
||||
fastapi = ["fastapi", "uvicorn"]
|
||||
fastapi = ["fastapi", "uvicorn", "opentelemetry-instrumentation-fastapi", "opentelemetry-exporter-otlp-proto-http", "opentelemetry-exporter-otlp-proto-common"]
|
||||
evals = ["scipy", "numpy", "scikit-learn"]
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
|
|
|
|||
138
arcade/tests/core/test_telemetry.py
Normal file
138
arcade/tests/core/test_telemetry.py
Normal file
|
|
@ -0,0 +1,138 @@
|
|||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
|
||||
from arcade.core.telemetry import OTELHandler, ShutdownError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app():
|
||||
return FastAPI()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def handler_disabled(app):
|
||||
return OTELHandler(app, enable=False)
|
||||
|
||||
|
||||
@patch("arcade.core.telemetry.logging")
|
||||
@patch("arcade.core.telemetry.FastAPIInstrumentor")
|
||||
@patch("arcade.core.telemetry.OTLPLogExporter")
|
||||
@patch("arcade.core.telemetry.OTLPMetricExporter")
|
||||
@patch("arcade.core.telemetry.OTLPSpanExporter")
|
||||
def test_init_with_enable_true(
|
||||
mock_span_exporter,
|
||||
mock_metric_exporter,
|
||||
mock_log_exporter,
|
||||
mock_instrumentor,
|
||||
mock_logging,
|
||||
app,
|
||||
):
|
||||
# Mock the methods that may cause network calls
|
||||
mock_span_exporter.return_value.shutdown = MagicMock()
|
||||
mock_metric_exporter.return_value.shutdown = MagicMock()
|
||||
mock_log_exporter.return_value.shutdown = MagicMock()
|
||||
|
||||
# Initialize OTELHandler within the scope of the mocks
|
||||
handler = OTELHandler(app, enable=True)
|
||||
|
||||
# Verify that the resource is set correctly
|
||||
assert handler.resource.attributes["service.name"] == "arcade-actor"
|
||||
assert "environment" in handler.resource.attributes
|
||||
|
||||
# Verify that initialization methods are called
|
||||
assert handler._tracer_provider is not None
|
||||
assert handler._tracer_span_exporter is not None
|
||||
assert handler._meter_provider is not None
|
||||
assert handler._meter_reader is not None
|
||||
assert handler._logger_provider is not None
|
||||
assert handler._log_processor is not None
|
||||
|
||||
# Verify that FastAPIInstrumentor is used
|
||||
mock_instrumentor.return_value.instrument_app.assert_called_once_with(app)
|
||||
|
||||
|
||||
@patch("arcade.core.telemetry.logging")
|
||||
@patch("arcade.core.telemetry.FastAPIInstrumentor")
|
||||
def test_init_with_enable_false(mock_instrumentor, mock_logging, app):
|
||||
handler = OTELHandler(app, enable=False)
|
||||
|
||||
# Verify that resources are not initialized
|
||||
assert handler._tracer_provider is None
|
||||
assert handler._tracer_span_exporter is None
|
||||
assert handler._meter_provider is None
|
||||
assert handler._meter_reader is None
|
||||
assert handler._logger_provider is None
|
||||
assert handler._log_processor is None
|
||||
|
||||
# Verify that FastAPIInstrumentor is not called
|
||||
mock_instrumentor.return_value.instrument_app.assert_not_called()
|
||||
|
||||
|
||||
def test_init_tracer_export_exception(app):
|
||||
# Simulate an exception during exporter initialization
|
||||
|
||||
with pytest.raises(ConnectionError) as exc_info:
|
||||
OTELHandler(app, enable=True)
|
||||
|
||||
assert "Could not connect to OpenTelemetry Tracer endpoint" in str(exc_info.value)
|
||||
|
||||
|
||||
@patch("arcade.core.telemetry.OTLPLogExporter")
|
||||
@patch("arcade.core.telemetry.OTLPMetricExporter")
|
||||
@patch("arcade.core.telemetry.OTLPSpanExporter")
|
||||
def test_shutdown(mock_span_exporter, mock_metric_exporter, mock_log_exporter, app):
|
||||
# Mock the shutdown methods
|
||||
mock_span_exporter.return_value.shutdown = MagicMock()
|
||||
mock_metric_exporter.return_value.shutdown = MagicMock()
|
||||
mock_log_exporter.return_value.shutdown = MagicMock()
|
||||
|
||||
handler = OTELHandler(app, enable=True)
|
||||
|
||||
# Call shutdown method
|
||||
handler.shutdown()
|
||||
|
||||
# Verify that shutdown methods are called
|
||||
mock_span_exporter.return_value.shutdown.assert_called_once()
|
||||
mock_metric_exporter.return_value.shutdown.assert_called_once()
|
||||
mock_log_exporter.return_value.shutdown.assert_called_once()
|
||||
|
||||
|
||||
def test_shutdown_tracer_not_initialized(handler_disabled):
|
||||
with pytest.raises(ShutdownError) as exc_info:
|
||||
handler_disabled._shutdown_tracer()
|
||||
assert "Tracer provider not initialized" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_shutdown_metrics_not_initialized(handler_disabled):
|
||||
with pytest.raises(ShutdownError) as exc_info:
|
||||
handler_disabled._shutdown_metrics()
|
||||
assert "Meter provider not initialized" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_shutdown_logging_not_initialized(handler_disabled):
|
||||
with pytest.raises(ShutdownError) as exc_info:
|
||||
handler_disabled._shutdown_logging()
|
||||
assert "Log provider not initialized" in str(exc_info.value)
|
||||
|
||||
|
||||
@patch("arcade.core.telemetry.get_meter_provider")
|
||||
@patch("arcade.core.telemetry.OTLPLogExporter")
|
||||
@patch("arcade.core.telemetry.OTLPMetricExporter")
|
||||
@patch("arcade.core.telemetry.OTLPSpanExporter")
|
||||
def test_get_meter(
|
||||
mock_span_exporter, mock_metric_exporter, mock_log_exporter, mock_get_meter_provider, app
|
||||
):
|
||||
# Mock the methods that may cause network calls
|
||||
mock_span_exporter.return_value.shutdown = MagicMock()
|
||||
mock_metric_exporter.return_value.shutdown = MagicMock()
|
||||
mock_log_exporter.return_value.shutdown = MagicMock()
|
||||
|
||||
handler = OTELHandler(app, enable=True)
|
||||
|
||||
# Call get_meter method
|
||||
handler.get_meter()
|
||||
|
||||
# Verify that get_meter_provider is called
|
||||
mock_get_meter_provider.assert_called_once()
|
||||
|
|
@ -8,6 +8,7 @@ ARG HOST=0.0.0.0
|
|||
# Set environment variables using the build arguments
|
||||
ENV PORT=${PORT}
|
||||
ENV HOST=${HOST}
|
||||
ENV OTEL_ENABLE=false
|
||||
ENV ARCADE_WORK_DIR=/app
|
||||
|
||||
# Install system dependencies
|
||||
|
|
@ -53,5 +54,5 @@ RUN set -e; \
|
|||
# Expose the port
|
||||
EXPOSE $PORT
|
||||
|
||||
# Run the arcade dev command
|
||||
CMD arcade dev --host $HOST --port $PORT
|
||||
# Run the arcade actorup (hidden cli command)
|
||||
CMD arcade actorup --host $HOST --port $PORT
|
||||
|
|
|
|||
Loading…
Reference in a new issue