diff --git a/CHANGELOG.md b/CHANGELOG.md index bebc788..0277f72 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -79,6 +79,16 @@ plus bug fixes and the first real HTTP-layer test coverage. `ARC_CLIENT_SEND_INSTALLATION_ID=0` or `send_installation_id=False`; pin the ID in containers with `ARC_CLIENT_INSTALLATION_ID`. See README, "Privacy". +- **Scheduler management** (`client.schedulers`, sync + async): inspect CQ and retention + scheduler status, reload active continuous-query schedules, and manually trigger + retention. The status route accepts authenticated tokens; control routes preserve + Arc administrator and enterprise-license gates. Retention triggers require + `confirm=True` because they may permanently delete data. Supported by Arc 26.09.3 and later. + On Arc 27.01.1 and later, a control refused because the license is missing, expired + or of a tier without the scheduler raises `ArcLicenseError` (HTTP 403); a scheduler + off in config is `ArcServerError` with `status_code == 503`, and a retention trigger + on a node that is not the primary writer, or while a cycle runs, is `ArcServerError` + with `status_code == 409`. - Arrow completion metadata in the existing `pyarrow.Table` schema for sync and async queries: policy cap and execution time are retained; failed, malformed, conflicting or unobservable announced completion metadata is rejected. diff --git a/README.md b/README.md index eca88fe..9a7f483 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ with ArcClient(host="localhost", token="your-token") as client: - **Databases**: List, create, delete databases and list measurements - **Management**: Retention policies, continuous queries, delete operations - **Enterprise**: RBAC, query governance, audit logs, query management, cluster, - compaction, storage tiering, backup/restore, and MQTT ingestion + compaction, scheduler controls, storage tiering, backup/restore, and MQTT ingestion - **Authentication**: Token management (create, rotate, revoke) ## Data Ingestion @@ -401,6 +401,28 @@ print(f"Would delete {result.deleted_count} rows") result = client.retention.execute(policy.id, dry_run=False, confirm=True) ``` +### Scheduler inspection and controls + +Scheduler status is available to authenticated tokens. Reloading continuous-query +schedules and manually triggering retention require an administrator token and +the corresponding enterprise scheduler feature. + +```python +status = client.schedulers.status() +client.schedulers.reload_continuous_queries() + +# Runs all active retention policies immediately and may permanently delete data. +client.schedulers.trigger_retention(confirm=True) +``` + +The retention operation refuses to send a request unless `confirm=True` is +supplied. The async client exposes the same methods: + +```python +async with AsyncArcClient(host="localhost", token="your-token") as client: + status = await client.schedulers.status() +``` + ### Continuous Queries Aggregate and downsample data automatically: @@ -513,6 +535,11 @@ client.compaction.trigger(database="metrics", measurement="cpu") client.compaction.cycles(limit=5) client.compaction.too_wide() +# Scheduler inspection and explicit operator controls +client.schedulers.status() # requires an authenticated token +client.schedulers.reload_continuous_queries() # administrator + licensed feature +# Retention may permanently delete data; explicit confirmation is required. +client.schedulers.trigger_retention(confirm=True) client.tiering.set_policy("metrics", hot_only=True, hot_max_age_days=7) client.backup.create(databases=["metrics"]) # scoped backup diff --git a/src/arc_client/async_client.py b/src/arc_client/async_client.py index cc8c1d9..f7b5713 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._schedulers: Any = None def _get_http(self) -> AsyncHTTPClient: """Get or create the HTTP client.""" @@ -259,6 +260,15 @@ def backup(self) -> Any: self._backup = AsyncBackupClient(self._get_http(), self._config) return self._backup + @property + def schedulers(self) -> Any: + """Get the async scheduler inspection and control client.""" + if self._schedulers is None: + from arc_client.management.async_schedulers import AsyncSchedulerClient + + self._schedulers = AsyncSchedulerClient(self._get_http(), self._config) + return self._schedulers + @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..ef9d551 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._schedulers: 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 schedulers(self) -> Any: + """Get the scheduler inspection and control client.""" + if self._schedulers is None: + from arc_client.management.schedulers import SchedulerClient + + self._schedulers = SchedulerClient(self._get_http(), self._config) + return self._schedulers + @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..1e60757 100644 --- a/src/arc_client/management/__init__.py +++ b/src/arc_client/management/__init__.py @@ -20,6 +20,7 @@ from arc_client.management.async_rbac import AsyncRBACClient from arc_client.management.async_reconciliation import AsyncReconciliationClient from arc_client.management.async_retention import AsyncRetentionClient +from arc_client.management.async_schedulers import AsyncSchedulerClient from arc_client.management.async_tiering import AsyncTieringClient from arc_client.management.audit import AuditClient from arc_client.management.backup import BackupClient @@ -36,6 +37,7 @@ from arc_client.management.rbac import RBACClient from arc_client.management.reconciliation import ReconciliationClient from arc_client.management.retention import RetentionClient +from arc_client.management.schedulers import SchedulerClient from arc_client.management.tiering import TieringClient __all__ = [ @@ -46,6 +48,8 @@ "AsyncDeleteClient", "RetentionClient", "AsyncRetentionClient", + "SchedulerClient", + "AsyncSchedulerClient", "DatabaseClient", "AsyncDatabaseClient", # Enterprise diff --git a/src/arc_client/management/async_schedulers.py b/src/arc_client/management/async_schedulers.py new file mode 100644 index 0000000..8025b83 --- /dev/null +++ b/src/arc_client/management/async_schedulers.py @@ -0,0 +1,104 @@ +"""Asynchronous scheduler inspection and control for Arc. + +Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; +control operations require an administrator token and the corresponding +enterprise scheduler feature. Since Arc 27.01.1 a missing, expired or +insufficient license is HTTP 403, raised as ``ArcLicenseError``; HTTP 503 +means the scheduler is off in config or failed to start, and ``status()`` +gives the reason. Earlier Arc versions answer 503 in both cases. +""" + +from __future__ import annotations + +from typing import Any + +from arc_client.config import ClientConfig +from arc_client.exceptions import ( + ArcError, + ArcLicenseError, + ArcPermissionError, + ArcValidationError, +) +from arc_client.http.async_http import AsyncHTTPClient +from arc_client.management.delete import DELETE_TIMEOUT_FLOOR + + +class AsyncSchedulerClient: + """Inspect scheduler status and perform explicit scheduler controls. + + Methods return raw JSON values because scheduler status and counters are + server-defined operator diagnostics, not a stable typed response schema. + """ + + def __init__(self, http: AsyncHTTPClient, config: ClientConfig) -> None: + self._http = http + self._config = config + + async def _json(self, method: str, path: str, **kwargs: Any) -> Any: + try: + response = await getattr(self._http, method)(path, **kwargs) + return response.json() + except ArcPermissionError as e: + # Arc refuses a scheduler control with 403 when the license is + # missing, expired or of a tier without the scheduler. Only the + # "Enterprise license required" form matches the generic 403 + # license detection; the tier and refused-reload forms name the + # license without that phrase. + message = str(e) + if "license" in message.lower(): + detail = message.split("Permission denied: ", 1)[-1] + raise ArcLicenseError(f"Enterprise license required: {detail}") from e + raise + except ArcError: + raise + except Exception as e: + raise ArcError(f"Scheduler request failed ({method.upper()} {path}): {e}") from e + + async def status(self) -> Any: + """Return the status of the CQ and retention schedulers. + + Available with any authenticated token. Returns raw JSON with each + scheduler's enabled state and, when unavailable, its reason. + """ + return await self._json("get", "/api/v1/schedulers/") + + async def reload_continuous_queries(self) -> Any: + """Reload active CQ schedules (administrator and licensed feature required). + + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. + """ + return await self._json("post", "/api/v1/schedulers/cq/reload") + + async def trigger_retention(self, confirm: bool = False) -> Any: + """Run all active retention policies immediately (administrator only). + + This can delete data according to the configured retention policies. + Pass ``confirm=True`` to explicitly authorize the immediate run. The + request allows at least 90 seconds because Arc runs every active policy + synchronously. In a cluster, send it to the primary writer: another node + refuses it, as Arc does while a retention cycle is already running. + Since Arc 27.01.1 both refusals are HTTP 409, raised as + ``ArcServerError`` with ``status_code == 409``; earlier versions + answer HTTP 500. + + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. + + Raises: + ArcValidationError: If ``confirm`` is not True. + """ + if not confirm: + raise ArcValidationError( + "Refusing to trigger retention without confirm=True. " + "This may permanently delete data." + ) + return await self._json( + "post", + "/api/v1/schedulers/retention/trigger", + timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR), + ) diff --git a/src/arc_client/management/schedulers.py b/src/arc_client/management/schedulers.py new file mode 100644 index 0000000..4b0d7e7 --- /dev/null +++ b/src/arc_client/management/schedulers.py @@ -0,0 +1,104 @@ +"""Synchronous scheduler inspection and control for Arc. + +Wraps ``/api/v1/schedulers``. Status is available to authenticated tokens; +control operations require an administrator token and the corresponding +enterprise scheduler feature. Since Arc 27.01.1 a missing, expired or +insufficient license is HTTP 403, raised as ``ArcLicenseError``; HTTP 503 +means the scheduler is off in config or failed to start, and ``status()`` +gives the reason. Earlier Arc versions answer 503 in both cases. +""" + +from __future__ import annotations + +from typing import Any + +from arc_client.config import ClientConfig +from arc_client.exceptions import ( + ArcError, + ArcLicenseError, + ArcPermissionError, + ArcValidationError, +) +from arc_client.http.sync_http import SyncHTTPClient +from arc_client.management.delete import DELETE_TIMEOUT_FLOOR + + +class SchedulerClient: + """Inspect scheduler status and perform explicit scheduler controls. + + Methods return raw JSON values because scheduler status and counters are + server-defined operator diagnostics, not a stable typed response schema. + """ + + def __init__(self, http: SyncHTTPClient, config: ClientConfig) -> None: + self._http = http + self._config = config + + def _json(self, method: str, path: str, **kwargs: Any) -> Any: + try: + response = getattr(self._http, method)(path, **kwargs) + return response.json() + except ArcPermissionError as e: + # Arc refuses a scheduler control with 403 when the license is + # missing, expired or of a tier without the scheduler. Only the + # "Enterprise license required" form matches the generic 403 + # license detection; the tier and refused-reload forms name the + # license without that phrase. + message = str(e) + if "license" in message.lower(): + detail = message.split("Permission denied: ", 1)[-1] + raise ArcLicenseError(f"Enterprise license required: {detail}") from e + raise + except ArcError: + raise + except Exception as e: + raise ArcError(f"Scheduler request failed ({method.upper()} {path}): {e}") from e + + def status(self) -> Any: + """Return the status of the CQ and retention schedulers. + + Available with any authenticated token. Returns raw JSON with each + scheduler's enabled state and, when unavailable, its reason. + """ + return self._json("get", "/api/v1/schedulers/") + + def reload_continuous_queries(self) -> Any: + """Reload active CQ schedules (administrator and licensed feature required). + + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. + """ + return self._json("post", "/api/v1/schedulers/cq/reload") + + def trigger_retention(self, confirm: bool = False) -> Any: + """Run all active retention policies immediately (administrator only). + + This can delete data according to the configured retention policies. + Pass ``confirm=True`` to explicitly authorize the immediate run. The + request allows at least 90 seconds because Arc runs every active policy + synchronously. In a cluster, send it to the primary writer: another node + refuses it, as Arc does while a retention cycle is already running. + Since Arc 27.01.1 both refusals are HTTP 409, raised as + ``ArcServerError`` with ``status_code == 409``; earlier versions + answer HTTP 500. + + Returns the raw JSON response from Arc. A license that is missing, + expired or does not include the scheduler raises ``ArcLicenseError`` + (HTTP 403). HTTP 503 means the scheduler is off in config or failed to + start; call ``status()`` for the reason. + + Raises: + ArcValidationError: If ``confirm`` is not True. + """ + if not confirm: + raise ArcValidationError( + "Refusing to trigger retention without confirm=True. " + "This may permanently delete data." + ) + return self._json( + "post", + "/api/v1/schedulers/retention/trigger", + timeout=max(self._config.timeout, DELETE_TIMEOUT_FLOOR), + ) diff --git a/tests/unit/http/test_schedulers.py b/tests/unit/http/test_schedulers.py new file mode 100644 index 0000000..c1356a5 --- /dev/null +++ b/tests/unit/http/test_schedulers.py @@ -0,0 +1,180 @@ +"""HTTP contract tests for scheduler inspection and explicit controls.""" + +from __future__ import annotations + +from typing import Any + +import pytest +from pytest_httpx import HTTPXMock + +from arc_client import ArcClient, AsyncArcClient +from arc_client.exceptions import ( + ArcLicenseError, + ArcPermissionError, + ArcServerError, + ArcValidationError, +) + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_scheduler_status(httpx_mock: HTTPXMock, async_mode: bool) -> None: + expected = { + "cq_scheduler": {"enabled": True, "job_count": 2}, + "retention_scheduler": {"enabled": False, "reason": "not configured"}, + } + httpx_mock.add_response(json=expected) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + result = await client.schedulers.status() + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + result = client.schedulers.status() + + assert result == expected + request = httpx_mock.get_requests()[0] + assert request.method == "GET" + assert request.url.path == "/api/v1/schedulers/" + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_scheduler_controls(httpx_mock: HTTPXMock, async_mode: bool) -> None: + httpx_mock.add_response(json={"message": "CQ scheduler reloaded", "job_count": 3}) + httpx_mock.add_response(json={"message": "Retention triggered successfully"}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + reload_result = await client.schedulers.reload_continuous_queries() + retention_result = await client.schedulers.trigger_retention(confirm=True) + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + reload_result = client.schedulers.reload_continuous_queries() + retention_result = client.schedulers.trigger_retention(confirm=True) + + assert reload_result["job_count"] == 3 + assert retention_result["message"] == "Retention triggered successfully" + requests = httpx_mock.get_requests() + assert [(request.method, request.url.path) for request in requests] == [ + ("POST", "/api/v1/schedulers/cq/reload"), + ("POST", "/api/v1/schedulers/retention/trigger"), + ] + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_retention_trigger_requires_explicit_confirmation( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcValidationError, match="confirm=True"): + await client.schedulers.trigger_retention() + else: + with ( + ArcClient(host="h", token="t", max_retries=0) as client, + pytest.raises(ArcValidationError, match="confirm=True"), + ): + client.schedulers.trigger_retention() + + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("async_mode", [False, True]) +async def test_scheduler_controls_preserve_permission_and_server_errors( + httpx_mock: HTTPXMock, async_mode: bool +) -> None: + httpx_mock.add_response(status_code=403, json={"error": "Admin permission required"}) + httpx_mock.add_response(status_code=503, json={"error": "Retention scheduler is not running"}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcPermissionError): + await client.schedulers.reload_continuous_queries() + with pytest.raises(ArcServerError) as error: + await client.schedulers.trigger_retention(confirm=True) + else: + with ArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcPermissionError): + client.schedulers.reload_continuous_queries() + with pytest.raises(ArcServerError) as error: + client.schedulers.trigger_retention(confirm=True) + + assert error.value.status_code == 503 + + +# The refusals Arc 27.01.1 answers on the control routes (arc#1433, arc#1445). +_CQ = "/api/v1/schedulers/cq/reload" +_RETENTION = "/api/v1/schedulers/retention/trigger" + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize( + ("path", "message"), + [ + (_CQ, "CQ scheduler is not running: Enterprise license required"), + (_CQ, "CQ scheduler is not running: License tier does not include CQ scheduler feature"), + (_CQ, "CQ scheduler reload refused: cq: the license does not include the CQ scheduler"), + ( + _RETENTION, + "Retention trigger refused: retention: the license does not include the " + "retention scheduler", + ), + ], +) +async def test_scheduler_license_refusals_raise_license_error( + httpx_mock: HTTPXMock, async_mode: bool, path: str, message: str +) -> None: + httpx_mock.add_response(status_code=403, json={"error": message}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcLicenseError) as error: + await _control(client.schedulers, path) + else: + with ( + ArcClient(host="h", token="t", max_retries=0) as client, + pytest.raises(ArcLicenseError) as error, + ): + _control(client.schedulers, path) + + assert message in str(error.value) + assert error.value.response_body == {"error": message} + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.parametrize( + ("path", "status", "message"), + [ + ( + _CQ, + 503, + "CQ scheduler is not running: CQ feature not enabled " + "(continuous_query.enabled=false), or the scheduler failed to start " + "(see the startup log)", + ), + (_RETENTION, 409, "Retention trigger refused: retention: node role is not primary writer"), + (_RETENTION, 409, "Retention trigger refused: retention: a cycle is already running"), + ], +) +async def test_scheduler_not_running_and_refusals_are_server_errors( + httpx_mock: HTTPXMock, async_mode: bool, path: str, status: int, message: str +) -> None: + httpx_mock.add_response(status_code=status, json={"error": message}) + + if async_mode: + async with AsyncArcClient(host="h", token="t", max_retries=0) as client: + with pytest.raises(ArcServerError) as error: + await _control(client.schedulers, path) + else: + with ( + ArcClient(host="h", token="t", max_retries=0) as client, + pytest.raises(ArcServerError) as error, + ): + _control(client.schedulers, path) + + assert error.value.status_code == status + + +def _control(schedulers: Any, path: str) -> Any: + if path == _CQ: + return schedulers.reload_continuous_queries() + return schedulers.trigger_retention(confirm=True)