Monitoring & Alerting API

The monitoring and alerting system provides comprehensive metrics collection, security violation tracking, and multi-channel alerting for Kailash workflows.

Monitoring and alerting system for Kailash SDK.

Provides comprehensive monitoring for validation failures, security violations, performance metrics, and alerting for critical events. Includes specialized AsyncSQL lock contention monitoring.

class kailash.monitoring.ValidationMetrics[source]

Bases: MetricsCollector

Metrics collector for validation operations.

__init__()[source]: Initialize validation metrics collector.

record_validation_attempt(node_type: str, success: bool, duration_ms: float, cached: bool = False)[source]

Record a validation attempt.

Parameters:

node_type (str) – Type of node being validated
success (bool) – Whether validation succeeded
duration_ms (float) – Validation duration in milliseconds
cached (bool) – Whether result came from cache

get_success_rate(time_window: timedelta = datetime.timedelta(seconds=3600)) → float[source]

Get validation success rate over time window.

Parameters:: time_window (timedelta)
Return type:: float

get_cache_hit_rate(time_window: timedelta = datetime.timedelta(seconds=3600)) → float[source]

Get cache hit rate over time window.

Parameters:: time_window (timedelta)
Return type:: float

clear_metrics(): Clear all metrics.

create_metric(name: str, metric_type: MetricType, description: str, unit: str = '') → MetricSeries

Create a new metric series.

Parameters:

name (str) – Metric name
metric_type (MetricType) – Type of metric
description (str) – Description of what this metric measures
unit (str) – Unit of measurement

Returns:

MetricSeries instance

Return type:

MetricSeries

get_all_metrics() → Dict[str, MetricSeries]

Get all metric series.

Return type:: Dict[str, MetricSeries]

get_metric(name: str) → MetricSeries | None

Get a metric series by name.

Parameters:: name (str)
Return type:: MetricSeries | None

increment(name: str, value: int | float = 1, labels: Dict[str, str] | None = None)

Increment a counter metric.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_histogram(name: str, value: int | float, labels: Dict[str, str] | None = None)

Record a histogram value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_timer(name: str, duration_ms: float, labels: Dict[str, str] | None = None)

Record a timer metric.

Parameters:

name (str)
duration_ms (float)
labels (Dict[str, str] | None)

set_gauge(name: str, value: int | float, labels: Dict[str, str] | None = None)

Set a gauge metric value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

class kailash.monitoring.SecurityMetrics[source]

Bases: MetricsCollector

Metrics collector for security events.

__init__()[source]: Initialize security metrics collector.

record_security_violation(violation_type: str, severity: MetricSeverity, source: str, details: Dict[str, Any] | None = None)[source]

Record a security violation.

Parameters:

violation_type (str) – Type of security violation
severity (MetricSeverity) – Severity level
source (str) – Source of the violation (node, connection, etc.)
details (Dict[str, Any] | None) – Additional violation details

record_blocked_connection(source_node: str, target_node: str, reason: str)[source]

Record a blocked connection.

Parameters:

source_node (str) – Source node identifier
target_node (str) – Target node identifier
reason (str) – Reason for blocking

get_violation_rate(time_window: timedelta = datetime.timedelta(seconds=3600)) → float[source]

Get security violation rate per minute.

Parameters:: time_window (timedelta)
Return type:: float

get_critical_violations(time_window: timedelta = datetime.timedelta(seconds=3600)) → int[source]

Get count of critical violations in time window.

Parameters:: time_window (timedelta)
Return type:: int

clear_metrics(): Clear all metrics.

create_metric(name: str, metric_type: MetricType, description: str, unit: str = '') → MetricSeries

Create a new metric series.

Parameters:

name (str) – Metric name
metric_type (MetricType) – Type of metric
description (str) – Description of what this metric measures
unit (str) – Unit of measurement

Returns:

MetricSeries instance

Return type:

MetricSeries

get_all_metrics() → Dict[str, MetricSeries]

Get all metric series.

Return type:: Dict[str, MetricSeries]

get_metric(name: str) → MetricSeries | None

Get a metric series by name.

Parameters:: name (str)
Return type:: MetricSeries | None

increment(name: str, value: int | float = 1, labels: Dict[str, str] | None = None)

Increment a counter metric.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_histogram(name: str, value: int | float, labels: Dict[str, str] | None = None)

Record a histogram value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_timer(name: str, duration_ms: float, labels: Dict[str, str] | None = None)

Record a timer metric.

Parameters:

name (str)
duration_ms (float)
labels (Dict[str, str] | None)

set_gauge(name: str, value: int | float, labels: Dict[str, str] | None = None)

Set a gauge metric value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

class kailash.monitoring.PerformanceMetrics[source]

Bases: MetricsCollector

Metrics collector for performance monitoring.

__init__()[source]: Initialize performance metrics collector.

record_operation(operation: str, duration_ms: float, success: bool)[source]

Record an operation performance.

Parameters:

operation (str) – Operation name
duration_ms (float) – Duration in milliseconds
success (bool) – Whether operation succeeded

update_system_metrics(memory_mb: float, cpu_percent: float, rps: float)[source]

Update system-level metrics.

Parameters:

memory_mb (float) – Memory usage in MB
cpu_percent (float) – CPU usage percentage
rps (float) – Requests per second

get_p95_response_time(time_window: timedelta = datetime.timedelta(seconds=3600)) → float | None[source]

Get 95th percentile response time.

Parameters:: time_window (timedelta)
Return type:: float | None

clear_metrics(): Clear all metrics.

create_metric(name: str, metric_type: MetricType, description: str, unit: str = '') → MetricSeries

Create a new metric series.

Parameters:

name (str) – Metric name
metric_type (MetricType) – Type of metric
description (str) – Description of what this metric measures
unit (str) – Unit of measurement

Returns:

MetricSeries instance

Return type:

MetricSeries

get_all_metrics() → Dict[str, MetricSeries]

Get all metric series.

Return type:: Dict[str, MetricSeries]

get_metric(name: str) → MetricSeries | None

Get a metric series by name.

Parameters:: name (str)
Return type:: MetricSeries | None

increment(name: str, value: int | float = 1, labels: Dict[str, str] | None = None)

Increment a counter metric.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_histogram(name: str, value: int | float, labels: Dict[str, str] | None = None)

Record a histogram value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_timer(name: str, duration_ms: float, labels: Dict[str, str] | None = None)

Record a timer metric.

Parameters:

name (str)
duration_ms (float)
labels (Dict[str, str] | None)

set_gauge(name: str, value: int | float, labels: Dict[str, str] | None = None)

Set a gauge metric value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

class kailash.monitoring.AlertManager(metrics_registry: MetricsRegistry)[source]

Bases: object

Alert manager for handling alerting rules and notifications.

Parameters:: metrics_registry (MetricsRegistry)

__init__(metrics_registry: MetricsRegistry)[source]

Initialize alert manager.

Parameters:: metrics_registry (MetricsRegistry) – Metrics registry to monitor

add_rule(rule: AlertRule)[source]

Add an alerting rule.

Parameters:: rule (AlertRule) – AlertRule to add

remove_rule(rule_name: str)[source]

Remove an alerting rule.

Parameters:: rule_name (str) – Name of rule to remove

add_notification_channel(channel: NotificationChannel)[source]

Add a notification channel.

Parameters:: channel (NotificationChannel) – NotificationChannel to add

start()[source]: Start the alert manager.

stop()[source]: Stop the alert manager.

get_active_alerts() → List[Alert][source]

Get all active (firing) alerts.

Return type:: List[Alert]

get_all_alerts() → List[Alert][source]

Get all alerts.

Return type:: List[Alert]

silence_alert(alert_id: str)[source]

Silence an alert.

Parameters:: alert_id (str) – Alert ID to silence

acknowledge_alert(alert_id: str)[source]

Acknowledge an alert (same as silence for now).

Parameters:: alert_id (str) – Alert ID to acknowledge

class kailash.monitoring.AlertRule(name: str, description: str, severity: AlertSeverity, metric_name: str, condition: str, threshold: int | float, time_window: timedelta = datetime.timedelta(seconds=300), evaluation_interval: timedelta = datetime.timedelta(seconds=60), notification_interval: timedelta = datetime.timedelta(seconds=900), labels: Dict[str, str]=<factory>, annotations: Dict[str, str]=<factory>, enabled: bool = True)[source]

Bases: object

Alert rule configuration.

Parameters:

name (str)
description (str)
severity (AlertSeverity)
metric_name (str)
condition (str)
threshold (int | float)
time_window (timedelta)
evaluation_interval (timedelta)
notification_interval (timedelta)
labels (Dict[str, str])
annotations (Dict[str, str])
enabled (bool)

name: str

description: str

severity: AlertSeverity

metric_name: str

condition: str

threshold: int | float

time_window: timedelta = datetime.timedelta(seconds=300)

evaluation_interval: timedelta = datetime.timedelta(seconds=60)

notification_interval: timedelta = datetime.timedelta(seconds=900)

labels: Dict[str, str]

annotations: Dict[str, str]

enabled: bool = True

evaluate(metric_series: MetricSeries) → bool[source]

Evaluate if alert condition is met.

Parameters:: metric_series (MetricSeries) – Metric series to evaluate
Returns:: True if alert condition is met
Return type:: bool

__init__(name: str, description: str, severity: AlertSeverity, metric_name: str, condition: str, threshold: int | float, time_window: timedelta = datetime.timedelta(seconds=300), evaluation_interval: timedelta = datetime.timedelta(seconds=60), notification_interval: timedelta = datetime.timedelta(seconds=900), labels: Dict[str, str]=<factory>, annotations: Dict[str, str]=<factory>, enabled: bool = True) → None

Parameters:

name (str)
description (str)
severity (AlertSeverity)
metric_name (str)
condition (str)
threshold (int | float)
time_window (timedelta)
evaluation_interval (timedelta)
notification_interval (timedelta)
labels (Dict[str, str])
annotations (Dict[str, str])
enabled (bool)

Return type:

None

class kailash.monitoring.AlertSeverity(value)[source]

Bases: Enum

Alert severity levels.

INFO = 'info'

WARNING = 'warning'

ERROR = 'error'

CRITICAL = 'critical'

classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name): Return the member matching name.

classmethod __iter__(): Return members in definition order.

classmethod __len__(): Return the number of members (no aliases)

class kailash.monitoring.AsyncSQLMetrics(enabled: bool = True, registry: Any = None)[source]

Bases: object

Prometheus metrics collector for AsyncSQL lock contention monitoring.

Parameters:

enabled (bool)
registry (Any)

__init__(enabled: bool = True, registry: Any = None)[source]

Initialize AsyncSQL metrics collector.

Parameters:

enabled (bool) – Whether to collect metrics (disabled if prometheus_client not available)
registry (Any) – Custom Prometheus registry (uses default if None)

record_lock_acquisition(pool_key: str, status: str, wait_time: float = 0.0)[source]

Record a lock acquisition event.

Parameters:

pool_key (str) – The pool key for the lock
status (str) – ‘success’, ‘timeout’, or ‘error’
wait_time (float) – Time spent waiting for the lock in seconds

set_active_locks(pool_key: str, count: int)[source]

Update the count of active locks for a pool.

Parameters:

pool_key (str) – The pool key
count (int) – Number of active locks

record_pool_operation(pool_key: str, operation: str)[source]

Record a pool operation event.

Parameters:

pool_key (str) – The pool key
operation (str) – ‘create’, ‘cleanup’, ‘acquire’, ‘release’

timed_lock_acquisition(pool_key: str)[source]

Context manager to time lock acquisition and automatically record metrics.

Usage:

async with metrics.timed_lock_acquisition('my_pool_key'):
    # Lock acquisition logic here
    async with some_lock:
        # Work while holding lock
        pass

Parameters:: pool_key (str)

kailash.monitoring.enable_metrics(registry: Any = None) → AsyncSQLMetrics[source]

Enable global AsyncSQL metrics collection.

Parameters:: registry (Any) – Custom Prometheus registry (uses default if None)
Returns:: The configured metrics instance
Return type:: AsyncSQLMetrics

kailash.monitoring.disable_metrics()[source]: Disable global AsyncSQL metrics collection.

kailash.monitoring.get_global_metrics() → AsyncSQLMetrics | None[source]

Get the global AsyncSQL metrics instance.

Return type:: AsyncSQLMetrics | None

kailash.monitoring.set_global_metrics(metrics: AsyncSQLMetrics | None)[source]

Set the global AsyncSQL metrics instance.

Parameters:: metrics (AsyncSQLMetrics | None)

kailash.monitoring.record_lock_acquisition(pool_key: str, status: str, wait_time: float = 0.0)[source]

Record a lock acquisition event using global metrics.

Parameters:

pool_key (str)
status (str)
wait_time (float)

kailash.monitoring.record_pool_operation(pool_key: str, operation: str)[source]

Record a pool operation event using global metrics.

Parameters:

pool_key (str)
operation (str)

kailash.monitoring.set_active_locks(pool_key: str, count: int)[source]

Update active locks count using global metrics.

Parameters:

pool_key (str)
count (int)

kailash.monitoring.integrate_with_async_sql()[source]

Example of how to integrate metrics with AsyncSQLDatabaseNode.

This would typically be called during AsyncSQL initialization or through a configuration setting.

Metrics Collection

Metrics collection and aggregation for monitoring system.

Provides detailed metrics for validation failures, security violations, and performance monitoring with time-series data collection.

class kailash.monitoring.metrics.MetricType(value)[source]

Bases: Enum

Types of metrics collected.

COUNTER = 'counter'

GAUGE = 'gauge'

HISTOGRAM = 'histogram'

TIMER = 'timer'

classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name): Return the member matching name.

classmethod __iter__(): Return members in definition order.

classmethod __len__(): Return the number of members (no aliases)

class kailash.monitoring.metrics.MetricSeverity(value)[source]

Bases: Enum

Severity levels for metrics.

LOW = 'low'

MEDIUM = 'medium'

HIGH = 'high'

CRITICAL = 'critical'

classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name): Return the member matching name.

classmethod __iter__(): Return members in definition order.

classmethod __len__(): Return the number of members (no aliases)

class kailash.monitoring.metrics.MetricPoint(timestamp: datetime, value: int | float, labels: Dict[str, str]=<factory>, metadata: Dict[str, ~typing.Any]=<factory>)[source]

Bases: object

Single metric data point.

Parameters:

timestamp (datetime)
value (int | float)
labels (Dict[str, str])
metadata (Dict[str, Any])

timestamp: datetime

value: int | float

labels: Dict[str, str]

metadata: Dict[str, Any]

__init__(timestamp: datetime, value: int | float, labels: Dict[str, str]=<factory>, metadata: Dict[str, ~typing.Any]=<factory>) → None

Parameters:

timestamp (datetime)
value (int | float)
labels (Dict[str, str])
metadata (Dict[str, Any])

Return type:

None

class kailash.monitoring.metrics.MetricSeries(name: str, metric_type: MetricType, description: str, unit: str = '', points: deque = <factory>)[source]

Bases: object

Time series of metric data points.

Parameters:

name (str)
metric_type (MetricType)
description (str)
unit (str)
points (deque)

name: str

metric_type: MetricType

description: str

unit: str = ''

points: deque

add_point(value: int | float, labels: Dict[str, str] | None = None, metadata: Dict[str, Any] | None = None)[source]

Add a new data point to the series.

Parameters:

value (int | float)
labels (Dict[str, str] | None)
metadata (Dict[str, Any] | None)

get_latest_value() → int | float | None[source]

Get the most recent metric value.

Return type:: int | float | None

get_average(time_window: timedelta | None = None) → float | None[source]

Get average value over time window.

Parameters:: time_window (timedelta | None)
Return type:: float | None

get_max(time_window: timedelta | None = None) → int | float | None[source]

Get maximum value over time window.

Parameters:: time_window (timedelta | None)
Return type:: int | float | None

get_rate(time_window: timedelta = datetime.timedelta(seconds=60)) → float | None[source]

Get rate of change over time window.

Parameters:: time_window (timedelta)
Return type:: float | None

__init__(name: str, metric_type: MetricType, description: str, unit: str = '', points: deque = <factory>) → None

Parameters:

name (str)
metric_type (MetricType)
description (str)
unit (str)
points (deque)

Return type:

None

class kailash.monitoring.metrics.MetricsCollector(max_series: int = 100)[source]

Bases: object

Base metrics collector.

Parameters:: max_series (int)

__init__(max_series: int = 100)[source]

Initialize metrics collector.

Parameters:: max_series (int) – Maximum number of metric series to track

create_metric(name: str, metric_type: MetricType, description: str, unit: str = '') → MetricSeries[source]

Create a new metric series.

Parameters:

name (str) – Metric name
metric_type (MetricType) – Type of metric
description (str) – Description of what this metric measures
unit (str) – Unit of measurement

Returns:

MetricSeries instance

Return type:

MetricSeries

increment(name: str, value: int | float = 1, labels: Dict[str, str] | None = None)[source]

Increment a counter metric.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

set_gauge(name: str, value: int | float, labels: Dict[str, str] | None = None)[source]

Set a gauge metric value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_timer(name: str, duration_ms: float, labels: Dict[str, str] | None = None)[source]

Record a timer metric.

Parameters:

name (str)
duration_ms (float)
labels (Dict[str, str] | None)

record_histogram(name: str, value: int | float, labels: Dict[str, str] | None = None)[source]

Record a histogram value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

get_metric(name: str) → MetricSeries | None[source]

Get a metric series by name.

Parameters:: name (str)
Return type:: MetricSeries | None

get_all_metrics() → Dict[str, MetricSeries][source]

Get all metric series.

Return type:: Dict[str, MetricSeries]

clear_metrics()[source]: Clear all metrics.

class kailash.monitoring.metrics.ValidationMetrics[source]

Bases: MetricsCollector

Metrics collector for validation operations.

__init__()[source]: Initialize validation metrics collector.

record_validation_attempt(node_type: str, success: bool, duration_ms: float, cached: bool = False)[source]

Record a validation attempt.

Parameters:

node_type (str) – Type of node being validated
success (bool) – Whether validation succeeded
duration_ms (float) – Validation duration in milliseconds
cached (bool) – Whether result came from cache

get_success_rate(time_window: timedelta = datetime.timedelta(seconds=3600)) → float[source]

Get validation success rate over time window.

Parameters:: time_window (timedelta)
Return type:: float

get_cache_hit_rate(time_window: timedelta = datetime.timedelta(seconds=3600)) → float[source]

Get cache hit rate over time window.

Parameters:: time_window (timedelta)
Return type:: float

clear_metrics(): Clear all metrics.

create_metric(name: str, metric_type: MetricType, description: str, unit: str = '') → MetricSeries

Create a new metric series.

Parameters:

name (str) – Metric name
metric_type (MetricType) – Type of metric
description (str) – Description of what this metric measures
unit (str) – Unit of measurement

Returns:

MetricSeries instance

Return type:

MetricSeries

get_all_metrics() → Dict[str, MetricSeries]

Get all metric series.

Return type:: Dict[str, MetricSeries]

get_metric(name: str) → MetricSeries | None

Get a metric series by name.

Parameters:: name (str)
Return type:: MetricSeries | None

increment(name: str, value: int | float = 1, labels: Dict[str, str] | None = None)

Increment a counter metric.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_histogram(name: str, value: int | float, labels: Dict[str, str] | None = None)

Record a histogram value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_timer(name: str, duration_ms: float, labels: Dict[str, str] | None = None)

Record a timer metric.

Parameters:

name (str)
duration_ms (float)
labels (Dict[str, str] | None)

set_gauge(name: str, value: int | float, labels: Dict[str, str] | None = None)

Set a gauge metric value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

class kailash.monitoring.metrics.SecurityMetrics[source]

Bases: MetricsCollector

Metrics collector for security events.

__init__()[source]: Initialize security metrics collector.

record_security_violation(violation_type: str, severity: MetricSeverity, source: str, details: Dict[str, Any] | None = None)[source]

Record a security violation.

Parameters:

violation_type (str) – Type of security violation
severity (MetricSeverity) – Severity level
source (str) – Source of the violation (node, connection, etc.)
details (Dict[str, Any] | None) – Additional violation details

record_blocked_connection(source_node: str, target_node: str, reason: str)[source]

Record a blocked connection.

Parameters:

source_node (str) – Source node identifier
target_node (str) – Target node identifier
reason (str) – Reason for blocking

get_violation_rate(time_window: timedelta = datetime.timedelta(seconds=3600)) → float[source]

Get security violation rate per minute.

Parameters:: time_window (timedelta)
Return type:: float

get_critical_violations(time_window: timedelta = datetime.timedelta(seconds=3600)) → int[source]

Get count of critical violations in time window.

Parameters:: time_window (timedelta)
Return type:: int

clear_metrics(): Clear all metrics.

create_metric(name: str, metric_type: MetricType, description: str, unit: str = '') → MetricSeries

Create a new metric series.

Parameters:

name (str) – Metric name
metric_type (MetricType) – Type of metric
description (str) – Description of what this metric measures
unit (str) – Unit of measurement

Returns:

MetricSeries instance

Return type:

MetricSeries

get_all_metrics() → Dict[str, MetricSeries]

Get all metric series.

Return type:: Dict[str, MetricSeries]

get_metric(name: str) → MetricSeries | None

Get a metric series by name.

Parameters:: name (str)
Return type:: MetricSeries | None

increment(name: str, value: int | float = 1, labels: Dict[str, str] | None = None)

Increment a counter metric.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_histogram(name: str, value: int | float, labels: Dict[str, str] | None = None)

Record a histogram value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_timer(name: str, duration_ms: float, labels: Dict[str, str] | None = None)

Record a timer metric.

Parameters:

name (str)
duration_ms (float)
labels (Dict[str, str] | None)

set_gauge(name: str, value: int | float, labels: Dict[str, str] | None = None)

Set a gauge metric value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

class kailash.monitoring.metrics.PerformanceMetrics[source]

Bases: MetricsCollector

Metrics collector for performance monitoring.

__init__()[source]: Initialize performance metrics collector.

record_operation(operation: str, duration_ms: float, success: bool)[source]

Record an operation performance.

Parameters:

operation (str) – Operation name
duration_ms (float) – Duration in milliseconds
success (bool) – Whether operation succeeded

update_system_metrics(memory_mb: float, cpu_percent: float, rps: float)[source]

Update system-level metrics.

Parameters:

memory_mb (float) – Memory usage in MB
cpu_percent (float) – CPU usage percentage
rps (float) – Requests per second

get_p95_response_time(time_window: timedelta = datetime.timedelta(seconds=3600)) → float | None[source]

Get 95th percentile response time.

Parameters:: time_window (timedelta)
Return type:: float | None

clear_metrics(): Clear all metrics.

create_metric(name: str, metric_type: MetricType, description: str, unit: str = '') → MetricSeries

Create a new metric series.

Parameters:

name (str) – Metric name
metric_type (MetricType) – Type of metric
description (str) – Description of what this metric measures
unit (str) – Unit of measurement

Returns:

MetricSeries instance

Return type:

MetricSeries

get_all_metrics() → Dict[str, MetricSeries]

Get all metric series.

Return type:: Dict[str, MetricSeries]

get_metric(name: str) → MetricSeries | None

Get a metric series by name.

Parameters:: name (str)
Return type:: MetricSeries | None

increment(name: str, value: int | float = 1, labels: Dict[str, str] | None = None)

Increment a counter metric.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_histogram(name: str, value: int | float, labels: Dict[str, str] | None = None)

Record a histogram value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

record_timer(name: str, duration_ms: float, labels: Dict[str, str] | None = None)

Record a timer metric.

Parameters:

name (str)
duration_ms (float)
labels (Dict[str, str] | None)

set_gauge(name: str, value: int | float, labels: Dict[str, str] | None = None)

Set a gauge metric value.

Parameters:

name (str)
value (int | float)
labels (Dict[str, str] | None)

class kailash.monitoring.metrics.MetricsRegistry[source]

Bases: object

Global registry for metrics collectors.

__init__()[source]: Initialize metrics registry.

register_collector(name: str, collector: MetricsCollector)[source]

Register a metrics collector.

Parameters:

name (str) – Collector name
collector (MetricsCollector) – MetricsCollector instance

get_collector(name: str) → MetricsCollector | None[source]

Get a metrics collector by name.

Parameters:: name (str) – Collector name
Returns:: MetricsCollector instance or None
Return type:: MetricsCollector | None

get_all_collectors() → Dict[str, MetricsCollector][source]

Get all registered collectors.

Return type:: Dict[str, MetricsCollector]

export_metrics(format: str = 'json') → str[source]

Export all metrics in specified format.

Parameters:: format (str) – Export format (“json”, “prometheus”)
Returns:: Formatted metrics string
Return type:: str

kailash.monitoring.metrics.get_metrics_registry() → MetricsRegistry[source]

Get the global metrics registry.

Return type:: MetricsRegistry

kailash.monitoring.metrics.render_prometheus_exposition(extra_lines: list[str] | None = None) → str[source]

Render the unified Prometheus exposition for a /metrics scrape (#1708).

Concatenates, in one OpenMetrics text body:

the custom MetricsRegistry (validation / security / performance),
the prometheus_client default registry — which includes both prometheus_client-native instruments (asyncsql, ML) AND the OTel meters bridged in by kailash.observability.configure_observability()’s Prometheus reader, and
optional extra_lines (e.g. connection-pool metrics).

Before #1708 the server /metrics exported only (1), so most collected metrics were invisible to Prometheus. Degrades gracefully when prometheus_client is not installed (part 2 is skipped).

Parameters:: extra_lines (list[str] | None)
Return type:: str

kailash.monitoring.metrics.get_validation_metrics() → ValidationMetrics[source]

Get the validation metrics collector.

Return type:: ValidationMetrics

kailash.monitoring.metrics.get_security_metrics() → SecurityMetrics[source]

Get the security metrics collector.

Return type:: SecurityMetrics

kailash.monitoring.metrics.get_performance_metrics() → PerformanceMetrics[source]

Get the performance metrics collector.

Return type:: PerformanceMetrics

Alert Management

Alerting system for monitoring validation failures and security violations.

Provides configurable alerting rules, notification channels, and alert management for critical events in the Kailash SDK validation system.

class kailash.monitoring.alerts.AlertSeverity(value)[source]

Bases: Enum

Alert severity levels.

INFO = 'info'

WARNING = 'warning'

ERROR = 'error'

CRITICAL = 'critical'

classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name): Return the member matching name.

classmethod __iter__(): Return members in definition order.

classmethod __len__(): Return the number of members (no aliases)

class kailash.monitoring.alerts.AlertStatus(value)[source]

Bases: Enum

Alert status.

PENDING = 'pending'

FIRING = 'firing'

RESOLVED = 'resolved'

SILENCED = 'silenced'

classmethod __contains__(member)

Return True if member is a member of this enum raises TypeError if member is not an enum member

note: in 3.12 TypeError will no longer be raised, and True will also be returned if member is the value of a member in this enum

classmethod __getitem__(name): Return the member matching name.

classmethod __iter__(): Return members in definition order.

classmethod __len__(): Return the number of members (no aliases)

class kailash.monitoring.alerts.Alert(id: str, rule_name: str, severity: AlertSeverity, title: str, description: str, labels: Dict[str, str]=<factory>, annotations: Dict[str, str]=<factory>, status: AlertStatus = AlertStatus.PENDING, created_at: datetime = <factory>, fired_at: datetime | None = None, resolved_at: datetime | None = None, last_notification: datetime | None = None, notification_count: int = 0)[source]

Bases: object

Alert instance.

Parameters:

id (str)
rule_name (str)
severity (AlertSeverity)
title (str)
description (str)
labels (Dict[str, str])
annotations (Dict[str, str])
status (AlertStatus)
created_at (datetime)
fired_at (datetime | None)
resolved_at (datetime | None)
last_notification (datetime | None)
notification_count (int)

id: str

rule_name: str

severity: AlertSeverity

title: str

description: str

labels: Dict[str, str]

annotations: Dict[str, str]

status: AlertStatus = 'pending'

created_at: datetime

fired_at: datetime | None = None

resolved_at: datetime | None = None

last_notification: datetime | None = None

notification_count: int = 0

fire()[source]: Mark alert as firing.

resolve()[source]: Mark alert as resolved.

silence()[source]: Silence the alert.

should_notify(notification_interval: timedelta) → bool[source]

Check if alert should send notification.

Parameters:: notification_interval (timedelta)
Return type:: bool

mark_notified()[source]: Mark that notification was sent.

__init__(id: str, rule_name: str, severity: AlertSeverity, title: str, description: str, labels: Dict[str, str]=<factory>, annotations: Dict[str, str]=<factory>, status: AlertStatus = AlertStatus.PENDING, created_at: datetime = <factory>, fired_at: datetime | None = None, resolved_at: datetime | None = None, last_notification: datetime | None = None, notification_count: int = 0) → None

Parameters:

id (str)
rule_name (str)
severity (AlertSeverity)
title (str)
description (str)
labels (Dict[str, str])
annotations (Dict[str, str])
status (AlertStatus)
created_at (datetime)
fired_at (datetime | None)
resolved_at (datetime | None)
last_notification (datetime | None)
notification_count (int)

Return type:

None

class kailash.monitoring.alerts.AlertRule(name: str, description: str, severity: AlertSeverity, metric_name: str, condition: str, threshold: int | float, time_window: timedelta = datetime.timedelta(seconds=300), evaluation_interval: timedelta = datetime.timedelta(seconds=60), notification_interval: timedelta = datetime.timedelta(seconds=900), labels: Dict[str, str]=<factory>, annotations: Dict[str, str]=<factory>, enabled: bool = True)[source]

Bases: object

Alert rule configuration.

Parameters:

name (str)
description (str)
severity (AlertSeverity)
metric_name (str)
condition (str)
threshold (int | float)
time_window (timedelta)
evaluation_interval (timedelta)
notification_interval (timedelta)
labels (Dict[str, str])
annotations (Dict[str, str])
enabled (bool)

name: str

description: str

severity: AlertSeverity

metric_name: str

condition: str

threshold: int | float

time_window: timedelta = datetime.timedelta(seconds=300)

evaluation_interval: timedelta = datetime.timedelta(seconds=60)

notification_interval: timedelta = datetime.timedelta(seconds=900)

labels: Dict[str, str]

annotations: Dict[str, str]

enabled: bool = True

evaluate(metric_series: MetricSeries) → bool[source]

Evaluate if alert condition is met.

Parameters:: metric_series (MetricSeries) – Metric series to evaluate
Returns:: True if alert condition is met
Return type:: bool

__init__(name: str, description: str, severity: AlertSeverity, metric_name: str, condition: str, threshold: int | float, time_window: timedelta = datetime.timedelta(seconds=300), evaluation_interval: timedelta = datetime.timedelta(seconds=60), notification_interval: timedelta = datetime.timedelta(seconds=900), labels: Dict[str, str]=<factory>, annotations: Dict[str, str]=<factory>, enabled: bool = True) → None

Parameters:

name (str)
description (str)
severity (AlertSeverity)
metric_name (str)
condition (str)
threshold (int | float)
time_window (timedelta)
evaluation_interval (timedelta)
notification_interval (timedelta)
labels (Dict[str, str])
annotations (Dict[str, str])
enabled (bool)

Return type:

None

class kailash.monitoring.alerts.NotificationChannel[source]

Bases: ABC

Base class for notification channels.

abstractmethod send_notification(alert: Alert, context: Dict[str, Any]) → bool[source]

Send notification for alert.

Parameters:

alert (Alert) – Alert to send notification for
context (Dict[str, Any]) – Additional context information

Returns:

True if notification was sent successfully

Return type:

bool

class kailash.monitoring.alerts.LogNotificationChannel(log_level: str = 'ERROR')[source]

Bases: NotificationChannel

Log-based notification channel.

Parameters:: log_level (str)

__init__(log_level: str = 'ERROR')[source]

Initialize log notification channel.

Parameters:: log_level (str) – Log level for notifications

send_notification(alert: Alert, context: Dict[str, Any]) → bool[source]

Send notification via logging.

Parameters:

alert (Alert)
context (Dict[str, Any])

Return type:

bool

class kailash.monitoring.alerts.EmailNotificationChannel(smtp_host: str, smtp_port: int, username: str, password: str, from_email: str, to_emails: List[str], use_tls: bool = True)[source]

Bases: NotificationChannel

Email notification channel.

Parameters:

smtp_host (str)
smtp_port (int)
username (str)
password (str)
from_email (str)
to_emails (List[str])
use_tls (bool)

__init__(smtp_host: str, smtp_port: int, username: str, password: str, from_email: str, to_emails: List[str], use_tls: bool = True)[source]

Initialize email notification channel.

Parameters:

smtp_host (str) – SMTP server host
smtp_port (int) – SMTP server port
username (str) – SMTP username
password (str) – SMTP password
from_email (str) – From email address
to_emails (List[str]) – List of recipient email addresses
use_tls (bool) – Whether to use TLS

send_notification(alert: Alert, context: Dict[str, Any]) → bool[source]

Send notification via email.

Parameters:

alert (Alert)
context (Dict[str, Any])

Return type:

bool

class kailash.monitoring.alerts.SlackNotificationChannel(webhook_url: str, channel: str = '#alerts')[source]

Bases: NotificationChannel

Slack notification channel.

Parameters:

webhook_url (str)
channel (str)

__init__(webhook_url: str, channel: str = '#alerts')[source]

Initialize Slack notification channel.

Parameters:

webhook_url (str) – Slack webhook URL
channel (str) – Slack channel to send alerts to

send_notification(alert: Alert, context: Dict[str, Any]) → bool[source]

Send notification via Slack.

Parameters:

alert (Alert)
context (Dict[str, Any])

Return type:

bool

class kailash.monitoring.alerts.WebhookNotificationChannel(webhook_url: str, headers: Dict[str, str] | None = None)[source]

Bases: NotificationChannel

Generic webhook notification channel.

Parameters:

webhook_url (str)
headers (Dict[str, str] | None)

__init__(webhook_url: str, headers: Dict[str, str] | None = None)[source]

Initialize webhook notification channel.

Parameters:

webhook_url (str) – Webhook URL
headers (Dict[str, str] | None) – Optional HTTP headers

send_notification(alert: Alert, context: Dict[str, Any]) → bool[source]

Send notification via webhook.

Parameters:

alert (Alert)
context (Dict[str, Any])

Return type:

bool

class kailash.monitoring.alerts.AlertManager(metrics_registry: MetricsRegistry)[source]

Bases: object

Alert manager for handling alerting rules and notifications.

Parameters:: metrics_registry (MetricsRegistry)

__init__(metrics_registry: MetricsRegistry)[source]

Initialize alert manager.

Parameters:: metrics_registry (MetricsRegistry) – Metrics registry to monitor

rules: Dict[str, AlertRule]

alerts: Dict[str, Alert]

notification_channels: List[NotificationChannel]

add_rule(rule: AlertRule)[source]

Add an alerting rule.

Parameters:: rule (AlertRule) – AlertRule to add

remove_rule(rule_name: str)[source]

Remove an alerting rule.

Parameters:: rule_name (str) – Name of rule to remove

add_notification_channel(channel: NotificationChannel)[source]

Add a notification channel.

Parameters:: channel (NotificationChannel) – NotificationChannel to add

start()[source]: Start the alert manager.

stop()[source]: Stop the alert manager.

get_active_alerts() → List[Alert][source]

Get all active (firing) alerts.

Return type:: List[Alert]

get_all_alerts() → List[Alert][source]

Get all alerts.

Return type:: List[Alert]

silence_alert(alert_id: str)[source]

Silence an alert.

Parameters:: alert_id (str) – Alert ID to silence

acknowledge_alert(alert_id: str)[source]

Acknowledge an alert (same as silence for now).

Parameters:: alert_id (str) – Alert ID to acknowledge

kailash.monitoring.alerts.create_default_alert_rules() → List[AlertRule][source]

Create default alert rules for common scenarios.

Return type:: List[AlertRule]

Usage Examples

Basic Monitoring Setup

from kailash.monitoring.metrics import get_validation_metrics, get_security_metrics
from kailash.monitoring.alerts import AlertManager, AlertRule, AlertSeverity
from kailash.monitoring.alerts import LogNotificationChannel

# Set up comprehensive monitoring
validation_metrics = get_validation_metrics()
security_metrics = get_security_metrics()
registry = get_metrics_registry()
alert_manager = AlertManager(registry)

# Configure alert rules
alert_manager.add_rule(AlertRule(
    name="high_validation_failures",
    description="Validation failure rate above 10%",
    severity=AlertSeverity.ERROR,
    metric_name="validation_failure",
    condition="> 5",
    threshold=5
))

alert_manager.add_notification_channel(LogNotificationChannel())
alert_manager.start()

Custom Metrics Collection

from kailash.monitoring.metrics import MetricsCollector, MetricType

# Create custom metrics collector
collector = MetricsCollector()

# Create and record metrics
response_time = collector.create_metric(
    "api_response_time",
    MetricType.TIMER,
    "API response time",
    "milliseconds"
)

collector.record_timer("api_response_time", 150.5)
collector.increment("api_requests")
collector.set_gauge("active_connections", 42)

Security Monitoring

from kailash.monitoring.metrics import get_security_metrics, MetricSeverity

security_metrics = get_security_metrics()

# Record security violations
security_metrics.record_security_violation(
    violation_type="sql_injection_attempt",
    severity=MetricSeverity.HIGH,
    source="workflow_connection",
    details={"query": "malicious_query"}
)

# Check critical violations
critical_count = security_metrics.get_critical_violations()
violation_rate = security_metrics.get_violation_rate()

Performance Monitoring

from kailash.monitoring.metrics import get_performance_metrics

performance_metrics = get_performance_metrics()

# Record operation performance
performance_metrics.record_operation(
    operation="workflow_execution",
    duration_ms=1250.0,
    success=True
)

# Update system metrics
performance_metrics.update_system_metrics(
    memory_mb=512.0,
    cpu_percent=25.5,
    rps=100.0
)

# Get performance statistics
p95_time = performance_metrics.get_p95_response_time()

Connection Validation Monitoring

from kailash.runtime.local import LocalRuntime
from kailash.monitoring.metrics import get_validation_metrics

# Enable connection validation with monitoring
runtime = LocalRuntime(connection_validation="strict")
validation_metrics = get_validation_metrics()

# Execute workflow with monitoring
results, run_id = runtime.execute(workflow.build())

# View validation metrics
success_rate = validation_metrics.get_success_rate()
cache_hit_rate = validation_metrics.get_cache_hit_rate()

print(f"Validation success rate: {success_rate:.2%}")
print(f"Cache hit rate: {cache_hit_rate:.2%}")

Metrics Export

from kailash.monitoring.metrics import get_metrics_registry

registry = get_metrics_registry()

# Export metrics in JSON format
json_metrics = registry.export_metrics("json")

# Export metrics in Prometheus format
prometheus_metrics = registry.export_metrics("prometheus")

print(json_metrics)