如何识别日志文件中的序列数据并在Kibana中统计展示？

Great question! You absolutely can pull off this sequence detection and counting using either Logstash filters or Kibana-native tools—let’s walk through both options to fit your workflow.

If you want to bake sequence detection directly into your pipeline (so Kibana can just query pre-processed data), Logstash’s aggregate filter is your go-to tool. It lets you track events across a shared identifier (like a session ID or user ID) and flag when your target sequence is matched.

Here’s a step-by-step breakdown with a sample config:

• First, extract key fields from your logs (like event_type and session_id) using grok or dissect—this gives you structured data to work with.
• Use the aggregate filter to build a timeline of events per session. When the exact sequence you’re looking for appears, add a custom field (e.g., detected_sequence) to mark it.
• Set a timeout to clean up old sessions so you don’t bloat memory.

Sample Logstash filter config:

filter {
  # Extract core fields from raw logs (adjust grok pattern to match your log format)
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{WORD:event_type} session:%{UUID:session_id}" }
  }

  # Track events per session and detect target sequences
  aggregate {
    task_id => "%{session_id}" # Group events by unique session ID
    code => "
      # Initialize an array to store events for the session
      map['events'] ||= []
      # Add the current event type to the session's event list
      map['events'] << event.get('event_type')
      
      # Define your target sequence here
      target_sequence = ['user_login', 'product_view', 'purchase']
      
      # Check if the end of the event list matches the target sequence
      if map['events'].last(target_sequence.length) == target_sequence
        event.set('detected_sequence', 'user_purchase_flow')
      end
    "
    timeout => 1800 # Clean up sessions that haven't had activity in 30 minutes
  }
}

Once processed, events that complete your sequence will have the detected_sequence field. You can then use this field in Kibana to easily count occurrences (via a Metric visualization) or filter for matching sessions.

If you prefer not to modify your Logstash pipeline, you can detect sequences directly in Kibana using Elasticsearch’s span_sequence query (supported in Kibana’s Discover, Visualize, and Dev Tools).

Example span_sequence Query #

This query will find all instances where your target sequence occurs in order, within a specified time window:

{
  "query": {
    "span_sequence": {
      "clauses": [
        { "span_term": { "event_type": "user_login" } },
        { "span_term": { "event_type": "product_view" } },
        { "span_term": { "event_type": "purchase" } }
      ],
      "within": { "size": 60, "unit": "m" } # Limit sequence to 60 minutes total
    }
  }
}

Counting Occurrences in Kibana #

1. Head to Visualize Library and create a new Metric visualization.
2. In the query bar, paste the span_sequence query (switch to "JSON" mode for the query).
3. The metric will show the total number of matching sequences. You can also add filters (like time ranges) or split the count by fields (e.g., session_id) to see individual instances.

Key Notes #

• Logstash pre-processing is better for high-volume logs (it reduces query load on Elasticsearch).
• Kibana queries are more flexible if you need to adjust sequences on the fly without reprocessing logs.
• Always use a unique identifier (like session_id) to ensure you’re tracking sequences for the same user/session—otherwise, you might false-positive on unrelated events.

内容的提问来源于stack exchange，提问作者Mangoski