StatusClient

Client for monitoring Kibana server health and statistics through the Status API.

The Status API provides information about the Kibana server’s operational state, including overall health status, individual service statuses, and detailed operational metrics.

class kibana._sync.client.status.StatusClient(client)

Bases: NamespaceClient

Client for Kibana Status and system information API operations.

The Status API provides health and operational information about the Kibana server and its dependencies. This is useful for monitoring, health checks, and troubleshooting. It also exposes the Kibana features registry (GET /api/features, technical preview in 9.4).

Status levels (Kibana 8/9 “v8” format):

• available: All services are operational

• degraded: Some services are experiencing issues but Kibana is functional

• unavailable: Critical services are down, Kibana may not be functional

• critical: Kibana is in a critical state (defined by the spec enum)

Example

>>> from kibana import Kibana
>>> client = Kibana("http://localhost:5601", api_key="...")
>>>
>>> # Check overall status
>>> status = client.status.get_status()
>>> print(status.body["status"]["overall"]["level"])
available
>>>
>>> # Check core service statuses (elasticsearch, savedObjects)
>>> for service, info in status.body["status"]["core"].items():
...     print(f"{service}: {info['level']}")
elasticsearch: available
savedObjects: available
>>>
>>> # Get operational statistics
>>> stats = client.status.get_stats()
>>> print(stats.body["process"]["memory"]["heap"]["used_bytes"])

Overview

The StatusClient provides methods to check Kibana server health and retrieve operational statistics. This is useful for monitoring, alerting, and health checks.

Checking Server Status

Get the current Kibana server status with the get_status() method:

from kibana import Kibana

client = Kibana("http://localhost:5601")

# Get server status
status = client.status.get_status()

# Check overall status level
overall_status = status.body["status"]["overall"]["level"]
print(f"Kibana status: {overall_status}")

# Status levels: "available", "degraded", or "unavailable"
if overall_status == "available":
    print("✓ Kibana is healthy")
elif overall_status == "degraded":
    print("⚠ Kibana is degraded")
else:
    print("✗ Kibana is unavailable")

Status Response Structure

The status response includes detailed information about each service:

status = client.status.get_status()

# Overall status
overall = status.body["status"]["overall"]
print(f"Overall: {overall['level']} - {overall['summary']}")

# Individual service statuses
for service_name, service_status in status.body["status"]["statuses"].items():
    level = service_status["level"]
    summary = service_status.get("summary", "")
    print(f"{service_name}: {level} - {summary}")

Example status levels:

• available - Service is fully operational

• degraded - Service is operational but with issues

• unavailable - Service is not operational

Retrieving Operational Statistics

Get detailed operational metrics with the get_stats() method:

# Get operational statistics
stats = client.status.get_stats()

# Process information
process = stats.body["process"]
print(f"Uptime: {process['uptime_in_millis']} ms")
print(f"Memory usage: {process['memory']['heap']['used_in_bytes']} bytes")

# OS information
os_info = stats.body["os"]
print(f"Platform: {os_info['platform']}")
print(f"Load average: {os_info['load']}")

# Response times
response_times = stats.body["response_times"]
print(f"Average response time: {response_times['avg_in_millis']} ms")
print(f"Max response time: {response_times['max_in_millis']} ms")

Statistics Response Structure

The statistics response includes:

• Process metrics: Uptime, memory usage, event loop delay

• OS metrics: Platform, CPU count, load average, memory

• Response times: Average, max response times

• Requests: Total requests, disconnects, status codes

• Concurrent connections: Current connection count

stats = client.status.get_stats()

# Memory usage
heap = stats.body["process"]["memory"]["heap"]
print(f"Heap used: {heap['used_in_bytes'] / 1024 / 1024:.2f} MB")
print(f"Heap total: {heap['total_in_bytes'] / 1024 / 1024:.2f} MB")
print(f"Heap limit: {heap['size_limit'] / 1024 / 1024:.2f} MB")

# Request statistics
requests = stats.body["requests"]
print(f"Total requests: {requests['total']}")
print(f"Disconnects: {requests['disconnects']}")

# Status code breakdown
for code, count in requests["status_codes"].items():
    print(f"HTTP {code}: {count} requests")

Health Check Integration

Use the Status API for health checks and monitoring:

def check_kibana_health(client):
    """Check if Kibana is healthy."""
    try:
        status = client.status.get_status()
        level = status.body["status"]["overall"]["level"]

        if level == "available":
            return True, "Kibana is healthy"
        elif level == "degraded":
            return False, "Kibana is degraded"
        else:
            return False, "Kibana is unavailable"
    except Exception as e:
        return False, f"Failed to check status: {e}"

# Use in monitoring
is_healthy, message = check_kibana_health(client)
if not is_healthy:
    # Send alert
    print(f"ALERT: {message}")

Monitoring Best Practices

Best practices for using the Status API:

1. Regular health checks: Poll the status endpoint periodically

2. Alert on degradation: Set up alerts for degraded or unavailable status

3. Track metrics over time: Store statistics for trend analysis

4. Monitor response times: Watch for increasing response times

5. Check memory usage: Alert on high memory usage

import time

def monitor_kibana(client, interval=60):
    """Monitor Kibana health continuously."""
    while True:
        try:
            # Check status
            status = client.status.get_status()
            level = status.body["status"]["overall"]["level"]

            # Get stats
            stats = client.status.get_stats()
            uptime = stats.body["process"]["uptime_in_millis"]
            heap_used = stats.body["process"]["memory"]["heap"]["used_in_bytes"]

            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}]")
            print(f"  Status: {level}")
            print(f"  Uptime: {uptime / 1000 / 60:.2f} minutes")
            print(f"  Heap: {heap_used / 1024 / 1024:.2f} MB")

            if level != "available":
                print(f"  WARNING: Kibana is {level}")

        except Exception as e:
            print(f"  ERROR: Failed to get status: {e}")

        time.sleep(interval)

Error Handling

Handle errors when checking status:

from kibana.exceptions import (
    ConnectionError,
    AuthenticationException,
    TransportError
)

try:
    status = client.status.get_status()
except AuthenticationException as e:
    print(f"Authentication failed: {e.message}")
except ConnectionError as e:
    print(f"Cannot connect to Kibana: {e}")
except TransportError as e:
    print(f"Transport error: {e}")

__init__(client)

Initialize the StatusClient.

Parameters:

client (Kibana) – The parent Kibana client instance to delegate requests to.

Example

>>> status_client = StatusClient(kibana_client)

get_status(*, v7format=None, v8format=None)

Get the current status of the Kibana server.

Returns comprehensive health information about the Kibana server and its dependencies. This endpoint is commonly used for health checks in monitoring systems and load balancers.

The default (v8) response includes:

• Overall status level (available, degraded, unavailable, critical)

• Core service statuses under status.core (elasticsearch, savedObjects)

• Per-plugin statuses under status.plugins

• Version information (Kibana version, build number)

• Server identification (name, UUID) and runtime metrics

Parameters:

• v7format (bool | None) – Set to True to get the response in the legacy v7 format, where status.overall has state/title keys and status.statuses is a list of service entries. Mutually exclusive with v8format.

• v8format (bool | None) – Set to True to explicitly request the v8 format (the default shape described below). Mutually exclusive with v7format; passing both yields a 400 Bad Request.

Returns:

ObjectApiResponse containing status information with the following structure (default v8 format):

• status.overall.level – Overall status level

• status.overall.summary – Human-readable status summary

• status.core – Dict of core service statuses (elasticsearch, savedObjects)

• status.plugins – Dict of per-plugin statuses

• version – Kibana version information

• name – Server name

• uuid – Server UUID

• metrics – Last-collected runtime metrics

Raises:

• BadRequestError – If both v7format and v8format are provided.

• ApiError – If Kibana returns an error response. Per the 9.4.3 spec, a 503 (with a status body) is returned when Kibana or an essential service is unavailable.

• TransportError – If unable to connect to Kibana.

Return type:

ObjectApiResponse[dict[str, Any]]

Note

This endpoint is anonymously accessible. Unauthenticated callers receive HTTP 200 with a redacted minimal body containing only {"status": {"overall": {"level": ...}}} — top-level name, uuid and version keys are absent in that case.

Example

>>> # Basic status check
>>> status = client.status.get_status()
>>> if status.body["status"]["overall"]["level"] == "available":
...     print("Kibana is healthy")
... else:
...     print("Kibana has issues")
Kibana is healthy
>>>
>>> # Check a core service status
>>> es_status = status.body["status"]["core"]["elasticsearch"]
>>> print(f"Elasticsearch: {es_status['level']}")
Elasticsearch: available
>>>
>>> # Get version information
>>> version = status.body["version"]
>>> print(f"Kibana {version['number']} (build {version['build_number']})")
Kibana 9.4.3 (build 102392)
>>>
>>> # Legacy v7 format: statuses is a list
>>> legacy = client.status.get_status(v7format=True)
>>> print(legacy.body["status"]["overall"]["state"])
green

get_stats(*, extended=None, legacy=None, exclude_usage=None)

Get operational statistics about the Kibana server.

Returns detailed performance and resource utilization metrics for the Kibana server. This is useful for monitoring, capacity planning, and performance troubleshooting.

Note

GET /api/stats is not part of the official Kibana 9.4.3 OpenAPI document, but it is served by the usage-collection plugin and verified working against a live 9.4.3 server.

The response includes:

• process: Process metrics — memory.heap.used_bytes / total_bytes, memory.resident_set_size_bytes, uptime_ms, event loop delay/utilization

• os: OS metrics — platform, platform_release, load, memory.total_bytes / free_bytes / used_bytes, uptime_ms

• response_times: HTTP response times (avg_ms, max_ms)

• requests: Request counts (total, disconnects, status_codes)

• concurrent_connections: Current connection count

• kibana: Server identification (uuid, name, version, status)

• elasticsearch_client: ES client socket/queue statistics

Parameters:

• extended (bool | None) – When True, include additional payload such as usage and the cluster_uuid.

• legacy (bool | None) – When True, format the extended payload in the legacy (camelCase) style, e.g. clusterUuid instead of cluster_uuid.

• exclude_usage (bool | None) – When True (with extended), skip collecting the usage payload. On live 9.4.3 the usage key is present but empty either way (usage collection moved out of /api/stats).

Returns:

• process: Process-level metrics (memory, uptime_ms, event loop)

• os: Operating system metrics (platform, load, memory)

• response_times: HTTP response time statistics (avg_ms, max_ms)

• requests: Request count statistics

• concurrent_connections: Current connection count

• kibana: Server metadata (uuid, name, version, status)

Return type:

ObjectApiResponse containing statistics with the following structure

Raises:

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to view stats.

• TransportError – If unable to connect to Kibana.

Example

>>> # Get server statistics
>>> stats = client.status.get_stats()
>>>
>>> # Check memory usage (9.x field names)
>>> heap = stats.body["process"]["memory"]["heap"]
>>> used_mb = heap["used_bytes"] / (1024 * 1024)
>>> total_mb = heap["total_bytes"] / (1024 * 1024)
>>> print(f"Heap: {used_mb:.1f}MB / {total_mb:.1f}MB")
Heap: 245.3MB / 512.0MB
>>>
>>> # Check uptime
>>> uptime_hours = stats.body["process"]["uptime_ms"] / 3600000
>>> print(f"Uptime: {uptime_hours:.1f} hours")
Uptime: 24.5 hours
>>>
>>> # Check response times
>>> response_times = stats.body.get("response_times", {})
>>> if "avg_ms" in response_times:
...     print(f"Avg response time: {response_times['avg_ms']:.0f}ms")
Avg response time: 45ms
>>>
>>> # Check concurrent connections
>>> connections = stats.body.get("concurrent_connections", 0)
>>> print(f"Active connections: {connections}")
Active connections: 12

get_features()

Get information about all Kibana features.

Features are used by spaces and security to refine and secure access to Kibana. Each feature describes its category, associated apps, catalogue entries, and the privileges it exposes.

Note

Technical preview in 9.4 — this endpoint (GET /api/features) may change or be removed in a future release.

Returns:

ObjectApiResponse whose body is a JSON array of feature objects (a ListApiResponse at runtime). Each feature object contains:

• id – Feature identifier (e.g. "dashboard_v2")

• name – Human-readable feature name

• category – Feature category (id, label, order)

• app – List of associated Kibana app IDs

• catalogue – List of associated catalogue entry IDs

• privileges – Privilege definitions (all/read), when the feature exposes security privileges

• order – Display ordering hint (optional)

Raises:

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read features.

• TransportError – If unable to connect to Kibana.

Return type:

ObjectApiResponse[Any]

Example

>>> features = client.status.get_features()
>>> for feature in features.body[:3]:
...     print(feature["id"], "-", feature["name"])
searchSynonyms - Synonyms
discover_v2 - Discover
dashboard_v2 - Dashboard

perform_request(method, path, *, params=None, headers=None, body=None)

Perform an HTTP request via the parent client with space context enhancement.

Parameters:

• method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)

• path (str) – API endpoint path

• params (dict[str, Any] | None) – Query parameters

• headers (dict[str, str] | None) – Request headers

• body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

AsyncStatusClient

Asynchronous version of the StatusClient for use with async/await syntax.

class kibana._async.client.status.AsyncStatusClient(client)

Bases: AsyncNamespaceClient

Async client for Kibana Status and system information API operations.

The Status API provides health and operational information about the Kibana server and its dependencies. This is useful for monitoring, health checks, and troubleshooting. It also exposes the Kibana features registry (GET /api/features, technical preview in 9.4).

Status levels (Kibana 8/9 “v8” format):

• available: All services are operational

• degraded: Some services are experiencing issues but Kibana is functional

• unavailable: Critical services are down, Kibana may not be functional

• critical: Kibana is in a critical state (defined by the spec enum)

Example

>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>>
>>> # Check overall status
>>> status = await client.status.get_status()
>>> print(status.body["status"]["overall"]["level"])
available
>>>
>>> # Check core service statuses (elasticsearch, savedObjects)
>>> for service, info in status.body["status"]["core"].items():
...     print(f"{service}: {info['level']}")
elasticsearch: available
savedObjects: available
>>>
>>> # Get operational statistics
>>> stats = await client.status.get_stats()
>>> print(stats.body["process"]["memory"]["heap"]["used_bytes"])

Usage

The AsyncStatusClient provides the same methods as StatusClient 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:
        # Get status (async)
        status = await client.status.get_status()
        print(f"Status: {status.body['status']['overall']['level']}")

        # Get stats (async)
        stats = await client.status.get_stats()
        print(f"Uptime: {stats.body['process']['uptime_in_millis']} ms")

asyncio.run(main())

Concurrent Monitoring

Monitor multiple Kibana instances concurrently:

import asyncio
from kibana import AsyncKibana

async def check_instance(url):
    """Check status of a single Kibana instance."""
    async with AsyncKibana(url) as client:
        status = await client.status.get_status()
        return {
            "url": url,
            "level": status.body["status"]["overall"]["level"]
        }

async def monitor_cluster():
    """Monitor multiple Kibana instances."""
    instances = [
        "http://kibana1:5601",
        "http://kibana2:5601",
        "http://kibana3:5601"
    ]

    # Check all instances concurrently
    results = await asyncio.gather(
        *[check_instance(url) for url in instances],
        return_exceptions=True
    )

    for result in results:
        if isinstance(result, Exception):
            print(f"Error: {result}")
        else:
            print(f"{result['url']}: {result['level']}")

asyncio.run(monitor_cluster())

Async Health Monitoring

Implement continuous async health monitoring:

import asyncio
from kibana import AsyncKibana

async def monitor_health(client, interval=60):
    """Monitor Kibana health continuously (async)."""
    while True:
        try:
            # Get status and stats concurrently
            status, stats = await asyncio.gather(
                client.status.get_status(),
                client.status.get_stats()
            )

            level = status.body["status"]["overall"]["level"]
            uptime = stats.body["process"]["uptime_in_millis"]

            print(f"Status: {level}, Uptime: {uptime / 1000:.2f}s")

            if level != "available":
                print(f"WARNING: Kibana is {level}")

        except Exception as e:
            print(f"ERROR: {e}")

        await asyncio.sleep(interval)

async def main():
    async with AsyncKibana("http://localhost:5601") as client:
        await monitor_health(client, interval=30)

asyncio.run(main())

__init__(client)

Initialize the AsyncStatusClient.

Parameters:

client (AsyncKibana) – The parent AsyncKibana client instance to delegate requests to.

Example

>>> status_client = AsyncStatusClient(kibana_client)

async get_status(*, v7format=None, v8format=None)

Get the current status of the Kibana server.

Returns comprehensive health information about the Kibana server and its dependencies. This endpoint is commonly used for health checks in monitoring systems and load balancers.

The default (v8) response includes:

• Overall status level (available, degraded, unavailable, critical)

• Core service statuses under status.core (elasticsearch, savedObjects)

• Per-plugin statuses under status.plugins

• Version information (Kibana version, build number)

• Server identification (name, UUID) and runtime metrics

Parameters:

• v7format (bool | None) – Set to True to get the response in the legacy v7 format, where status.overall has state/title keys and status.statuses is a list of service entries. Mutually exclusive with v8format.

• v8format (bool | None) – Set to True to explicitly request the v8 format (the default shape described below). Mutually exclusive with v7format; passing both yields a 400 Bad Request.

Returns:

ObjectApiResponse containing status information with the following structure (default v8 format):

• status.overall.level – Overall status level

• status.overall.summary – Human-readable status summary

• status.core – Dict of core service statuses (elasticsearch, savedObjects)

• status.plugins – Dict of per-plugin statuses

• version – Kibana version information

• name – Server name

• uuid – Server UUID

• metrics – Last-collected runtime metrics

Raises:

• BadRequestError – If both v7format and v8format are provided.

• ApiError – If Kibana returns an error response. Per the 9.4.3 spec, a 503 (with a status body) is returned when Kibana or an essential service is unavailable.

• TransportError – If unable to connect to Kibana.

Return type:

ObjectApiResponse[dict[str, Any]]

Note

This endpoint is anonymously accessible. Unauthenticated callers receive HTTP 200 with a redacted minimal body containing only {"status": {"overall": {"level": ...}}} — top-level name, uuid and version keys are absent in that case.

Example

>>> # Basic status check
>>> status = await client.status.get_status()
>>> if status.body["status"]["overall"]["level"] == "available":
...     print("Kibana is healthy")
... else:
...     print("Kibana has issues")
Kibana is healthy
>>>
>>> # Check a core service status
>>> es_status = status.body["status"]["core"]["elasticsearch"]
>>> print(f"Elasticsearch: {es_status['level']}")
Elasticsearch: available
>>>
>>> # Get version information
>>> version = status.body["version"]
>>> print(f"Kibana {version['number']} (build {version['build_number']})")
Kibana 9.4.3 (build 102392)
>>>
>>> # Legacy v7 format: statuses is a list
>>> legacy = await client.status.get_status(v7format=True)
>>> print(legacy.body["status"]["overall"]["state"])
green

async get_stats(*, extended=None, legacy=None, exclude_usage=None)

Get operational statistics about the Kibana server.

Returns detailed performance and resource utilization metrics for the Kibana server. This is useful for monitoring, capacity planning, and performance troubleshooting.

Note

GET /api/stats is not part of the official Kibana 9.4.3 OpenAPI document, but it is served by the usage-collection plugin and verified working against a live 9.4.3 server.

The response includes:

• process: Process metrics — memory.heap.used_bytes / total_bytes, memory.resident_set_size_bytes, uptime_ms, event loop delay/utilization

• os: OS metrics — platform, platform_release, load, memory.total_bytes / free_bytes / used_bytes, uptime_ms

• response_times: HTTP response times (avg_ms, max_ms)

• requests: Request counts (total, disconnects, status_codes)

• concurrent_connections: Current connection count

• kibana: Server identification (uuid, name, version, status)

• elasticsearch_client: ES client socket/queue statistics

Parameters:

• extended (bool | None) – When True, include additional payload such as usage and the cluster_uuid.

• legacy (bool | None) – When True, format the extended payload in the legacy (camelCase) style, e.g. clusterUuid instead of cluster_uuid.

• exclude_usage (bool | None) – When True (with extended), skip collecting the usage payload. On live 9.4.3 the usage key is present but empty either way (usage collection moved out of /api/stats).

Returns:

• process: Process-level metrics (memory, uptime_ms, event loop)

• os: Operating system metrics (platform, load, memory)

• response_times: HTTP response time statistics (avg_ms, max_ms)

• requests: Request count statistics

• concurrent_connections: Current connection count

• kibana: Server metadata (uuid, name, version, status)

Return type:

ObjectApiResponse containing statistics with the following structure

Raises:

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to view stats.

• TransportError – If unable to connect to Kibana.

Example

>>> # Get server statistics
>>> stats = await client.status.get_stats()
>>>
>>> # Check memory usage (9.x field names)
>>> heap = stats.body["process"]["memory"]["heap"]
>>> used_mb = heap["used_bytes"] / (1024 * 1024)
>>> total_mb = heap["total_bytes"] / (1024 * 1024)
>>> print(f"Heap: {used_mb:.1f}MB / {total_mb:.1f}MB")
Heap: 245.3MB / 512.0MB
>>>
>>> # Check uptime
>>> uptime_hours = stats.body["process"]["uptime_ms"] / 3600000
>>> print(f"Uptime: {uptime_hours:.1f} hours")
Uptime: 24.5 hours
>>>
>>> # Check response times
>>> response_times = stats.body.get("response_times", {})
>>> if "avg_ms" in response_times:
...     print(f"Avg response time: {response_times['avg_ms']:.0f}ms")
Avg response time: 45ms
>>>
>>> # Check concurrent connections
>>> connections = stats.body.get("concurrent_connections", 0)
>>> print(f"Active connections: {connections}")
Active connections: 12

async get_features()

Get information about all Kibana features.

Features are used by spaces and security to refine and secure access to Kibana. Each feature describes its category, associated apps, catalogue entries, and the privileges it exposes.

Note

Technical preview in 9.4 — this endpoint (GET /api/features) may change or be removed in a future release.

Returns:

ObjectApiResponse whose body is a JSON array of feature objects (a ListApiResponse at runtime). Each feature object contains:

• id – Feature identifier (e.g. "dashboard_v2")

• name – Human-readable feature name

• category – Feature category (id, label, order)

• app – List of associated Kibana app IDs

• catalogue – List of associated catalogue entry IDs

• privileges – Privilege definitions (all/read), when the feature exposes security privileges

• order – Display ordering hint (optional)

Raises:

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read features.

• TransportError – If unable to connect to Kibana.

Return type:

ObjectApiResponse[Any]

Example

>>> features = await client.status.get_features()
>>> for feature in features.body[:3]:
...     print(feature["id"], "-", feature["name"])
searchSynonyms - Synonyms
discover_v2 - Discover
dashboard_v2 - Dashboard

async perform_request(method, path, *, params=None, headers=None, body=None)

Perform an async HTTP request via the parent client with space context enhancement.

Parameters:

• method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)

• path (str) – API endpoint path

• params (dict[str, Any] | None) – Query parameters

• headers (dict[str, str] | None) – Request headers

• body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]