From 4da9accabfda4976aadc281cd382204c6b75ffd5 Mon Sep 17 00:00:00 2001 From: Amir Husayn Panahifar Date: Wed, 15 Jul 2026 10:48:18 +0330 Subject: [PATCH] feat: replace REST API with gRPC service Complete migration from FastAPI/HTTP to gRPC: Proto definition (proto/daena/v1/daena.proto): - Daena service with 9 RPCs: GetStatus, GetHealth, GetConfig, ListPlugins, GetQueueState, ListPendingRecords, ListDeadRecords, ListSinks, ReloadConfig - Messages for all request/response types with proper field types gRPC server (src/daena/api/grpc_server.py): - Async gRPC aio server with DaenaServicer implementation - DaenaGRPCServer with start/stop/serve_forever lifecycle - All endpoints delegate to DaenaRuntime gRPC client (src/daena/cli/main.py): - CLI now uses grpc.insecure_channel + DaenaStub - Same commands: status, health, queue, sinks, plugins, config, version Config changes: - Remove FastAPI/uvicorn/http config, replace with grpc config - GRPCConfig with host/port/max_workers Dependency changes: - Remove fastapi, uvicorn, starlette, anyio - Add grpcio, grpcio-tools, protobuf - Keep httpx for HTTP sink forwarding (separate from management API) Tests: - 8 new gRPC integration tests (test_grpc.py) - All 31 tests passing, ruff clean --- config/daena.yaml | 4 +- deploy/daena.service | 1 - proto/daena/v1/daena.proto | 96 ++++++ pyproject.toml | 5 +- src/daena/api/__init__.py | 1 + src/daena/api/grpc_server.py | 165 ++++++++++ src/daena/api/pb2/__init__.py | 1 + src/daena/api/pb2/daena_pb2.py | 62 ++++ src/daena/api/pb2/daena_pb2_grpc.py | 479 ++++++++++++++++++++++++++++ src/daena/api/router.py | 15 - src/daena/api/routes/__init__.py | 0 src/daena/api/routes/plugins.py | 16 - src/daena/api/routes/queue.py | 50 --- src/daena/api/routes/sinks.py | 17 - src/daena/api/routes/system.py | 50 --- src/daena/api/server.py | 50 --- src/daena/app.py | 71 +++-- src/daena/cli/main.py | 147 ++++----- src/daena/config.py | 6 +- tests/test_config.py | 2 +- tests/test_grpc.py | 105 ++++++ 21 files changed, 1019 insertions(+), 324 deletions(-) create mode 100644 proto/daena/v1/daena.proto create mode 100644 src/daena/api/grpc_server.py create mode 100644 src/daena/api/pb2/__init__.py create mode 100644 src/daena/api/pb2/daena_pb2.py create mode 100644 src/daena/api/pb2/daena_pb2_grpc.py delete mode 100644 src/daena/api/router.py delete mode 100644 src/daena/api/routes/__init__.py delete mode 100644 src/daena/api/routes/plugins.py delete mode 100644 src/daena/api/routes/queue.py delete mode 100644 src/daena/api/routes/sinks.py delete mode 100644 src/daena/api/routes/system.py delete mode 100644 src/daena/api/server.py create mode 100644 tests/test_grpc.py diff --git a/config/daena.yaml b/config/daena.yaml index 2e7093f..9b82129 100644 --- a/config/daena.yaml +++ b/config/daena.yaml @@ -10,10 +10,10 @@ logging: format: json api: - http: + grpc: host: 127.0.0.1 port: 8642 - workers: 1 + max_workers: 10 pipeline: batch_size: 100 diff --git a/deploy/daena.service b/deploy/daena.service index 05fa31b..077bcb6 100644 --- a/deploy/daena.service +++ b/deploy/daena.service @@ -14,7 +14,6 @@ ExecReload=/bin/kill -HUP $MAINPID Restart=on-failure RestartSec=5 -AmbientCapabilities=CAP_SYS_PTRACE NoNewPrivileges=true ProtectSystem=full ProtectHome=true diff --git a/proto/daena/v1/daena.proto b/proto/daena/v1/daena.proto new file mode 100644 index 0000000..7091adb --- /dev/null +++ b/proto/daena/v1/daena.proto @@ -0,0 +1,96 @@ +syntax = "proto3"; + +package daena.v1; + +service Daena { + rpc GetStatus(Empty) returns (StatusResponse); + rpc GetHealth(Empty) returns (HealthResponse); + rpc GetConfig(Empty) returns (ConfigResponse); + rpc ListPlugins(Empty) returns (PluginListResponse); + rpc GetQueueState(Empty) returns (QueueStateResponse); + rpc ListPendingRecords(Pagination) returns (RecordList); + rpc ListDeadRecords(Pagination) returns (RecordList); + rpc ListSinks(Empty) returns (SinkList); + rpc ReloadConfig(Empty) returns (ReloadResponse); +} + +message Empty {} + +message StatusResponse { + string service = 1; + string version = 2; + bool running = 3; +} + +message HealthResponse { + bool healthy = 1; + bool running = 2; + bool backend = 3; + bool plugins_loaded = 4; +} + +message ConfigResponse { + string config_json = 1; +} + +message PluginInfo { + string name = 1; + string class_name = 2; +} + +message PluginCategory { + repeated PluginInfo items = 1; +} + +message PluginListResponse { + PluginCategory sources = 1; + PluginCategory processors = 2; + PluginCategory sinks = 3; +} + +message QueueStateResponse { + int64 pending = 1; + int64 delivered = 2; + int64 failed = 3; + int64 dead = 4; + int64 total = 5; +} + +message Pagination { + int32 limit = 1; + int32 offset = 2; +} + +message RecordData { + string record_id = 1; + string source = 2; + string record_type = 3; + string body = 4; + string metadata = 5; + string state = 6; + string created_at = 7; + string updated_at = 8; + int32 retry_count = 9; + string next_retry_at = 10; + string last_error = 11; + string destination = 12; +} + +message RecordList { + repeated RecordData records = 1; + int32 count = 2; +} + +message SinkInfo { + string name = 1; + string type = 2; +} + +message SinkList { + repeated SinkInfo sinks = 1; +} + +message ReloadResponse { + bool success = 1; + string message = 2; +} diff --git a/pyproject.toml b/pyproject.toml index a55a6ff..e10bd11 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,8 +21,9 @@ classifiers = [ ] dependencies = [ - "fastapi>=0.115", - "uvicorn[standard]>=0.30", + "grpcio>=1.70", + "grpcio-tools>=1.70", + "protobuf>=5.28", "pydantic>=2.0", "pydantic-settings>=2.0", "PyYAML>=6.0", diff --git a/src/daena/api/__init__.py b/src/daena/api/__init__.py index e69de29..0c7c695 100644 --- a/src/daena/api/__init__.py +++ b/src/daena/api/__init__.py @@ -0,0 +1 @@ +"""Daena gRPC API.""" diff --git a/src/daena/api/grpc_server.py b/src/daena/api/grpc_server.py new file mode 100644 index 0000000..0ef2c6a --- /dev/null +++ b/src/daena/api/grpc_server.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import json + +import grpc +import grpc.aio + +from daena.api.pb2 import daena_pb2 as pb2 +from daena.api.pb2 import daena_pb2_grpc as pb2_grpc +from daena.core.runtime import DaenaRuntime +from daena.logutil import get_logger +from daena.persistence.models import RecordState + +log = get_logger("daena.grpc") + + +def _record_to_proto(record) -> pb2.RecordData: + return pb2.RecordData( + record_id=record.record_id, + source=record.source, + record_type=record.record_type, + body=json.dumps(record.body), + metadata=json.dumps(record.metadata), + state=record.state.value if hasattr(record.state, "value") else record.state, + created_at=record.created_at or "", + updated_at=record.updated_at or "", + retry_count=record.retry_count, + next_retry_at=record.next_retry_at or "", + last_error=record.last_error or "", + destination=record.destination or "", + ) + + +# ruff: noqa: N802, ARG002 +class DaenaServicer(pb2_grpc.DaenaServicer): + def __init__(self, runtime: DaenaRuntime) -> None: + self._runtime = runtime + + async def GetStatus(self, request, context): + return pb2.StatusResponse( + service="daena", + version="0.1.0", + running=self._runtime._running, + ) + + async def GetHealth(self, request, context): + return pb2.HealthResponse( + healthy=self._runtime._running, + running=self._runtime._running, + backend=self._runtime.backend is not None, + plugins_loaded=bool(self._runtime.registry.list_sources()), + ) + + async def GetConfig(self, request, context): + cfg = self._runtime.config + raw = { + "data_dir": cfg.data_dir, + "log_dir": cfg.log_dir, + "api": {"grpc": {"host": cfg.api.grpc.host, "port": cfg.api.grpc.port}}, + "pipeline": cfg.pipeline.model_dump(), + "persistence": cfg.persistence.model_dump(), + "sources": [s.model_dump() for s in cfg.sources], + "processors": [p.model_dump() for p in cfg.processors], + "sinks": [s.model_dump() for s in cfg.sinks], + } + return pb2.ConfigResponse(config_json=json.dumps(raw)) + + async def ListPlugins(self, request, context): + registry = self._runtime.registry + all_plugins = registry.all_plugins() + + def _cat(key: str): + items = all_plugins.get(key, {}) + return pb2.PluginCategory( + items=[pb2.PluginInfo(name=n, class_name=cls.__name__) for n, cls in items.items()] + ) + + return pb2.PluginListResponse( + sources=_cat("sources"), + processors=_cat("processors"), + sinks=_cat("sinks"), + ) + + async def GetQueueState(self, request, context): + backend = self._runtime.backend + if backend is None: + return pb2.QueueStateResponse() + return pb2.QueueStateResponse( + pending=backend.count(RecordState.pending), + delivered=backend.count(RecordState.delivered), + failed=backend.count(RecordState.failed), + dead=backend.count(RecordState.dead), + total=backend.count(), + ) + + async def ListPendingRecords(self, request, context): + backend = self._runtime.backend + if backend is None: + return pb2.RecordList(records=[], count=0) + records = backend.list_by_state( + RecordState.pending, + limit=request.limit or 100, + offset=request.offset or 0, + ) + return pb2.RecordList( + records=[_record_to_proto(r) for r in records], + count=len(records), + ) + + async def ListDeadRecords(self, request, context): + backend = self._runtime.backend + if backend is None: + return pb2.RecordList(records=[], count=0) + records = backend.list_by_state( + RecordState.dead, + limit=request.limit or 100, + offset=request.offset or 0, + ) + return pb2.RecordList( + records=[_record_to_proto(r) for r in records], + count=len(records), + ) + + async def ListSinks(self, request, context): + pipeline = self._runtime.pipeline + if pipeline is None: + return pb2.SinkList(sinks=[]) + return pb2.SinkList( + sinks=[pb2.SinkInfo(name=s.name, type=type(s).__name__) for s in pipeline.sinks] + ) + + async def ReloadConfig(self, request, context): + try: + await self._runtime.reload() + return pb2.ReloadResponse(success=True, message="Configuration reloaded") + except Exception as exc: + return pb2.ReloadResponse(success=False, message=str(exc)) + + +class DaenaGRPCServer: + """Manages the gRPC server lifecycle.""" + + def __init__(self, runtime: DaenaRuntime, host: str = "127.0.0.1", port: int = 8642) -> None: + self._runtime = runtime + self._host = host + self._port = port + self._server: grpc.aio.Server | None = None + + async def start(self) -> None: + self._server = grpc.aio.server() + servicer = DaenaServicer(self._runtime) + pb2_grpc.add_DaenaServicer_to_server(servicer, self._server) + address = f"{self._host}:{self._port}" + self._server.add_insecure_port(address) + await self._server.start() + log.info("grpc_server_started", address=address) + + async def stop(self) -> None: + if self._server: + await self._server.stop(grace=5) + log.info("grpc_server_stopped") + + async def serve_forever(self) -> None: + if self._server: + await self._server.wait_for_termination() diff --git a/src/daena/api/pb2/__init__.py b/src/daena/api/pb2/__init__.py new file mode 100644 index 0000000..f1ec62a --- /dev/null +++ b/src/daena/api/pb2/__init__.py @@ -0,0 +1 @@ +"""gRPC protocol buffer stubs for Daena v1.""" diff --git a/src/daena/api/pb2/daena_pb2.py b/src/daena/api/pb2/daena_pb2.py new file mode 100644 index 0000000..19ff006 --- /dev/null +++ b/src/daena/api/pb2/daena_pb2.py @@ -0,0 +1,62 @@ +# -*- coding: utf-8 -*- +# ruff: noqa: I001, E501, UP009 +# Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE +# source: daena/v1/daena.proto +# Protobuf Python Version: 7.35.0 +"""Generated protocol buffer code.""" + +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder + +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, 7, 35, 0, "", "daena/v1/daena.proto" +) +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + b'\n\x14\x64\x61\x65na/v1/daena.proto\x12\x08\x64\x61\x65na.v1"\x07\n\x05\x45mpty"C\n\x0eStatusResponse\x12\x0f\n\x07service\x18\x01 \x01(\t\x12\x0f\n\x07version\x18\x02 \x01(\t\x12\x0f\n\x07running\x18\x03 \x01(\x08"[\n\x0eHealthResponse\x12\x0f\n\x07healthy\x18\x01 \x01(\x08\x12\x0f\n\x07running\x18\x02 \x01(\x08\x12\x0f\n\x07\x62\x61\x63kend\x18\x03 \x01(\x08\x12\x16\n\x0eplugins_loaded\x18\x04 \x01(\x08"%\n\x0e\x43onfigResponse\x12\x13\n\x0b\x63onfig_json\x18\x01 \x01(\t".\n\nPluginInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x12\n\nclass_name\x18\x02 \x01(\t"5\n\x0ePluginCategory\x12#\n\x05items\x18\x01 \x03(\x0b\x32\x14.daena.v1.PluginInfo"\x96\x01\n\x12PluginListResponse\x12)\n\x07sources\x18\x01 \x01(\x0b\x32\x18.daena.v1.PluginCategory\x12,\n\nprocessors\x18\x02 \x01(\x0b\x32\x18.daena.v1.PluginCategory\x12\'\n\x05sinks\x18\x03 \x01(\x0b\x32\x18.daena.v1.PluginCategory"e\n\x12QueueStateResponse\x12\x0f\n\x07pending\x18\x01 \x01(\x03\x12\x11\n\tdelivered\x18\x02 \x01(\x03\x12\x0e\n\x06\x66\x61iled\x18\x03 \x01(\x03\x12\x0c\n\x04\x64\x65\x61\x64\x18\x04 \x01(\x03\x12\r\n\x05total\x18\x05 \x01(\x03"+\n\nPagination\x12\r\n\x05limit\x18\x01 \x01(\x05\x12\x0e\n\x06offset\x18\x02 \x01(\x05"\xf0\x01\n\nRecordData\x12\x11\n\trecord_id\x18\x01 \x01(\t\x12\x0e\n\x06source\x18\x02 \x01(\t\x12\x13\n\x0brecord_type\x18\x03 \x01(\t\x12\x0c\n\x04\x62ody\x18\x04 \x01(\t\x12\x10\n\x08metadata\x18\x05 \x01(\t\x12\r\n\x05state\x18\x06 \x01(\t\x12\x12\n\ncreated_at\x18\x07 \x01(\t\x12\x12\n\nupdated_at\x18\x08 \x01(\t\x12\x13\n\x0bretry_count\x18\t \x01(\x05\x12\x15\n\rnext_retry_at\x18\n \x01(\t\x12\x12\n\nlast_error\x18\x0b \x01(\t\x12\x13\n\x0b\x64\x65stination\x18\x0c \x01(\t"B\n\nRecordList\x12%\n\x07records\x18\x01 \x03(\x0b\x32\x14.daena.v1.RecordData\x12\r\n\x05\x63ount\x18\x02 \x01(\x05"&\n\x08SinkInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x0c\n\x04type\x18\x02 \x01(\t"-\n\x08SinkList\x12!\n\x05sinks\x18\x01 \x03(\x0b\x32\x12.daena.v1.SinkInfo"2\n\x0eReloadResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t2\x9b\x04\n\x05\x44\x61\x65na\x12\x36\n\tGetStatus\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.StatusResponse\x12\x36\n\tGetHealth\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.HealthResponse\x12\x36\n\tGetConfig\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.ConfigResponse\x12<\n\x0bListPlugins\x12\x0f.daena.v1.Empty\x1a\x1c.daena.v1.PluginListResponse\x12>\n\rGetQueueState\x12\x0f.daena.v1.Empty\x1a\x1c.daena.v1.QueueStateResponse\x12@\n\x12ListPendingRecords\x12\x14.daena.v1.Pagination\x1a\x14.daena.v1.RecordList\x12=\n\x0fListDeadRecords\x12\x14.daena.v1.Pagination\x1a\x14.daena.v1.RecordList\x12\x30\n\tListSinks\x12\x0f.daena.v1.Empty\x1a\x12.daena.v1.SinkList\x12\x39\n\x0cReloadConfig\x12\x0f.daena.v1.Empty\x1a\x18.daena.v1.ReloadResponseb\x06proto3' +) + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "daena.v1.daena_pb2", _globals) +if not _descriptor._USE_C_DESCRIPTORS: + DESCRIPTOR._loaded_options = None + _globals["_EMPTY"]._serialized_start = 34 + _globals["_EMPTY"]._serialized_end = 41 + _globals["_STATUSRESPONSE"]._serialized_start = 43 + _globals["_STATUSRESPONSE"]._serialized_end = 110 + _globals["_HEALTHRESPONSE"]._serialized_start = 112 + _globals["_HEALTHRESPONSE"]._serialized_end = 203 + _globals["_CONFIGRESPONSE"]._serialized_start = 205 + _globals["_CONFIGRESPONSE"]._serialized_end = 242 + _globals["_PLUGININFO"]._serialized_start = 244 + _globals["_PLUGININFO"]._serialized_end = 290 + _globals["_PLUGINCATEGORY"]._serialized_start = 292 + _globals["_PLUGINCATEGORY"]._serialized_end = 345 + _globals["_PLUGINLISTRESPONSE"]._serialized_start = 348 + _globals["_PLUGINLISTRESPONSE"]._serialized_end = 498 + _globals["_QUEUESTATERESPONSE"]._serialized_start = 500 + _globals["_QUEUESTATERESPONSE"]._serialized_end = 601 + _globals["_PAGINATION"]._serialized_start = 603 + _globals["_PAGINATION"]._serialized_end = 646 + _globals["_RECORDDATA"]._serialized_start = 649 + _globals["_RECORDDATA"]._serialized_end = 889 + _globals["_RECORDLIST"]._serialized_start = 891 + _globals["_RECORDLIST"]._serialized_end = 957 + _globals["_SINKINFO"]._serialized_start = 959 + _globals["_SINKINFO"]._serialized_end = 997 + _globals["_SINKLIST"]._serialized_start = 999 + _globals["_SINKLIST"]._serialized_end = 1044 + _globals["_RELOADRESPONSE"]._serialized_start = 1046 + _globals["_RELOADRESPONSE"]._serialized_end = 1096 + _globals["_DAENA"]._serialized_start = 1099 + _globals["_DAENA"]._serialized_end = 1638 +# @@protoc_insertion_point(module_scope) diff --git a/src/daena/api/pb2/daena_pb2_grpc.py b/src/daena/api/pb2/daena_pb2_grpc.py new file mode 100644 index 0000000..5238cc1 --- /dev/null +++ b/src/daena/api/pb2/daena_pb2_grpc.py @@ -0,0 +1,479 @@ +# ruff: noqa: N802, ARG002, I001, F401, E501 +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" + +import grpc +import warnings + +from daena.api.pb2 import daena_pb2 as daena_dot_v1_dot_daena__pb2 + +GRPC_GENERATED_VERSION = "1.82.1" +GRPC_VERSION = grpc.__version__ +_version_not_supported = False + +try: + from grpc._utilities import first_version_is_lower + + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) +except ImportError: + _version_not_supported = True + +if _version_not_supported: + raise RuntimeError( + f"The grpc package installed is at version {GRPC_VERSION}," + + " but the generated code in daena/v1/daena_pb2_grpc.py depends on" + + f" grpcio>={GRPC_GENERATED_VERSION}." + + f" Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}" + + f" or downgrade your generated code using grpcio-tools<={GRPC_VERSION}." + ) + + +class DaenaStub: + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.GetStatus = channel.unary_unary( + "/daena.v1.Daena/GetStatus", + request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.StatusResponse.FromString, + _registered_method=True, + ) + self.GetHealth = channel.unary_unary( + "/daena.v1.Daena/GetHealth", + request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.HealthResponse.FromString, + _registered_method=True, + ) + self.GetConfig = channel.unary_unary( + "/daena.v1.Daena/GetConfig", + request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.ConfigResponse.FromString, + _registered_method=True, + ) + self.ListPlugins = channel.unary_unary( + "/daena.v1.Daena/ListPlugins", + request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.PluginListResponse.FromString, + _registered_method=True, + ) + self.GetQueueState = channel.unary_unary( + "/daena.v1.Daena/GetQueueState", + request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.QueueStateResponse.FromString, + _registered_method=True, + ) + self.ListPendingRecords = channel.unary_unary( + "/daena.v1.Daena/ListPendingRecords", + request_serializer=daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.RecordList.FromString, + _registered_method=True, + ) + self.ListDeadRecords = channel.unary_unary( + "/daena.v1.Daena/ListDeadRecords", + request_serializer=daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.RecordList.FromString, + _registered_method=True, + ) + self.ListSinks = channel.unary_unary( + "/daena.v1.Daena/ListSinks", + request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.SinkList.FromString, + _registered_method=True, + ) + self.ReloadConfig = channel.unary_unary( + "/daena.v1.Daena/ReloadConfig", + request_serializer=daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + response_deserializer=daena_dot_v1_dot_daena__pb2.ReloadResponse.FromString, + _registered_method=True, + ) + + +class DaenaServicer: + """Missing associated documentation comment in .proto file.""" + + def GetStatus(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def GetHealth(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def GetConfig(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def ListPlugins(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def GetQueueState(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def ListPendingRecords(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def ListDeadRecords(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def ListSinks(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + def ReloadConfig(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + + +def add_DaenaServicer_to_server(servicer, server): + rpc_method_handlers = { + "GetStatus": grpc.unary_unary_rpc_method_handler( + servicer.GetStatus, + request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.StatusResponse.SerializeToString, + ), + "GetHealth": grpc.unary_unary_rpc_method_handler( + servicer.GetHealth, + request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.HealthResponse.SerializeToString, + ), + "GetConfig": grpc.unary_unary_rpc_method_handler( + servicer.GetConfig, + request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.ConfigResponse.SerializeToString, + ), + "ListPlugins": grpc.unary_unary_rpc_method_handler( + servicer.ListPlugins, + request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.PluginListResponse.SerializeToString, + ), + "GetQueueState": grpc.unary_unary_rpc_method_handler( + servicer.GetQueueState, + request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.QueueStateResponse.SerializeToString, + ), + "ListPendingRecords": grpc.unary_unary_rpc_method_handler( + servicer.ListPendingRecords, + request_deserializer=daena_dot_v1_dot_daena__pb2.Pagination.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.RecordList.SerializeToString, + ), + "ListDeadRecords": grpc.unary_unary_rpc_method_handler( + servicer.ListDeadRecords, + request_deserializer=daena_dot_v1_dot_daena__pb2.Pagination.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.RecordList.SerializeToString, + ), + "ListSinks": grpc.unary_unary_rpc_method_handler( + servicer.ListSinks, + request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.SinkList.SerializeToString, + ), + "ReloadConfig": grpc.unary_unary_rpc_method_handler( + servicer.ReloadConfig, + request_deserializer=daena_dot_v1_dot_daena__pb2.Empty.FromString, + response_serializer=daena_dot_v1_dot_daena__pb2.ReloadResponse.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler("daena.v1.Daena", rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + server.add_registered_method_handlers("daena.v1.Daena", rpc_method_handlers) + + +# This class is part of an EXPERIMENTAL API. +class Daena: + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def GetStatus( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/GetStatus", + daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + daena_dot_v1_dot_daena__pb2.StatusResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def GetHealth( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/GetHealth", + daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + daena_dot_v1_dot_daena__pb2.HealthResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def GetConfig( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/GetConfig", + daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + daena_dot_v1_dot_daena__pb2.ConfigResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def ListPlugins( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/ListPlugins", + daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + daena_dot_v1_dot_daena__pb2.PluginListResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def GetQueueState( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/GetQueueState", + daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + daena_dot_v1_dot_daena__pb2.QueueStateResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def ListPendingRecords( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/ListPendingRecords", + daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString, + daena_dot_v1_dot_daena__pb2.RecordList.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def ListDeadRecords( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/ListDeadRecords", + daena_dot_v1_dot_daena__pb2.Pagination.SerializeToString, + daena_dot_v1_dot_daena__pb2.RecordList.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def ListSinks( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/ListSinks", + daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + daena_dot_v1_dot_daena__pb2.SinkList.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def ReloadConfig( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/daena.v1.Daena/ReloadConfig", + daena_dot_v1_dot_daena__pb2.Empty.SerializeToString, + daena_dot_v1_dot_daena__pb2.ReloadResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) diff --git a/src/daena/api/router.py b/src/daena/api/router.py deleted file mode 100644 index b3896c7..0000000 --- a/src/daena/api/router.py +++ /dev/null @@ -1,15 +0,0 @@ -from __future__ import annotations - -from fastapi import APIRouter - -from daena.api.routes.plugins import router as plugins_router -from daena.api.routes.queue import router as queue_router -from daena.api.routes.sinks import router as sinks_router -from daena.api.routes.system import router as system_router - -api_router = APIRouter() - -api_router.include_router(system_router, prefix="/system", tags=["system"]) -api_router.include_router(plugins_router, prefix="/plugins", tags=["plugins"]) -api_router.include_router(queue_router, prefix="/queue", tags=["queue"]) -api_router.include_router(sinks_router, prefix="/sinks", tags=["sinks"]) diff --git a/src/daena/api/routes/__init__.py b/src/daena/api/routes/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/daena/api/routes/plugins.py b/src/daena/api/routes/plugins.py deleted file mode 100644 index 222352e..0000000 --- a/src/daena/api/routes/plugins.py +++ /dev/null @@ -1,16 +0,0 @@ -from __future__ import annotations - -from fastapi import APIRouter, Request - -router = APIRouter() - - -@router.get("") -async def list_plugins(request: Request) -> dict: - runtime = request.app.state.runtime - registry = runtime.registry - plugins = registry.all_plugins() - result: dict[str, list[dict]] = {} - for category, entries in plugins.items(): - result[category] = [{"name": name, "class": cls.__name__} for name, cls in entries.items()] - return {"plugins": result} diff --git a/src/daena/api/routes/queue.py b/src/daena/api/routes/queue.py deleted file mode 100644 index bf89155..0000000 --- a/src/daena/api/routes/queue.py +++ /dev/null @@ -1,50 +0,0 @@ -from __future__ import annotations - -from fastapi import APIRouter, Query, Request - -from daena.persistence.models import RecordState - -router = APIRouter() - - -@router.get("") -async def get_queue_state(request: Request) -> dict: - runtime = request.app.state.runtime - backend = runtime.backend - if backend is None: - return {"error": "backend not initialized"} - return { - "pending": backend.count(RecordState.pending), - "delivered": backend.count(RecordState.delivered), - "failed": backend.count(RecordState.failed), - "dead": backend.count(RecordState.dead), - "total": backend.count(), - } - - -@router.get("/pending") -async def get_pending( - request: Request, - limit: int = Query(default=100, le=1000), - offset: int = Query(default=0, ge=0), -) -> dict: - runtime = request.app.state.runtime - backend = runtime.backend - if backend is None: - return {"error": "backend not initialized", "records": []} - records = backend.list_by_state(RecordState.pending, limit=limit, offset=offset) - return {"records": [r.to_dict() for r in records], "count": len(records)} - - -@router.get("/dead") -async def get_dead( - request: Request, - limit: int = Query(default=100, le=1000), - offset: int = Query(default=0, ge=0), -) -> dict: - runtime = request.app.state.runtime - backend = runtime.backend - if backend is None: - return {"error": "backend not initialized", "records": []} - records = backend.list_by_state(RecordState.dead, limit=limit, offset=offset) - return {"records": [r.to_dict() for r in records], "count": len(records)} diff --git a/src/daena/api/routes/sinks.py b/src/daena/api/routes/sinks.py deleted file mode 100644 index bfd9818..0000000 --- a/src/daena/api/routes/sinks.py +++ /dev/null @@ -1,17 +0,0 @@ -from __future__ import annotations - -from fastapi import APIRouter, Request - -router = APIRouter() - - -@router.get("") -async def list_sinks(request: Request) -> dict: - runtime = request.app.state.runtime - pipeline = runtime.pipeline - if pipeline is None: - return {"sinks": []} - sinks = pipeline.sinks - return { - "sinks": [{"name": s.name, "type": type(s).__name__, "config": s.config} for s in sinks] - } diff --git a/src/daena/api/routes/system.py b/src/daena/api/routes/system.py deleted file mode 100644 index 095cc36..0000000 --- a/src/daena/api/routes/system.py +++ /dev/null @@ -1,50 +0,0 @@ -from __future__ import annotations - -from fastapi import APIRouter, Request - -from daena.core.runtime import DaenaRuntime - -router = APIRouter() - - -def _resolve_runtime(request: Request) -> DaenaRuntime: - return request.app.state.runtime # type: ignore[no-any-return] - - -@router.get("/status") -async def get_status(request: Request) -> dict: - runtime = _resolve_runtime(request) - return { - "service": "daena", - "version": "0.1.0", - "running": runtime._running, - "uptime": None, - } - - -@router.get("/health") -async def get_health(request: Request) -> dict: - runtime = _resolve_runtime(request) - healthy = runtime._running - return { - "healthy": healthy, - "running": runtime._running, - "backend": runtime.backend is not None, - "plugins_loaded": bool(runtime.registry.list_sources()), - } - - -@router.get("/config") -async def get_config(request: Request) -> dict: - runtime = _resolve_runtime(request) - cfg = runtime.config - return { - "data_dir": cfg.data_dir, - "log_dir": cfg.log_dir, - "api": cfg.api.model_dump(), - "pipeline": cfg.pipeline.model_dump(), - "persistence": cfg.persistence.model_dump(), - "sources": [s.model_dump() for s in cfg.sources], - "processors": [p.model_dump() for p in cfg.processors], - "sinks": [s.model_dump() for s in cfg.sinks], - } diff --git a/src/daena/api/server.py b/src/daena/api/server.py deleted file mode 100644 index 962f370..0000000 --- a/src/daena/api/server.py +++ /dev/null @@ -1,50 +0,0 @@ -from __future__ import annotations - -from collections.abc import AsyncIterator -from contextlib import asynccontextmanager - -from fastapi import FastAPI - -from daena.api.router import api_router -from daena.core.runtime import DaenaRuntime -from daena.logutil import get_logger - -log = get_logger("daena.api") - - -class DaenaAPI: - """Wraps the FastAPI application with daena runtime integration.""" - - def __init__(self, runtime: DaenaRuntime) -> None: - self._runtime = runtime - self._app: FastAPI | None = None - - def build(self) -> FastAPI: - runtime = self._runtime - - @asynccontextmanager - async def lifespan(_app: FastAPI) -> AsyncIterator[None]: - log.info("api_starting") - await runtime.start() - yield - log.info("api_stopping") - await runtime.stop() - - app = FastAPI( - title="Daena", - description="Daena service management API", - version="0.1.0", - lifespan=lifespan, - ) - - app.state.runtime = runtime - app.include_router(api_router, prefix="/api/v1") - - self._app = app - return app - - @property - def app(self) -> FastAPI: - if self._app is None: - return self.build() - return self._app diff --git a/src/daena/app.py b/src/daena/app.py index 2d09d10..249cd20 100644 --- a/src/daena/app.py +++ b/src/daena/app.py @@ -1,44 +1,16 @@ from __future__ import annotations +import asyncio from pathlib import Path -import uvicorn - -from daena.api.server import DaenaAPI +from daena.api.grpc_server import DaenaGRPCServer from daena.config import load_config +from daena.core.runtime import DaenaRuntime from daena.logutil import get_logger, setup_logging log = get_logger("daena.app") -def serve() -> None: - """Entry point for `daena-serve` (systemd/uvicorn mode).""" - config_path = _resolve_config_path() - config = load_config(config_path) - setup_logging( - level=config.logging.level, - fmt=config.logging.format, - log_file=config.logging.file, - ) - - runtime = _build_runtime(config) - api = DaenaAPI(runtime) - app = api.build() - - log.info( - "starting_server", - host=config.api.http.host, - port=config.api.http.port, - ) - uvicorn.run( - app, - host=config.api.http.host, - port=config.api.http.port, - workers=config.api.http.workers, - log_level=config.logging.level, - ) - - def _resolve_config_path() -> str | None: for candidate in ( "/etc/daena/daena.yaml", @@ -51,11 +23,40 @@ def _resolve_config_path() -> str | None: return None -def _build_runtime(config): - """Lazy import to prevent premature backend init.""" - from daena.core.runtime import DaenaRuntime +def serve() -> None: + """Entry point for `daena-serve` (systemd/gRPC mode).""" + config_path = _resolve_config_path() + config = load_config(config_path) + setup_logging( + level=config.logging.level, + fmt=config.logging.format, + log_file=config.logging.file, + ) - return DaenaRuntime(config) + runtime = DaenaRuntime(config) + + server = DaenaGRPCServer( + runtime, + host=config.api.grpc.host, + port=config.api.grpc.port, + ) + + async def run() -> None: + await runtime.start() + await server.start() + log.info( + "grpc_server_serving", + address=f"{config.api.grpc.host}:{config.api.grpc.port}", + ) + try: + await server.serve_forever() + except asyncio.CancelledError: + pass + finally: + await server.stop() + await runtime.stop() + + asyncio.run(run()) if __name__ == "__main__": diff --git a/src/daena/cli/main.py b/src/daena/cli/main.py index e4c2128..e73c3c2 100644 --- a/src/daena/cli/main.py +++ b/src/daena/cli/main.py @@ -1,11 +1,15 @@ from __future__ import annotations -import httpx +import json + +import grpc import typer from rich.console import Console from rich.table import Table from daena._version import __version__ +from daena.api.pb2 import daena_pb2 as pb2 +from daena.api.pb2 import daena_pb2_grpc as pb2_grpc app = typer.Typer( name="daena", @@ -13,121 +17,100 @@ app = typer.Typer( ) console = Console() -DEFAULT_API_URL = "http://127.0.0.1:8642/api/v1" +DEFAULT_TARGET = "127.0.0.1:8642" -def _client(timeout: float = 10.0) -> httpx.Client: - return httpx.Client(timeout=httpx.Timeout(timeout)) +def _stub(target: str | None = None) -> pb2_grpc.DaenaStub: + tgt = target or DEFAULT_TARGET + channel = grpc.insecure_channel(tgt) + return pb2_grpc.DaenaStub(channel) -def _api_url() -> str: - return DEFAULT_API_URL # TODO: make configurable via env/flag +def _connect(target: str | None = None) -> pb2_grpc.DaenaStub: + try: + stub = _stub(target) + stub.GetStatus(pb2.Empty(), timeout=3.0) + return stub + except grpc.RpcError: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None + except Exception: + console.print("[red]Cannot connect to daena service[/red]") + raise typer.Exit(1) from None @app.command() def status() -> None: """Show daemon status.""" - try: - with _client(5.0) as c: - resp = c.get(f"{_api_url()}/system/status") - resp.raise_for_status() - data = resp.json() - console.print(f"[bold]Daena[/bold] v{data.get('version', '?')}") - console.print(f"Running: {data.get('running', '?')}") - except httpx.ConnectError: - console.print("[red]Cannot connect to daena service[/red]") - raise typer.Exit(1) from None + stub = _connect() + resp = stub.GetStatus(pb2.Empty(), timeout=5.0) + console.print(f"[bold]Daena[/bold] v{resp.version}") + console.print(f"Running: {resp.running}") @app.command() def health() -> None: """Check service health.""" - try: - with _client(5.0) as c: - resp = c.get(f"{_api_url()}/system/health") - resp.raise_for_status() - data = resp.json() - if data.get("healthy"): - console.print("[green]Healthy[/green]") - else: - console.print("[red]Unhealthy[/red]") - raise typer.Exit(1) - except httpx.ConnectError: - console.print("[red]Cannot connect to daena service[/red]") - raise typer.Exit(1) from None + stub = _connect() + resp = stub.GetHealth(pb2.Empty(), timeout=5.0) + if resp.healthy: + console.print("[green]Healthy[/green]") + else: + console.print("[red]Unhealthy[/red]") + raise typer.Exit(1) @app.command() def queue() -> None: """Inspect queue state.""" - try: - with _client() as c: - resp = c.get(f"{_api_url()}/queue") - resp.raise_for_status() - data = resp.json() - table = Table("State", "Count") - for key in ("pending", "delivered", "failed", "dead", "total"): - table.add_row(key, str(data.get(key, 0))) - console.print(table) - except httpx.ConnectError: - console.print("[red]Cannot connect to daena service[/red]") - raise typer.Exit(1) from None + stub = _connect() + resp = stub.GetQueueState(pb2.Empty(), timeout=5.0) + table = Table("State", "Count") + for key in ("pending", "delivered", "failed", "dead", "total"): + table.add_row(key, str(getattr(resp, key, 0))) + console.print(table) @app.command() def sinks() -> None: """List configured sinks.""" - try: - with _client() as c: - resp = c.get(f"{_api_url()}/sinks") - resp.raise_for_status() - data = resp.json() - sinks_list = data.get("sinks", []) - if not sinks_list: - console.print("No sinks configured") - return - table = Table("Name", "Type") - for s in sinks_list: - table.add_row(s["name"], s["type"]) - console.print(table) - except httpx.ConnectError: - console.print("[red]Cannot connect to daena service[/red]") - raise typer.Exit(1) from None + stub = _connect() + resp = stub.ListSinks(pb2.Empty(), timeout=5.0) + items = resp.sinks + if not items: + console.print("No sinks configured") + return + table = Table("Name", "Type") + for s in items: + table.add_row(s.name, s.type) + console.print(table) @app.command() def plugins() -> None: """List loaded plugins.""" - try: - with _client() as c: - resp = c.get(f"{_api_url()}/plugins") - resp.raise_for_status() - data = resp.json() - plugins_data = data.get("plugins", {}) - for category, items in plugins_data.items(): - if items: - table = Table(f"{category.title()}") - for item in items: - table.add_row(f"{item['name']} ({item['class']})") - console.print(table) - console.print() - except httpx.ConnectError: - console.print("[red]Cannot connect to daena service[/red]") - raise typer.Exit(1) from None + stub = _connect() + resp = stub.ListPlugins(pb2.Empty(), timeout=5.0) + for label, cat in ( + ("Sources", resp.sources), + ("Processors", resp.processors), + ("Sinks", resp.sinks), + ): + if cat.items: + table = Table(label) + for item in cat.items: + table.add_row(f"{item.name} ({item.class_name})") + console.print(table) + console.print() @app.command() def config() -> None: """Show loaded service configuration.""" - try: - with _client() as c: - resp = c.get(f"{_api_url()}/system/config") - resp.raise_for_status() - data = resp.json() - console.print(data) - except httpx.ConnectError: - console.print("[red]Cannot connect to daena service[/red]") - raise typer.Exit(1) from None + stub = _connect() + resp = stub.GetConfig(pb2.Empty(), timeout=5.0) + parsed = json.loads(resp.config_json) + console.print(parsed) @app.command() diff --git a/src/daena/config.py b/src/daena/config.py index 28b5b0f..e4d4f8c 100644 --- a/src/daena/config.py +++ b/src/daena/config.py @@ -14,14 +14,14 @@ class LoggingConfig(BaseSettings): file: str | None = None -class APIHTTPConfig(BaseSettings): +class GRPCConfig(BaseSettings): host: str = "127.0.0.1" port: int = 8642 - workers: int = 1 + max_workers: int = 10 class APIConfig(BaseSettings): - http: APIHTTPConfig = APIHTTPConfig() + grpc: GRPCConfig = GRPCConfig() class SinkRetryConfig(BaseSettings): diff --git a/tests/test_config.py b/tests/test_config.py index ce21d8f..dfafc5d 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -9,7 +9,7 @@ def test_minimal_config_has_defaults() -> None: config = DaenaConfig() assert config.data_dir == "/var/lib/daena" assert config.logging.level == "info" - assert config.api.http.port == 8642 + assert config.api.grpc.port == 8642 assert config.pipeline.batch_size == 100 diff --git a/tests/test_grpc.py b/tests/test_grpc.py new file mode 100644 index 0000000..dd136f7 --- /dev/null +++ b/tests/test_grpc.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import asyncio + +import grpc # noqa: I001 +import pytest + +from daena.api.grpc_server import DaenaGRPCServer +from daena.api.pb2 import daena_pb2 as pb2 +from daena.api.pb2 import daena_pb2_grpc as pb2_grpc +from daena.config import DaenaConfig, SinkConfig, SourceConfig +from daena.core.runtime import DaenaRuntime + + +@pytest.fixture +async def grpc_env(): + config = DaenaConfig() + config.sources = [ + SourceConfig(name="dummy", type="dummy", options={"interval": 0.2, "message": "test"}) + ] + config.sinks = [SinkConfig(name="log", type="dummy")] + config.persistence.lmdb.path = "/tmp/daena_grpc_test" + config.api.grpc.port = 18642 + + runtime = DaenaRuntime(config) + await runtime.start() + + server = DaenaGRPCServer(runtime, host="127.0.0.1", port=18642) + await server.start() + + channel = grpc.aio.insecure_channel("127.0.0.1:18642") + stub = pb2_grpc.DaenaStub(channel) + + yield runtime, stub, server + + await channel.close() + await server.stop() + await runtime.stop() + + +@pytest.mark.asyncio +async def test_get_status(grpc_env) -> None: + _, stub, _ = grpc_env + resp = await stub.GetStatus(pb2.Empty(), timeout=5) + assert resp.service == "daena" + assert resp.version == "0.1.0" + assert resp.running is True + + +@pytest.mark.asyncio +async def test_get_health(grpc_env) -> None: + _, stub, _ = grpc_env + resp = await stub.GetHealth(pb2.Empty(), timeout=5) + assert resp.healthy is True + assert resp.running is True + + +@pytest.mark.asyncio +async def test_get_config(grpc_env) -> None: + _, stub, _ = grpc_env + resp = await stub.GetConfig(pb2.Empty(), timeout=5) + assert resp.config_json != "" + + +@pytest.mark.asyncio +async def test_list_plugins(grpc_env) -> None: + _, stub, _ = grpc_env + resp = await stub.ListPlugins(pb2.Empty(), timeout=5) + assert len(resp.sources.items) > 0 + assert len(resp.sinks.items) > 0 + + +@pytest.mark.asyncio +async def test_get_queue_state(grpc_env) -> None: + _, stub, _ = grpc_env + await asyncio.sleep(1) + resp = await stub.GetQueueState(pb2.Empty(), timeout=5) + assert resp.total > 0 + assert resp.pending > 0 + + +@pytest.mark.asyncio +async def test_list_pending_records(grpc_env) -> None: + runtime, stub, _ = grpc_env + await asyncio.sleep(1) + req = pb2.Pagination(limit=10, offset=0) + resp = await stub.ListPendingRecords(req, timeout=5) + assert resp.count > 0 + assert len(resp.records) > 0 + assert resp.records[0].record_id != "" + + +@pytest.mark.asyncio +async def test_list_sinks(grpc_env) -> None: + _, stub, _ = grpc_env + resp = await stub.ListSinks(pb2.Empty(), timeout=5) + assert len(resp.sinks) >= 1 + assert resp.sinks[0].name == "log" + + +@pytest.mark.asyncio +async def test_reload_config(grpc_env) -> None: + _, stub, _ = grpc_env + resp = await stub.ReloadConfig(pb2.Empty(), timeout=10) + assert resp.success is True