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

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 with the GenerateQuery method, which transforms a rule's high-level definitions into executable Elasticsearch DSL.

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.

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:

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:

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).

Action Execution and Notification Delivery

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

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, 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.

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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}