mlrun.common.schemas.alert#

class mlrun.common.schemas.alert.AlertActivation(*, id: int, name: str, project: str, severity: AlertSeverity, activation_time: datetime, entity_id: str, entity_kind: EventEntityKind, criteria: AlertCriteria, event_kind: EventKind, number_of_events: int, notifications: list[mlrun.common.schemas.notification.NotificationState], reset_time: datetime | None = None)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

activation_time: datetime#
criteria: AlertCriteria#
entity_id: str#
entity_kind: EventEntityKind#
event_kind: EventKind#
group_key(attributes: list[str]) Any | tuple[source]#

Dynamically create a key for grouping based on the provided attributes. - If there's only one attribute, return the value directly (not a single-element tuple). - If there are multiple attributes, return them as a tuple for grouping.

This ensures grouping behaves intuitively without redundant tuple representations.

id: int#
name: str#
notifications: list[mlrun.common.schemas.notification.NotificationState]#
number_of_events: int#
project: str#
reset_time: datetime | None#
severity: AlertSeverity#
class mlrun.common.schemas.alert.AlertActivations(*, activations: list[mlrun.common.schemas.alert.AlertActivation], pagination: dict | None = None)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

activations: list[mlrun.common.schemas.alert.AlertActivation]#
aggregate_by(group_by_attrs: list[str], aggregation_function: Callable[[list[mlrun.common.schemas.alert.AlertActivation]], Any]) dict[source]#

Aggregate alert activations by specified attributes using a given aggregation function.

Args: :param group_by_attrs: Attributes to group by. :param aggregation_function: Function to aggregate grouped activations.

Returns:

A dictionary where keys are tuples of attribute values and values are the result of the aggregation function.

Example

# Aggregate by name and entity_id and count number of activations in each group activations.aggregate_by(["name", "entity_id"], lambda activations: len(activations))

group_by(*attributes: str) dict[source]#

Group alert activations by specified attributes.

Args: :param attributes: Attributes to group by.

Returns:

A dictionary where keys are tuples of attribute values and values are lists of AlertActivation objects.

Example

# Group by project and severity grouped = activations.group_by("project", "severity")

pagination: dict | None#
class mlrun.common.schemas.alert.AlertActiveState(value)[source]#

Bases: StrEnum

An enumeration.

ACTIVE = 'active'#
INACTIVE = 'inactive'#
class mlrun.common.schemas.alert.AlertConfig(*, project: str, id: int = None, name: str, description: str | None = '', summary: str, created: str | datetime = None, severity: AlertSeverity, entities: EventEntities, trigger: AlertTrigger, criteria: AlertCriteria | None = None, reset_policy: ResetPolicy = ResetPolicy.AUTO, notifications: ConstrainedListValue[AlertNotification], state: AlertActiveState = AlertActiveState.INACTIVE, count: int | None = 0, updated: datetime = None, **extra_data: Any)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

class Config[source]#

Bases: object

extra = 'allow'#
count: int | None#
created: str | datetime#
criteria: AlertCriteria | None#
description: str | None#
entities: EventEntities#
get_raw_notifications() list[mlrun.common.schemas.notification.Notification][source]#
id: int#
name: str#
notifications: ConstrainedListValue[AlertNotification]#
project: str#
reset_policy: ResetPolicy#
severity: AlertSeverity#
state: AlertActiveState#
summary: str#
trigger: AlertTrigger#
updated: datetime#
class mlrun.common.schemas.alert.AlertCriteria(*, count: int = 1, period: str = None)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

count: int#
period: str#
class mlrun.common.schemas.alert.AlertNotification(*, notification: Notification, cooldown_period: str = None)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

cooldown_period: str#
notification: Notification#
class mlrun.common.schemas.alert.AlertSeverity(value)[source]#

Bases: StrEnum

An enumeration.

HIGH = 'high'#
LOW = 'low'#
MEDIUM = 'medium'#
class mlrun.common.schemas.alert.AlertTemplate(*, template_id: int = None, template_name: str, template_description: str | None = 'String explaining the purpose of this template', system_generated: bool = False, summary: str | None = "String to be sent in the generated notifications e.g. 'Model {{project}}/{{entity}} is drifting.'See AlertConfig.summary description", severity: AlertSeverity, trigger: AlertTrigger, criteria: AlertCriteria | None = None, reset_policy: ResetPolicy = ResetPolicy.AUTO)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

criteria: AlertCriteria | None#
reset_policy: ResetPolicy#
severity: AlertSeverity#
summary: str | None#
system_generated: bool#
template_description: str | None#
template_id: int#
template_name: str#
templates_differ(other)[source]#
trigger: AlertTrigger#
class mlrun.common.schemas.alert.AlertTrigger(*, events: list[mlrun.common.schemas.alert.EventKind] = [], prometheus_alert: str = None)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

events: list[mlrun.common.schemas.alert.EventKind]#
prometheus_alert: str#
class mlrun.common.schemas.alert.AlertsModes(value)[source]#

Bases: StrEnum

An enumeration.

disabled = 'disabled'#
enabled = 'enabled'#
class mlrun.common.schemas.alert.Event(*, kind: EventKind, timestamp: str | datetime = None, entity: EventEntities, value_dict: dict | None = None)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

entity: EventEntities#
is_valid()[source]#
kind: EventKind#
timestamp: str | datetime#
value_dict: dict | None#
class mlrun.common.schemas.alert.EventEntities(*, kind: EventEntityKind, project: str, ids: ConstrainedListValue[str])[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

ids: ConstrainedListValue[str]#
kind: EventEntityKind#
project: str#
class mlrun.common.schemas.alert.EventEntityKind(value)[source]#

Bases: StrEnum

An enumeration.

JOB = 'job'#
MODEL_ENDPOINT_RESULT = 'model-endpoint-result'#
MODEL_MONITORING_APPLICATION = 'model-monitoring-application'#
class mlrun.common.schemas.alert.EventKind(value)[source]#

Bases: StrEnum

An enumeration.

CONCEPT_DRIFT_DETECTED = 'concept-drift-detected'#
CONCEPT_DRIFT_SUSPECTED = 'concept-drift-suspected'#
DATA_DRIFT_DETECTED = 'data-drift-detected'#
DATA_DRIFT_SUSPECTED = 'data-drift-suspected'#
FAILED = 'failed'#
MM_APP_ANOMALY_DETECTED = 'mm-app-anomaly-detected'#
MM_APP_ANOMALY_SUSPECTED = 'mm-app-anomaly-suspected'#
MM_APP_FAILED = 'mm-app-failed'#
MODEL_PERFORMANCE_DETECTED = 'model-performance-detected'#
MODEL_PERFORMANCE_SUSPECTED = 'model-performance-suspected'#
SYSTEM_PERFORMANCE_DETECTED = 'system-performance-detected'#
SYSTEM_PERFORMANCE_SUSPECTED = 'system-performance-suspected'#
class mlrun.common.schemas.alert.ResetPolicy(value)[source]#

Bases: StrEnum

An enumeration.

AUTO = 'auto'#
MANUAL = 'manual'#