SlosClient¶
Client for the Kibana SLOs (Service Level Objectives) API.
SLOs let you set measurable targets (availability, latency, …) for your services based on Elasticsearch data and track error budgets against those targets. The SLO APIs require an appropriate license (Platinum or trial).
All endpoints are space-scoped: the official API paths are rooted at
/s/{spaceId}/api/observability/slos. Every method therefore accepts
space_id (None targets the default space) and validate_spaces to
override space-existence validation per call.
- class kibana._sync.client.slos.SlosClient(client, default_space_id=None, validate_spaces=True)[source]¶
Bases:
NamespaceClientClient for the Kibana SLOs (Service Level Objectives) API.
SLOs let you set measurable targets (availability, latency, …) for your services based on Elasticsearch data and track error budgets against those targets. The SLO APIs require an appropriate license (Platinum or trial) and Kibana 8.12+ / 9.x.
All endpoints are space-scoped: the official API paths are rooted at
/s/{spaceId}/api/observability/slos. Every method therefore acceptsspace_id(Nonetargets the default space) andvalidate_spacesto override space-existence validation per call.Example
>>> from kibana import Kibana >>> client = Kibana("http://localhost:5601", api_key="...") >>> >>> # Create an SLO based on a custom KQL indicator >>> slo = client.slos.create( ... name="my-service availability", ... description="99% of requests are good over 30 days", ... indicator={ ... "type": "sli.kql.custom", ... "params": { ... "index": "my-service-logs", ... "good": "status: ok", ... "total": "", ... "timestampField": "@timestamp", ... }, ... }, ... time_window={"duration": "30d", "type": "rolling"}, ... budgeting_method="occurrences", ... objective={"target": 0.99}, ... ) >>> slo_id = slo.body["id"] >>> >>> # Fetch it back, then clean up >>> client.slos.get(slo_id=slo_id) >>> client.slos.delete(slo_id=slo_id)
Creating SLOs
Create an SLO with the
create()method:from kibana import Kibana client = Kibana("http://localhost:5601", api_key="your_api_key") # Create an SLO based on a custom KQL indicator slo = client.slos.create( name="my-service availability", description="99% of requests are good over 30 days", indicator={ "type": "sli.kql.custom", "params": { "index": "my-service-logs", "good": "status: ok", "total": "", "timestampField": "@timestamp", }, }, time_window={"duration": "30d", "type": "rolling"}, budgeting_method="occurrences", objective={"target": 0.99}, ) slo_id = slo.body["id"]
Finding and Retrieving SLOs
# Get a single SLO (with its computed summary) slo = client.slos.get(slo_id=slo_id) # Find SLOs with a KQL query results = client.slos.find(kql_query="slo.name:my-service*") for slo in results.body["results"]: print(slo["name"], slo["summary"]["status"]) # List SLO definitions definitions = client.slos.find_definitions(search="my-service*")
SLO Lifecycle
# Disable and re-enable an SLO client.slos.disable(slo_id=slo_id) client.slos.enable(slo_id=slo_id) # Reset an SLO (recompute rollup data from scratch) client.slos.reset(slo_id=slo_id) # Delete an SLO client.slos.delete(slo_id=slo_id)
Bulk Operations
# Bulk delete SLOs (asynchronous server-side task) task = client.slos.bulk_delete(slo_ids=[slo_id, "another-slo-id"]) # Check the status of the bulk delete task status = client.slos.bulk_delete_status(task_id=task.body["taskId"])
- find(*, kql_query=None, size=None, search_after=None, page=None, per_page=None, sort_by=None, sort_direction=None, hide_stale=None, space_id=None, validate_spaces=None)[source]¶
Get a paginated list of SLOs with their summaries.
GET /api/observability/slos- Parameters:
kql_query (str | None) – A valid KQL query to filter SLOs with, e.g.
'slo.name:latency* and slo.tags : "prod"'.size (int | None) – The page size to use for cursor-based pagination (must be used together with
search_after). Default 1.search_after (list[str] | None) – The cursor to use for fetching the results from, when using a cursor-based pagination.
page (int | None) – The page number to return (default 1). Offset-based pagination; mutually exclusive with
size/search_after.per_page (int | None) – The number of SLOs to return per page (default 25, maximum 5000).
sort_by (str | None) – Sort by field. One of
"sli_value","status","error_budget_consumed","error_budget_remaining"(default"status").sort_direction (str | None) – Sort order,
"asc"(default) or"desc".hide_stale (bool | None) – Hide stale SLOs from the list as defined by the stale SLO threshold in SLO settings.
space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with
page,perPage,totalandresults(a list of SLOs with theirsummary).- Raises:
BadRequestError – If a query parameter is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = client.slos.find(kql_query="slo.name:my-service*") >>> for slo in resp.body["results"]: ... print(slo["name"], slo["summary"]["status"])
- create(*, name, description, indicator, time_window, budgeting_method, objective, id=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶
Create an SLO.
POST /api/observability/slos- Parameters:
name (str) – A name for the SLO.
description (str) – A description for the SLO.
indicator (dict[str, Any]) – The indicator (SLI) definition, e.g.
{"type": "sli.kql.custom", "params": {...}}. Supported types:sli.apm.transactionDuration,sli.apm.transactionErrorRate,sli.kql.custom,sli.metric.custom,sli.histogram.custom,sli.metric.timeslice.time_window (dict[str, Any]) – The SLO time window, e.g.
{"duration": "30d", "type": "rolling"}or{"duration": "1M", "type": "calendarAligned"}.budgeting_method (str) –
"occurrences"or"timeslices".objective (dict[str, Any]) – The objective, e.g.
{"target": 0.99}; timeslices budgeting also usestimesliceTargetandtimesliceWindow.id (str | None) – An optional unique identifier for the SLO (8-36 characters); autogenerated when omitted.
settings (dict[str, Any] | None) – Optional settings such as
syncDelay,frequency,syncFieldandpreventInitialBackfill.group_by (str | list[str] | None) – Optional field or list of fields to generate one SLO per distinct value (e.g.
"service.name").artifacts (dict[str, Any] | None) – Optional links to related assets, e.g.
{"dashboards": [{"id": "..."}]}.space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse containing the
idof the created SLO.- Raises:
BadRequestError – If the SLO definition is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
ConflictError – If an SLO with the same id already exists (409).
- Return type:
Example
>>> resp = client.slos.create( ... name="availability", ... description="99% good over 7 days", ... indicator={ ... "type": "sli.kql.custom", ... "params": { ... "index": "logs-*", ... "good": "status: ok", ... "total": "", ... "timestampField": "@timestamp", ... }, ... }, ... time_window={"duration": "7d", "type": "rolling"}, ... budgeting_method="occurrences", ... objective={"target": 0.99}, ... ) >>> print(resp.body["id"])
- get(*, slo_id, instance_id=None, space_id=None, validate_spaces=None)[source]¶
Get an SLO by its identifier.
GET /api/observability/slos/{sloId}- Parameters:
slo_id (str) – The SLO identifier.
instance_id (str | None) – The SLO instance identifier when retrieving a specific instance of a grouped SLO.
space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with the SLO definition and its
summary.- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> slo = client.slos.get(slo_id="my-slo-id") >>> print(slo.body["name"], slo.body["enabled"])
- update(*, slo_id, name=None, description=None, indicator=None, time_window=None, budgeting_method=None, objective=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶
Update an SLO.
PUT /api/observability/slos/{sloId}Performs a partial update: only the provided fields are changed. Updating fields that are part of the rollup transform (indicator, time window, budgeting method, objective, group_by or settings) triggers a re-computation of the SLO data.
- Parameters:
slo_id (str) – The SLO identifier.
name (str | None) – A new name for the SLO.
description (str | None) – A new description for the SLO.
indicator (dict[str, Any] | None) – The indicator (SLI) definition.
budgeting_method (str | None) –
"occurrences"or"timeslices".objective (dict[str, Any] | None) – The objective, e.g.
{"target": 0.99}.settings (dict[str, Any] | None) – Settings such as
syncDelay,frequency,syncFieldandpreventInitialBackfill.group_by (str | list[str] | None) – Field or list of fields to generate one SLO per distinct value.
artifacts (dict[str, Any] | None) – Links to related assets, e.g.
{"dashboards": [{"id": "..."}]}.space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with the updated SLO definition.
- Raises:
BadRequestError – If the update payload is invalid (400).
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = client.slos.update( ... slo_id="my-slo-id", ... description="Updated description", ... tags=["prod"], ... )
- delete(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Delete an SLO and its associated summary and rollup data.
DELETE /api/observability/slos/{sloId}- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> client.slos.delete(slo_id="my-slo-id")
- enable(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Enable an SLO, resuming data rollup and summary computation.
POST /api/observability/slos/{sloId}/enable- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> client.slos.enable(slo_id="my-slo-id")
- disable(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Disable an SLO, stopping data rollup and summary computation.
POST /api/observability/slos/{sloId}/disable- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> client.slos.disable(slo_id="my-slo-id")
- reset(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Reset an SLO, deleting its data and recomputing it from scratch.
POST /api/observability/slos/{sloId}/_reset- Parameters:
- Returns:
ObjectApiResponse with the reset SLO definition.
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = client.slos.reset(slo_id="my-slo-id") >>> print(resp.body["version"])
- delete_instances(*, instances, space_id=None, validate_spaces=None)[source]¶
Batch delete rollup and summary data for SLO instances.
POST /api/observability/slos/_delete_instancesDeletes the rollup and summary data for the given list of SLO id / instance id pairs. Useful for removing stale data of instances of a grouped SLO that no longer receive updates.
- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
BadRequestError – If the payload is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> client.slos.delete_instances( ... instances=[ ... {"sloId": "my-slo-id", "instanceId": "my-service"}, ... ] ... )
- bulk_delete(*, slo_ids, space_id=None, validate_spaces=None)[source]¶
Bulk delete SLO definitions and their summary and rollup data.
POST /api/observability/slos/_bulk_deleteThe deletion occurs asynchronously: the response contains a
taskIdthat can be polled withbulk_delete_status()to retrieve the operation outcome.- Parameters:
- Returns:
ObjectApiResponse containing the
taskIdof the bulk deletion.- Raises:
BadRequestError – If the payload is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = client.slos.bulk_delete(slo_ids=["slo-1", "slo-2"]) >>> status = client.slos.bulk_delete_status( ... task_id=resp.body["taskId"] ... )
- bulk_delete_status(*, task_id, space_id=None, validate_spaces=None)[source]¶
Retrieve the status of an asynchronous bulk deletion.
GET /api/observability/slos/_bulk_delete/{taskId}- Parameters:
task_id (str) – The task id returned by
bulk_delete().space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with
isDone, per-SLOresultsand an optionalerror.- Raises:
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> status = client.slos.bulk_delete_status(task_id="...") >>> if status.body["isDone"]: ... for result in status.body["results"]: ... print(result["id"], result["success"])
- bulk_purge_rollup(*, slo_ids, purge_policy, space_id=None, validate_spaces=None)[source]¶
Batch delete rollup and summary data according to a purge policy.
POST /api/observability/slos/_bulk_purge_rollupThe deletion occurs asynchronously (Elasticsearch delete-by-query) and the response contains an Elasticsearch task id.
Note
Kibana 9.4.3 expects
purgeTypevalues"fixed_age"/"fixed_time"(snake_case), although the official OpenAPI spec documents"fixed-age"/"fixed-time".- Parameters:
slo_ids (list[str]) – A list of SLO ids whose rollup data should be purged.
purge_policy (dict[str, Any]) – Policy dictating which SLI documents to purge, either
{"purgeType": "fixed_age", "age": "7d"}or{"purgeType": "fixed_time", "timestamp": "2024-01-01T00:00:00.000Z"}.space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse containing the Elasticsearch
taskIdof the purge operation.- Raises:
BadRequestError – If the purge policy is invalid or would purge data inside an SLO time window (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = client.slos.bulk_purge_rollup( ... slo_ids=["my-slo-id"], ... purge_policy={"purgeType": "fixed_age", "age": "30d"}, ... ) >>> print(resp.body["taskId"])
- find_definitions(*, include_outdated_only=None, include_health=None, tags=None, search=None, page=None, per_page=None, space_id=None, validate_spaces=None)[source]¶
Get the SLO definitions (without computed summaries).
GET /api/observability/slos/_definitionsNote
The official OpenAPI spec documents this operation under
/internal/observability/slos/_definitions, but Kibana 9.4.3 serves it on the public/api/observability/slos/_definitionspath (the internal path returns 404); this client uses the public path.- Parameters:
include_outdated_only (bool | None) – Indicates if the API returns only outdated SLOs or all SLOs.
include_health (bool | None) – Include the health of the SLO transforms in the response.
tags (str | None) – Filter SLOs by tag (comma-separated list of tags).
search (str | None) – Filter SLOs by name, e.g.
"my-service*".page (int | None) – The page number to return.
per_page (int | None) – The number of SLOs to return per page (default 100, maximum 1000).
space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with
page,perPage,totalandresults(a list of SLO definitions).- Raises:
BadRequestError – If a query parameter is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = client.slos.find_definitions(search="my-service*") >>> print(resp.body["total"])
- __init__(client, default_space_id=None, validate_spaces=True)¶
Initialize NamespaceClient with optional space support.
AsyncSlosClient¶
Asynchronous version of the SlosClient for use with async/await syntax.
- class kibana._async.client.slos.AsyncSlosClient(client, default_space_id=None, validate_spaces=True)[source]¶
Bases:
AsyncNamespaceClientAsync client for the Kibana SLOs (Service Level Objectives) API.
SLOs let you set measurable targets (availability, latency, …) for your services based on Elasticsearch data and track error budgets against those targets. The SLO APIs require an appropriate license (Platinum or trial) and Kibana 8.12+ / 9.x.
All endpoints are space-scoped: the official API paths are rooted at
/s/{spaceId}/api/observability/slos. Every method therefore acceptsspace_id(Nonetargets the default space) andvalidate_spacesto override space-existence validation per call.Example
>>> from kibana import AsyncKibana >>> client = AsyncKibana("http://localhost:5601", api_key="...") >>> >>> # Create an SLO based on a custom KQL indicator >>> slo = await client.slos.create( ... name="my-service availability", ... description="99% of requests are good over 30 days", ... indicator={ ... "type": "sli.kql.custom", ... "params": { ... "index": "my-service-logs", ... "good": "status: ok", ... "total": "", ... "timestampField": "@timestamp", ... }, ... }, ... time_window={"duration": "30d", "type": "rolling"}, ... budgeting_method="occurrences", ... objective={"target": 0.99}, ... ) >>> slo_id = slo.body["id"] >>> >>> # Fetch it back, then clean up >>> await client.slos.get(slo_id=slo_id) >>> await client.slos.delete(slo_id=slo_id)
Usage
The AsyncSlosClient provides the same methods as SlosClient but all methods are async and must be awaited:
from kibana import AsyncKibana import asyncio async def main(): async with AsyncKibana("http://localhost:5601") as client: # Create an SLO (async) slo = await client.slos.create( name="async availability", description="Availability SLO created asynchronously", indicator={ "type": "sli.kql.custom", "params": { "index": "my-service-logs", "good": "status: ok", "total": "", "timestampField": "@timestamp", }, }, time_window={"duration": "30d", "type": "rolling"}, budgeting_method="occurrences", objective={"target": 0.99}, ) # Find SLOs (async) results = await client.slos.find() # Delete (async) await client.slos.delete(slo_id=slo.body["id"]) asyncio.run(main())
- async find(*, kql_query=None, size=None, search_after=None, page=None, per_page=None, sort_by=None, sort_direction=None, hide_stale=None, space_id=None, validate_spaces=None)[source]¶
Get a paginated list of SLOs with their summaries.
GET /api/observability/slos- Parameters:
kql_query (str | None) – A valid KQL query to filter SLOs with, e.g.
'slo.name:latency* and slo.tags : "prod"'.size (int | None) – The page size to use for cursor-based pagination (must be used together with
search_after). Default 1.search_after (list[str] | None) – The cursor to use for fetching the results from, when using a cursor-based pagination.
page (int | None) – The page number to return (default 1). Offset-based pagination; mutually exclusive with
size/search_after.per_page (int | None) – The number of SLOs to return per page (default 25, maximum 5000).
sort_by (str | None) – Sort by field. One of
"sli_value","status","error_budget_consumed","error_budget_remaining"(default"status").sort_direction (str | None) – Sort order,
"asc"(default) or"desc".hide_stale (bool | None) – Hide stale SLOs from the list as defined by the stale SLO threshold in SLO settings.
space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with
page,perPage,totalandresults(a list of SLOs with theirsummary).- Raises:
BadRequestError – If a query parameter is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = await client.slos.find(kql_query="slo.name:my-service*") >>> for slo in resp.body["results"]: ... print(slo["name"], slo["summary"]["status"])
- async create(*, name, description, indicator, time_window, budgeting_method, objective, id=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶
Create an SLO.
POST /api/observability/slos- Parameters:
name (str) – A name for the SLO.
description (str) – A description for the SLO.
indicator (dict[str, Any]) – The indicator (SLI) definition, e.g.
{"type": "sli.kql.custom", "params": {...}}. Supported types:sli.apm.transactionDuration,sli.apm.transactionErrorRate,sli.kql.custom,sli.metric.custom,sli.histogram.custom,sli.metric.timeslice.time_window (dict[str, Any]) – The SLO time window, e.g.
{"duration": "30d", "type": "rolling"}or{"duration": "1M", "type": "calendarAligned"}.budgeting_method (str) –
"occurrences"or"timeslices".objective (dict[str, Any]) – The objective, e.g.
{"target": 0.99}; timeslices budgeting also usestimesliceTargetandtimesliceWindow.id (str | None) – An optional unique identifier for the SLO (8-36 characters); autogenerated when omitted.
settings (dict[str, Any] | None) – Optional settings such as
syncDelay,frequency,syncFieldandpreventInitialBackfill.group_by (str | list[str] | None) – Optional field or list of fields to generate one SLO per distinct value (e.g.
"service.name").artifacts (dict[str, Any] | None) – Optional links to related assets, e.g.
{"dashboards": [{"id": "..."}]}.space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse containing the
idof the created SLO.- Raises:
BadRequestError – If the SLO definition is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
ConflictError – If an SLO with the same id already exists (409).
- Return type:
Example
>>> resp = await client.slos.create( ... name="availability", ... description="99% good over 7 days", ... indicator={ ... "type": "sli.kql.custom", ... "params": { ... "index": "logs-*", ... "good": "status: ok", ... "total": "", ... "timestampField": "@timestamp", ... }, ... }, ... time_window={"duration": "7d", "type": "rolling"}, ... budgeting_method="occurrences", ... objective={"target": 0.99}, ... ) >>> print(resp.body["id"])
- async get(*, slo_id, instance_id=None, space_id=None, validate_spaces=None)[source]¶
Get an SLO by its identifier.
GET /api/observability/slos/{sloId}- Parameters:
slo_id (str) – The SLO identifier.
instance_id (str | None) – The SLO instance identifier when retrieving a specific instance of a grouped SLO.
space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with the SLO definition and its
summary.- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> slo = await client.slos.get(slo_id="my-slo-id") >>> print(slo.body["name"], slo.body["enabled"])
- async update(*, slo_id, name=None, description=None, indicator=None, time_window=None, budgeting_method=None, objective=None, settings=None, group_by=None, tags=None, artifacts=None, space_id=None, validate_spaces=None)[source]¶
Update an SLO.
PUT /api/observability/slos/{sloId}Performs a partial update: only the provided fields are changed. Updating fields that are part of the rollup transform (indicator, time window, budgeting method, objective, group_by or settings) triggers a re-computation of the SLO data.
- Parameters:
slo_id (str) – The SLO identifier.
name (str | None) – A new name for the SLO.
description (str | None) – A new description for the SLO.
indicator (dict[str, Any] | None) – The indicator (SLI) definition.
budgeting_method (str | None) –
"occurrences"or"timeslices".objective (dict[str, Any] | None) – The objective, e.g.
{"target": 0.99}.settings (dict[str, Any] | None) – Settings such as
syncDelay,frequency,syncFieldandpreventInitialBackfill.group_by (str | list[str] | None) – Field or list of fields to generate one SLO per distinct value.
artifacts (dict[str, Any] | None) – Links to related assets, e.g.
{"dashboards": [{"id": "..."}]}.space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with the updated SLO definition.
- Raises:
BadRequestError – If the update payload is invalid (400).
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = await client.slos.update( ... slo_id="my-slo-id", ... description="Updated description", ... tags=["prod"], ... )
- async delete(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Delete an SLO and its associated summary and rollup data.
DELETE /api/observability/slos/{sloId}- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> await client.slos.delete(slo_id="my-slo-id")
- async enable(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Enable an SLO, resuming data rollup and summary computation.
POST /api/observability/slos/{sloId}/enable- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> await client.slos.enable(slo_id="my-slo-id")
- async disable(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Disable an SLO, stopping data rollup and summary computation.
POST /api/observability/slos/{sloId}/disable- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> await client.slos.disable(slo_id="my-slo-id")
- async reset(*, slo_id, space_id=None, validate_spaces=None)[source]¶
Reset an SLO, deleting its data and recomputing it from scratch.
POST /api/observability/slos/{sloId}/_reset- Parameters:
- Returns:
ObjectApiResponse with the reset SLO definition.
- Raises:
NotFoundError – If the SLO does not exist (404).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = await client.slos.reset(slo_id="my-slo-id") >>> print(resp.body["version"])
- async delete_instances(*, instances, space_id=None, validate_spaces=None)[source]¶
Batch delete rollup and summary data for SLO instances.
POST /api/observability/slos/_delete_instancesDeletes the rollup and summary data for the given list of SLO id / instance id pairs. Useful for removing stale data of instances of a grouped SLO that no longer receive updates.
- Parameters:
- Returns:
ObjectApiResponse with an empty body (HTTP 204 on success).
- Raises:
BadRequestError – If the payload is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> await client.slos.delete_instances( ... instances=[ ... {"sloId": "my-slo-id", "instanceId": "my-service"}, ... ] ... )
- async bulk_delete(*, slo_ids, space_id=None, validate_spaces=None)[source]¶
Bulk delete SLO definitions and their summary and rollup data.
POST /api/observability/slos/_bulk_deleteThe deletion occurs asynchronously: the response contains a
taskIdthat can be polled withbulk_delete_status()to retrieve the operation outcome.- Parameters:
- Returns:
ObjectApiResponse containing the
taskIdof the bulk deletion.- Raises:
BadRequestError – If the payload is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = await client.slos.bulk_delete(slo_ids=["slo-1", "slo-2"]) >>> status = await client.slos.bulk_delete_status( ... task_id=resp.body["taskId"] ... )
- async bulk_delete_status(*, task_id, space_id=None, validate_spaces=None)[source]¶
Retrieve the status of an asynchronous bulk deletion.
GET /api/observability/slos/_bulk_delete/{taskId}- Parameters:
task_id (str) – The task id returned by
bulk_delete().space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with
isDone, per-SLOresultsand an optionalerror.- Raises:
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> status = await client.slos.bulk_delete_status(task_id="...") >>> if status.body["isDone"]: ... for result in status.body["results"]: ... print(result["id"], result["success"])
- async bulk_purge_rollup(*, slo_ids, purge_policy, space_id=None, validate_spaces=None)[source]¶
Batch delete rollup and summary data according to a purge policy.
POST /api/observability/slos/_bulk_purge_rollupThe deletion occurs asynchronously (Elasticsearch delete-by-query) and the response contains an Elasticsearch task id.
Note
Kibana 9.4.3 expects
purgeTypevalues"fixed_age"/"fixed_time"(snake_case), although the official OpenAPI spec documents"fixed-age"/"fixed-time".- Parameters:
slo_ids (list[str]) – A list of SLO ids whose rollup data should be purged.
purge_policy (dict[str, Any]) – Policy dictating which SLI documents to purge, either
{"purgeType": "fixed_age", "age": "7d"}or{"purgeType": "fixed_time", "timestamp": "2024-01-01T00:00:00.000Z"}.space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse containing the Elasticsearch
taskIdof the purge operation.- Raises:
BadRequestError – If the purge policy is invalid or would purge data inside an SLO time window (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = await client.slos.bulk_purge_rollup( ... slo_ids=["my-slo-id"], ... purge_policy={"purgeType": "fixed_age", "age": "30d"}, ... ) >>> print(resp.body["taskId"])
- async find_definitions(*, include_outdated_only=None, include_health=None, tags=None, search=None, page=None, per_page=None, space_id=None, validate_spaces=None)[source]¶
Get the SLO definitions (without computed summaries).
GET /api/observability/slos/_definitionsNote
The official OpenAPI spec documents this operation under
/internal/observability/slos/_definitions, but Kibana 9.4.3 serves it on the public/api/observability/slos/_definitionspath (the internal path returns 404); this client uses the public path.- Parameters:
include_outdated_only (bool | None) – Indicates if the API returns only outdated SLOs or all SLOs.
include_health (bool | None) – Include the health of the SLO transforms in the response.
tags (str | None) – Filter SLOs by tag (comma-separated list of tags).
search (str | None) – Filter SLOs by name, e.g.
"my-service*".page (int | None) – The page number to return.
per_page (int | None) – The number of SLOs to return per page (default 100, maximum 1000).
space_id (str | None) – Space to operate on;
Nonetargets the default space.validate_spaces (bool | None) – Override space-existence validation for this call.
- Returns:
ObjectApiResponse with
page,perPage,totalandresults(a list of SLO definitions).- Raises:
BadRequestError – If a query parameter is invalid (400).
AuthenticationException – If authentication fails (401).
AuthorizationException – If insufficient privileges (403).
- Return type:
Example
>>> resp = await client.slos.find_definitions(search="my-service*") >>> print(resp.body["total"])
- __init__(client, default_space_id=None, validate_spaces=True)¶
Initialize AsyncNamespaceClient with optional space support.