From a4238cc6cc52e57eb2eeba4266509793a06cedb7 Mon Sep 17 00:00:00 2001 From: lecodev-26 Date: Sun, 1 Nov 2026 10:28:06 +0100 Subject: [PATCH 1/2] feat(diagnostics): expose write protocol health and stats --- CHANGELOG.md | 4 + README.md | 24 +++++ src/arc_client/async_client.py | 12 +++ src/arc_client/client.py | 10 +++ src/arc_client/management/__init__.py | 7 +- .../management/async_write_diagnostics.py | 48 ++++++++++ .../management/write_diagnostics.py | 48 ++++++++++ tests/unit/http/test_write_diagnostics.py | 87 +++++++++++++++++++ 8 files changed, 239 insertions(+), 1 deletion(-) create mode 100644 src/arc_client/management/async_write_diagnostics.py create mode 100644 src/arc_client/management/write_diagnostics.py create mode 100644 tests/unit/http/test_write_diagnostics.py diff --git a/CHANGELOG.md b/CHANGELOG.md index bebc788..ba4863f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,10 @@ plus bug fixes and the first real HTTP-layer test coverage. - Admin `logs`, `memstats`, `duckdb_memory` and `free_os_memory(confirm=True)`. - `auth.cache_stats()` and `auth.invalidate_cache()`. +- **Write-protocol diagnostics** (`client.write_diagnostics`, sync + async): + inspect Line Protocol health and statistics, and retrieve MessagePack + protocol specifications and decoder/buffer statistics. Responses are raw + server-defined JSON; administrative flush remains out of scope. - **Edge sync administration** (`client.edge_sync`): - Hub: spoke registry and air-gap bundle import. - Spoke: status, ledger, sync, requeue/dismiss, and bundle export/ack. diff --git a/README.md b/README.md index eca88fe..ec14353 100644 --- a/README.md +++ b/README.md @@ -375,6 +375,30 @@ async def main(): asyncio.run(main()) ``` +## Write Protocol Diagnostics + +Inspect the health and runtime counters of the Line Protocol and MessagePack +writers. These read-only endpoints require an authenticated token. + +```python +# Synchronous +health = client.write_diagnostics.line_protocol_health() +line_stats = client.write_diagnostics.line_protocol_stats() +msgpack_spec = client.write_diagnostics.msgpack_spec() +msgpack_stats = client.write_diagnostics.msgpack_stats() + +# Asynchronous equivalents +async with AsyncArcClient(host="localhost", token="your-token") as async_client: + health = await async_client.write_diagnostics.line_protocol_health() + line_stats = await async_client.write_diagnostics.line_protocol_stats() + msgpack_spec = await async_client.write_diagnostics.msgpack_spec() + msgpack_stats = await async_client.write_diagnostics.msgpack_stats() +``` + +The methods return the server's JSON response unchanged. They are available on +Arc 26.09.3 and later. The administrative Line Protocol flush endpoint +is intentionally not included in this read-only diagnostics interface. + ## Management Operations ### Retention Policies diff --git a/src/arc_client/async_client.py b/src/arc_client/async_client.py index cc8c1d9..8fe175b 100644 --- a/src/arc_client/async_client.py +++ b/src/arc_client/async_client.py @@ -101,6 +101,7 @@ def __init__( self._edge_sync: Any = None self._diagnostics: Any = None self._reconciliation: Any = None + self._write_diagnostics: Any = None def _get_http(self) -> AsyncHTTPClient: """Get or create the HTTP client.""" @@ -259,6 +260,17 @@ def backup(self) -> Any: self._backup = AsyncBackupClient(self._get_http(), self._config) return self._backup + @property + def write_diagnostics(self) -> Any: + """Get async read-only Line Protocol and MessagePack diagnostics.""" + if self._write_diagnostics is None: + from arc_client.management.async_write_diagnostics import AsyncWriteDiagnosticsClient + + self._write_diagnostics = AsyncWriteDiagnosticsClient( + self._get_http(), self._config + ) + return self._write_diagnostics + @property def mqtt(self) -> Any: """Get the async MQTT ingestion management client (enterprise).""" diff --git a/src/arc_client/client.py b/src/arc_client/client.py index 5a5a1ae..16f7d0f 100644 --- a/src/arc_client/client.py +++ b/src/arc_client/client.py @@ -101,6 +101,7 @@ def __init__( self._edge_sync: Any = None self._diagnostics: Any = None self._reconciliation: Any = None + self._write_diagnostics: Any = None def _get_http(self) -> SyncHTTPClient: """Get or create the HTTP client.""" @@ -257,6 +258,15 @@ def backup(self) -> Any: self._backup = BackupClient(self._get_http(), self._config) return self._backup + @property + def write_diagnostics(self) -> Any: + """Get read-only Line Protocol and MessagePack diagnostics.""" + if self._write_diagnostics is None: + from arc_client.management.write_diagnostics import WriteDiagnosticsClient + + self._write_diagnostics = WriteDiagnosticsClient(self._get_http(), self._config) + return self._write_diagnostics + @property def mqtt(self) -> Any: """Get the MQTT ingestion management client (enterprise).""" diff --git a/src/arc_client/management/__init__.py b/src/arc_client/management/__init__.py index f155a87..6816a00 100644 --- a/src/arc_client/management/__init__.py +++ b/src/arc_client/management/__init__.py @@ -2,7 +2,8 @@ Retention, continuous queries, delete, databases, and the enterprise surface (RBAC, governance, audit, query management, cluster, compaction, tiering, -backup, MQTT) and operations (reconciliation, diagnostics, edge sync). +backup, MQTT, write diagnostics) and operations (reconciliation, diagnostics, +edge sync). """ from arc_client.management.async_audit import AsyncAuditClient @@ -21,6 +22,7 @@ from arc_client.management.async_reconciliation import AsyncReconciliationClient from arc_client.management.async_retention import AsyncRetentionClient from arc_client.management.async_tiering import AsyncTieringClient +from arc_client.management.async_write_diagnostics import AsyncWriteDiagnosticsClient from arc_client.management.audit import AuditClient from arc_client.management.backup import BackupClient from arc_client.management.cluster import ClusterClient @@ -37,6 +39,7 @@ from arc_client.management.reconciliation import ReconciliationClient from arc_client.management.retention import RetentionClient from arc_client.management.tiering import TieringClient +from arc_client.management.write_diagnostics import WriteDiagnosticsClient __all__ = [ # Core management @@ -46,6 +49,8 @@ "AsyncDeleteClient", "RetentionClient", "AsyncRetentionClient", + "WriteDiagnosticsClient", + "AsyncWriteDiagnosticsClient", "DatabaseClient", "AsyncDatabaseClient", # Enterprise diff --git a/src/arc_client/management/async_write_diagnostics.py b/src/arc_client/management/async_write_diagnostics.py new file mode 100644 index 0000000..ba2870a --- /dev/null +++ b/src/arc_client/management/async_write_diagnostics.py @@ -0,0 +1,48 @@ +"""Asynchronous diagnostics for Arc write protocols. + +Wraps the read-only Line Protocol health/statistics and MessagePack +specification/statistics endpoints. All endpoints require authentication. +Responses are returned as raw JSON values because Arc does not promise a +stable schema for these operator diagnostics. +""" + +from __future__ import annotations + +from typing import Any + +from arc_client.config import ClientConfig +from arc_client.exceptions import ArcError +from arc_client.http.async_http import AsyncHTTPClient + + +class AsyncWriteDiagnosticsClient: + """Inspect writer health and statistics; return raw, server-defined JSON.""" + + def __init__(self, http: AsyncHTTPClient, config: ClientConfig) -> None: + self._http = http + self._config = config + + async def _json(self, path: str) -> Any: + try: + response = await self._http.get(path) + return response.json() + except ArcError: + raise + except Exception as e: + raise ArcError(f"Write diagnostics request failed (GET {path}): {e}") from e + + async def line_protocol_health(self) -> Any: + """Return raw JSON health status for the Line Protocol writer.""" + return await self._json("/api/v1/write/line-protocol/health") + + async def line_protocol_stats(self) -> Any: + """Return raw JSON request, record, byte, and error counters for Line Protocol.""" + return await self._json("/api/v1/write/line-protocol/stats") + + async def msgpack_spec(self) -> Any: + """Return the raw JSON MessagePack write-protocol specification.""" + return await self._json("/api/v1/write/msgpack/spec") + + async def msgpack_stats(self) -> Any: + """Return raw JSON MessagePack decoder and Arrow buffer statistics.""" + return await self._json("/api/v1/write/msgpack/stats") diff --git a/src/arc_client/management/write_diagnostics.py b/src/arc_client/management/write_diagnostics.py new file mode 100644 index 0000000..6cd8102 --- /dev/null +++ b/src/arc_client/management/write_diagnostics.py @@ -0,0 +1,48 @@ +"""Synchronous diagnostics for Arc write protocols. + +Wraps the read-only Line Protocol health/statistics and MessagePack +specification/statistics endpoints. All endpoints require authentication. +Responses are returned as raw JSON values because Arc does not promise a +stable schema for these operator diagnostics. +""" + +from __future__ import annotations + +from typing import Any + +from arc_client.config import ClientConfig +from arc_client.exceptions import ArcError +from arc_client.http.sync_http import SyncHTTPClient + + +class WriteDiagnosticsClient: + """Inspect writer health and statistics; return raw, server-defined JSON.""" + + def __init__(self, http: SyncHTTPClient, config: ClientConfig) -> None: + self._http = http + self._config = config + + def _json(self, path: str) -> Any: + try: + response = self._http.get(path) + return response.json() + except ArcError: + raise + except Exception as e: + raise ArcError(f"Write diagnostics request failed (GET {path}): {e}") from e + + def line_protocol_health(self) -> Any: + """Return raw JSON health status for the Line Protocol writer.""" + return self._json("/api/v1/write/line-protocol/health") + + def line_protocol_stats(self) -> Any: + """Return raw JSON request, record, byte, and error counters for Line Protocol.""" + return self._json("/api/v1/write/line-protocol/stats") + + def msgpack_spec(self) -> Any: + """Return the raw JSON MessagePack write-protocol specification.""" + return self._json("/api/v1/write/msgpack/spec") + + def msgpack_stats(self) -> Any: + """Return raw JSON MessagePack decoder and Arrow buffer statistics.""" + return self._json("/api/v1/write/msgpack/stats") diff --git a/tests/unit/http/test_write_diagnostics.py b/tests/unit/http/test_write_diagnostics.py new file mode 100644 index 0000000..988e8a6 --- /dev/null +++ b/tests/unit/http/test_write_diagnostics.py @@ -0,0 +1,87 @@ +"""HTTP contract tests for Line Protocol and MessagePack diagnostics.""" + +from __future__ import annotations + +import pytest +from pytest_httpx import HTTPXMock + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ArcPermissionError, ArcServerError + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_write_protocol_diagnostics(httpx_mock: HTTPXMock, async_mode: bool) -> None: + expected = [ + {"status": "healthy", "service": "line_protocol_writer"}, + { + "status": "success", + "stats": { + "total_requests": 12, + "total_records": 34, + "total_bytes": 1024, + "total_bytes_gzipped": 512, + "total_errors": 1, + }, + }, + {"version": "2.0", "protocol": "MessagePack"}, + {"decoder": {"records": 5}, "buffer": {"buffers": 2}}, + ] + paths = [ + "/api/v1/write/line-protocol/health", + "/api/v1/write/line-protocol/stats", + "/api/v1/write/msgpack/spec", + "/api/v1/write/msgpack/stats", + ] + for payload in expected: + httpx_mock.add_response(json=payload) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + diagnostics = client.write_diagnostics + results = [ + await diagnostics.line_protocol_health(), + await diagnostics.line_protocol_stats(), + await diagnostics.msgpack_spec(), + await diagnostics.msgpack_stats(), + ] + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + diagnostics = client.write_diagnostics + results = [ + diagnostics.line_protocol_health(), + diagnostics.line_protocol_stats(), + diagnostics.msgpack_spec(), + diagnostics.msgpack_stats(), + ] + + assert results == expected + requests = httpx_mock.get_requests() + assert [(request.method, request.url.path) for request in requests] == [ + ("GET", path) for path in paths + ] + assert all(request.headers["Authorization"] == "Bearer t" for request in requests) + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_write_diagnostics_preserve_permission_and_server_errors( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + httpx_mock.add_response(status_code=403, json={"error": "Authentication required"}) + httpx_mock.add_response(status_code=503, json={"error": "Writer unavailable"}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcPermissionError) as permission_error: + await client.write_diagnostics.line_protocol_health() + with pytest.raises(ArcServerError) as error: + await client.write_diagnostics.msgpack_stats() + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcPermissionError) as permission_error: + client.write_diagnostics.line_protocol_health() + with pytest.raises(ArcServerError) as error: + client.write_diagnostics.msgpack_stats() + + assert permission_error.value.response_body == {"error": "Authentication required"} + assert error.value.status_code == 503 + assert error.value.response_body == {"error": "Writer unavailable"} From 242e516c846cc1228ed53823deb2c6724d6b474c Mon Sep 17 00:00:00 2001 From: lecodev-26 Date: Mon, 2 Nov 2026 22:25:04 +0100 Subject: [PATCH 2/2] docs(diagnostics): clarify stale MessagePack auth metadata --- src/arc_client/async_client.py | 4 +++- src/arc_client/management/async_write_diagnostics.py | 8 +++++++- src/arc_client/management/write_diagnostics.py | 8 +++++++- 3 files changed, 17 insertions(+), 3 deletions(-) diff --git a/src/arc_client/async_client.py b/src/arc_client/async_client.py index 8fe175b..73ca247 100644 --- a/src/arc_client/async_client.py +++ b/src/arc_client/async_client.py @@ -264,7 +264,9 @@ def backup(self) -> Any: def write_diagnostics(self) -> Any: """Get async read-only Line Protocol and MessagePack diagnostics.""" if self._write_diagnostics is None: - from arc_client.management.async_write_diagnostics import AsyncWriteDiagnosticsClient + from arc_client.management.async_write_diagnostics import ( + AsyncWriteDiagnosticsClient, + ) self._write_diagnostics = AsyncWriteDiagnosticsClient( self._get_http(), self._config diff --git a/src/arc_client/management/async_write_diagnostics.py b/src/arc_client/management/async_write_diagnostics.py index ba2870a..69e61f3 100644 --- a/src/arc_client/management/async_write_diagnostics.py +++ b/src/arc_client/management/async_write_diagnostics.py @@ -40,7 +40,13 @@ async def line_protocol_stats(self) -> Any: return await self._json("/api/v1/write/line-protocol/stats") async def msgpack_spec(self) -> Any: - """Return the raw JSON MessagePack write-protocol specification.""" + """Return the raw JSON MessagePack write-protocol specification. + + Arc's current response contains stale authentication metadata: it says + authentication is not implemented and lists only ``x-api-key``. The + endpoint does require authentication, so do not rely on that section + until the server-side specification is corrected. + """ return await self._json("/api/v1/write/msgpack/spec") async def msgpack_stats(self) -> Any: diff --git a/src/arc_client/management/write_diagnostics.py b/src/arc_client/management/write_diagnostics.py index 6cd8102..009f936 100644 --- a/src/arc_client/management/write_diagnostics.py +++ b/src/arc_client/management/write_diagnostics.py @@ -40,7 +40,13 @@ def line_protocol_stats(self) -> Any: return self._json("/api/v1/write/line-protocol/stats") def msgpack_spec(self) -> Any: - """Return the raw JSON MessagePack write-protocol specification.""" + """Return the raw JSON MessagePack write-protocol specification. + + Arc's current response contains stale authentication metadata: it says + authentication is not implemented and lists only ``x-api-key``. The + endpoint does require authentication, so do not rely on that section + until the server-side specification is corrected. + """ return self._json("/api/v1/write/msgpack/spec") def msgpack_stats(self) -> Any: