# How INFINI Console's Alerting Engine Evaluates Conditions and Executes Actions Across Elasticsearch Clusters

> Discover how INFINI Console's alerting engine evaluates Elasticsearch conditions and executes actions across clusters. Learn about DSL queries, sliding-window counters, and notification policies.

- Repository: [INFINI Labs/console](https://github.com/infinilabs/console)
- Tags: internals
- Published: 2026-03-04

---

**INFINI Console's alerting engine uses a stateless `Engine` type to generate Elasticsearch DSL queries, retrieve metrics from cluster-specific clients, evaluate rule conditions using sliding-window counters, and execute notification channels while respecting throttling and escalation policies.**

The alerting engine in the `infinilabs/console` repository provides continuous monitoring capabilities across distributed Elasticsearch environments. It processes alert rules through a four-stage pipeline—query generation, data retrieval, condition evaluation, and action execution—enabling operators to track metrics and receive notifications from any number of clusters through a unified interface.

## Query Generation and DSL Construction

The evaluation cycle begins in [`service/alerting/elasticsearch/engine.go`](https://github.com/infinilabs/console/blob/main/service/alerting/elasticsearch/engine.go) with the `GenerateQuery` method, which transforms a rule's high-level definitions into executable Elasticsearch DSL.

```go
func (engine *Engine) GenerateQuery(rule *alerting.Rule, filterParam *alerting.FilterParam) (interface{}, error) {
    filter, err := engine.GenerateRawFilter(rule, filterParam)
    timeFilter, err := engine.generateTimeFilter(rule, filterParam)
    rule.Metrics.Filter = filter
    rule.Metrics.TimeFilter = timeFilter
    return insight.GenerateQuery(&rule.Metrics.Metric)
}

```

The engine handles two filter types:

- **Raw filters** – When `rule.Resource.RawFilter` is present, the engine uses the Elasticsearch query verbatim; otherwise, it recursively converts structured `FilterQuery` objects into DSL via `ConvertFilterQueryToDsl`.
- **Time filters** – The window is derived dynamically from the rule's `bucket_size` or from an explicit `filterParam` used during ad-hoc tests.

The `insight` package then assembles the final query, injecting `date_histogram`, `terms`, and statistical aggregations required by `rule.Metrics.Items`.

## Cross-Cluster Data Retrieval

Multi-cluster support is implemented through client isolation. The `ExecuteQuery` method retrieves a dedicated Elasticsearch client using the cluster identifier stored in the rule's resource configuration.

```go
func (engine *Engine) ExecuteQuery(rule *alerting.Rule, filterParam *alerting.FilterParam) (*alerting.QueryResult, error) {
    esClient := elastic.GetClient(rule.Resource.ID)
    // ...
    searchRes, err := esClient.SearchWithRawQueryDSL(indexName, queryDslBytes)
    // ...
}

```

- **`rule.Resource.ID`** specifies the target cluster, with a separate `elastic.Client` instance maintained for each Elasticsearch endpoint configured in the console.
- **`rule.Resource.Objects`** defines the comma-separated index names to query within that cluster.

Because the engine receives a full client instance per rule execution, the same process can evaluate rules targeting **different clusters** concurrently without state conflicts.

## Condition Evaluation Logic

Once data is retrieved, the engine evaluates conditions through a multi-step process orchestrated by `CheckCondition` and `GetTargetMetricData`.

### Metric Formula Resolution

The engine first generates an expression representing the rule's metric calculation:

```go
metricExpression, _ := rule.Metrics.GenerateExpression()
for i, cond := range rule.Conditions.Items {
    expression, _ := cond.GenerateConditionExpression()
    rule.Conditions.Items[i].Expression = strings.ReplaceAll(expression, "result", metricExpression)
}

```

Each condition's expression (e.g., `"result > 80"`) may reference the placeholder `result`, which the engine substitutes with the actual metric formula (e.g., `cpu_usage / cpu_total * 100`).

### Per-Bucket Evaluation with Sliding Windows

The engine evaluates conditions against each metric bucket using the `govaluate` library:

```go
targetMetricData, queryResult, err := engine.GetTargetMetricData(rule, true, nil)
// ...
for _, cond := range rule.Conditions.Items {
    evaluateResult, _ := expression.Evaluate(map[string]interface{}{"result": valueExpressionResult})
    // Track consecutive matches according to MinimumPeriodMatch
}

```

Conditions are sorted by priority (`alerting.PriorityWeights`). A **sliding-window counter** (`triggerCount`) enforces `MinimumPeriodMatch`, ensuring alerts only fire after a condition persists for the configured number of consecutive evaluation cycles.

### Bucket-Diff Conditions

For rules defining `BucketConditions`, the engine collapses raw bucket counts into a time-ordered map and evaluates conditions based on **absolute count changes** or **content-change flags** (`ContentChangeState`). This logic resides in `CheckBucketCondition` (approximately lines 600–740 of [`engine.go`](https://github.com/infinilabs/console/blob/main/engine.go)).

## Action Execution and Notification Delivery

After condition evaluation completes, the `Do` method orchestrates notification delivery and state persistence.

```go
func (engine *Engine) Do(rule *alerting.Rule) error {
    alertItem := &alerting.Alert{... State: alerting.AlertStateOK}
    checkResults, err := engine.CheckCondition(rule)
    // ...
    attachTitleMessageToCtx(title, message, paramsCtx)
    actionResults, _ := performChannels(notifyCfg.Normal, paramsCtx, false)
}

```

The execution flow includes:

- **Channel resolution** – `common.RetrieveChannel` loads concrete implementations (SMTP, DingTalk, webhook) from the database.
- **Template rendering** – Notification titles and messages are processed through `ResolveMessage` in [`service/alerting/common/template.go`](https://github.com/infinilabs/console/blob/main/service/alerting/common/template.go), substituting context variables.
- **Throttling** – The engine respects `ThrottlePeriod` and `EscalationThrottlePeriod` by checking a KV store (`kv.AddValue`) before executing channels.
- **Escalation** – If escalation policies are enabled, a second channel execution pass occurs after the configured interval.

All action results are recorded in the `Alert` document (`ActionExecutionResults`, `RecoverActionResults`, `EscalationActionResults`) and persisted to Elasticsearch via `orm.Save`.

## Stateless Architecture for Concurrent Multi-Cluster Monitoring

The `Engine` type is designed to be **stateless**—a new instance is created for each rule execution. This architecture ensures that the scheduler can safely run thousands of rules across hundreds of clusters concurrently without interference. Each task maintains its own `rule.Resource.ID`, guaranteeing that queries and actions are isolated to their intended Elasticsearch endpoints.

```go
engine := elasticsearch.Engine{}
task := engine.GenerateTask(rule)
scheduler.Add(task, rule.Schedule)

```

## Summary

- **Cluster Isolation** – Rules target specific clusters via `rule.Resource.ID`, with `elastic.GetClient` providing dedicated client instances for each endpoint.
- **Dynamic Query Building** – The `GenerateQuery` method combines raw filters, time windows, and metric aggregations into executable Elasticsearch DSL.
- **Conditional Logic** – The engine evaluates metric formulas using `govaluate`, applies sliding-window counters for `MinimumPeriodMatch`, and handles complex bucket-diff comparisons.
- **Reliable Delivery** – Actions execute through resolved channels with built-in throttling, escalation policies, and persistent audit trails stored in Elasticsearch documents.

## Frequently Asked Questions

### How does INFINI Console route alert queries to different Elasticsearch clusters?

The engine uses `elastic.GetClient(rule.Resource.ID)` in [`service/alerting/elasticsearch/engine.go`](https://github.com/infinilabs/console/blob/main/service/alerting/elasticsearch/engine.go) to retrieve a pre-configured client instance for the cluster specified in the rule's `resource.id` field. This ensures each rule executes against its designated endpoint, allowing a single console instance to monitor multiple clusters simultaneously.

### What expression engine evaluates alert conditions in INFINI Console?

The alerting engine uses the `govaluate` library to evaluate condition expressions. It substitutes the placeholder `result` with the rule's metric formula (e.g., `cpu / total * 100`) and evaluates the resulting expression against each metric bucket's calculated value.

### How does the alerting engine prevent notification spam?

The engine implements throttling through `ThrottlePeriod` and `EscalationThrottlePeriod` configurations. Before executing channels, it checks a KV store to verify that the required quiet period has elapsed since the last notification, suppressing duplicate alerts for the same rule.

### Can INFINI Console alert on changes between consecutive metric buckets?

Yes. Through `BucketConditions` and the `CheckBucketCondition` method, the engine can evaluate alerts based on absolute count changes or content-change flags between buckets. This enables detection of sudden spikes, drops, or content modifications in time-series data.