drain3

package module
v2.0.0-...-7b555aa Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 23, 2026 License: MIT Imports: 21 Imported by: 0

README

drain3

A Go implementation of the Drain3 log template mining library.

Drain3 processes streaming log messages and automatically extracts log templates by replacing variable parts with wildcards (<*>). It is based on the Drain algorithm for online log parsing.

Installation

go get github.com/FEINIAO233/drain3/v2

Quick Start

package main

import (
    "fmt"
    "log"

    "github.com/FEINIAO233/drain3/v2"
)

func main() {
    tm, err := drain3.New()
    if err != nil {
        log.Fatal(err)
    }

    messages := []string{
        "Failed password for user admin from 192.168.1.1 port 22",
        "Failed password for user root from 10.0.0.1 port 22",
        "Accepted password for user admin from 192.168.1.1 port 22",
        "Accepted password for user admin from 10.0.0.1 port 22",
    }

    for _, msg := range messages {
        result := tm.AddLogMessage(msg)
        if result.InputError != nil {
            log.Printf("drain3 rejected input: %v", result.InputError)
            continue
        }
        if result.MaskError != nil {
            log.Printf("drain3 masking failed: %v", result.MaskError)
            continue
        }
        if result.Cluster == nil {
            log.Print("drain3 returned no cluster")
            continue
        }
        if result.SaveError != nil {
            // Mining succeeded, but the state snapshot was not persisted.
            log.Printf("drain3 state save failed: %v", result.SaveError)
        }
        fmt.Printf("[%s] %s\n", result.ChangeType, result.Cluster.GetTemplate())
    }

    fmt.Printf("\nDiscovered %d templates:\n", tm.ClusterCount())
    for _, c := range tm.Clusters() {
        fmt.Printf("  [size=%d] %s\n", c.Size, c.GetTemplate())
    }
}

Features

  • Online log parsing — processes log messages one at a time in a streaming fashion
  • Template extraction — automatically discovers log templates by replacing variable tokens with <*>
  • Masking — pre-process log messages with regex-based masking (IP addresses, numbers, etc.) before template mining
  • Persistence — save and restore state to/from files or custom backends (with optional zlib compression)
  • Thread-safe — safe for concurrent use from multiple goroutines
  • Configurable — functional options API or YAML config, with sensible defaults
  • Parameter extraction — extract variable values from log messages given a discovered template

Configuration

Pass options directly when creating a TemplateMiner:

tm, err := drain3.New(
    drain3.WithSimTh(0.5),
    drain3.WithDepth(5),
    drain3.WithMaxClusters(1000),
    drain3.WithExtraDelimiters("=", ":"),
    drain3.WithMasking(`\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}`, "IP"),
    drain3.WithMasking(`\b\d+\b`, "NUM"),
    drain3.WithFilePersistence("/tmp/drain3_state.json"),
)

Available options:

Option Default Description
WithSimTh(float64) 0.4 Similarity threshold (0.0–1.0)
WithDepth(int) 4 Prefix tree depth
WithMaxChildren(int) 100 Max children per tree node
WithMaxClusters(int) 10000 Max clusters (0 explicitly selects unlimited)
WithExtraDelimiters(string...) none Additional characters to split tokens on
WithParamStr(string) <*> Wildcard placeholder string
WithParametrizeNumericTokens(bool) true Route numeric tokens to the wildcard node
WithMasking(pattern, name) none Add a regex masking rule (can be called multiple times)
WithMaskPrefix(string) < Prefix for mask tokens
WithMaskSuffix(string) > Suffix for mask tokens
WithRegexTimeout(time.Duration) 100ms Maximum execution time per masking regex (0 disables it)
WithMaskErrorPolicy(policy) reject Reject masking failures or continue with partial masking
WithMaxMessageBytes(int) 65536 Maximum input message size (0 disables)
WithMaxStateBytes(int) 67108864 Maximum persisted state size (0 disables)
WithMaxDecompressedStateBytes(int) 268435456 Maximum decoded state size (0 disables)
WithParameterExtractionCacheCapacity(int) 3000 Compiled extraction regex LRU size (0 disables caching)
WithPersistence(PersistenceHandler) nil Custom persistence backend
WithFilePersistence(path) none File-based persistence
WithSnapshotInterval(minutes) 5 Auto-save interval
WithCompressState(bool) true Zlib compression for persisted state
WithProfiling(bool) false Enable built-in profiler
YAML Config (alternative)
drain:
  sim_th: 0.5
  depth: 5
  max_children: 100
  max_clusters: 1000
  extra_delimiters: ["=", ":"]
  parametrize_numeric_tokens: true

snapshot:
  snapshot_interval_minutes: 5
  compress_state: true

masking:
  mask_prefix: "<"
  mask_suffix: ">"
  regex_timeout_millis: 100
  error_policy: reject
  instructions:
    - pattern: '\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'
      mask_with: "IP"
    - pattern: '\b\d+\b'
      mask_with: "NUM"

limits:
  max_message_bytes: 65536
  max_state_bytes: 67108864
  max_decompressed_state_bytes: 268435456
  parameter_extraction_cache_capacity: 3000

profiling:
  enabled: false

cfg, _ := drain3.LoadConfig("drain3.yaml")
tm, _ := drain3.New(drain3.WithConfig(cfg))

API

TemplateMiner

The high-level API that integrates all components.

// Create with defaults
tm, _ := drain3.New()

// Create with options
tm, _ := drain3.New(
    drain3.WithSimTh(0.5),
    drain3.WithMasking(`\d+`, "NUM"),
)

// Process a log message
result := tm.AddLogMessage("user alice logged in from 10.0.0.1")
// result.Cluster    - the matched/created LogCluster
// result.ChangeType - ChangeNone, ChangeClusterCreated, or ChangeClusterTemplateChanged
// result.InputError - input rejected by a resource limit, if any
// result.MaskError  - regex timeout/execution error, if any
// result.SaveError  - persistence error, if any

// Match without modifying state
cluster := tm.Match("user bob logged in from 10.0.0.2", drain3.SearchStrategyNever)
// Use MatchWithError when masking timeout/error details are required.

// Save/restore state
tm.SaveState()
tm.LoadState()

Persisted state is validated before it is applied. Loading corrupted or incompatible state returns an error and leaves the original persistence data untouched. Snapshot encoding, compression, and persistence I/O run without holding the Drain matching lock. Snapshots contain a schema version and configuration fingerprint; the prefix tree is rebuilt from validated clusters using the current configuration when state is loaded.

This module is the v2 API and uses the import path github.com/FEINIAO233/drain3/v2. When upgrading an existing collector, use a new persistence key or state location such as drain3-v2; do not mix v1 and v2 miner state or cluster IDs. Treat InputError and MaskError as rejected messages under the default policy, and always check that Cluster is non-nil before using it. SaveError means mining succeeded but the resulting state snapshot was not persisted.

Drain

The core algorithm engine, usable standalone without masking or persistence.

cfg := drain3.DefaultDrainConfig()
cfg.SimTh = 0.5
d := drain3.NewDrain(cfg)

cluster, changeType := d.AddLogMessage("hello world")
matched := d.Match("hello world", drain3.SearchStrategyNever)

NewDrain panics on invalid programmatic configuration. Use NewDrainChecked when configuration values come from users or another dynamic source.

Parameter Extraction

Extract variable values from messages using discovered templates.

pe := tm.NewParameterExtractor()
params := pe.ExtractParameters("user <*> logged in", "user alice logged in", false)
// params[0].Value == "alice", params[0].MaskName == "*"
Search Strategies
Strategy Behavior
SearchStrategyNever Tree search only; fastest
SearchStrategyFallback Tree search first, then full scan if no match
SearchStrategyAlways Full scan of all clusters; most thorough

Match is inference-only and requires a perfect template match. SimTh controls training and does not relax inference matching.

Extra delimiters are literal strings, not regular expressions. For example, WithExtraDelimiters(".") splits dotted addresses at literal dots.

The safe default retains at most 10,000 clusters. Use WithMaxClusters(0) only when an unlimited model is intentional and externally monitored.

Public mining methods are safe for concurrent use. Runtime configuration and exported struct fields must be treated as read-only after construction.

Dependencies

Acknowledgements

This is a Go port of the Python Drain3 library by IBM Research, which implements the Drain algorithm from:

Pinjia He, Jieming Zhu, Zibin Zheng, and Michael R. Lyu. Drain: An Online Log Parsing Approach with Fixed Depth Tree, Proceedings of the IEEE International Conference on Web Services (ICWS), 2017.

License

MIT

Documentation

Overview

Package drain3 implements the Drain log template mining algorithm in Go. It is a faithful port of the Python Drain3 library (github.com/logpai/Drain3).

Drain3 processes streaming log messages and automatically extracts log templates by replacing variable parts with wildcards (<*>).

Index

Examples

Constants

View Source 
const DefaultParamStr = "<*>"

DefaultParamStr is the wildcard placeholder used for variable tokens in templates.

Variables

This section is empty.

Functions

This section is empty.

Types

type AddLogMessageResult 

type AddLogMessageResult struct {
	Cluster    *LogCluster
	ChangeType ChangeType
	// InputError reports that the message was rejected before mining, for
	// example because it exceeded the configured size limit.
	InputError error
	// MaskError reports masking failures such as regex timeouts. The message is
	// mined only when MaskErrorPolicyContinue is configured.
	MaskError error
	// SaveError is non-nil if auto-saving state to persistence failed.
	SaveError error
}

AddLogMessageResult is the result of processing a log message.

type ChangeType 

type ChangeType int

ChangeType indicates what happened when a log message was processed. 

const (
	// ChangeNone means the message matched an existing cluster with no template change.
	ChangeNone ChangeType = iota
	// ChangeClusterCreated means a new cluster was created for the message.
	ChangeClusterCreated
	// ChangeClusterTemplateChanged means an existing cluster's template was updated.
	ChangeClusterTemplateChanged
)

func (ChangeType) String 

func (c ChangeType) String() string

type ClusterStore 

type ClusterStore interface {
	Get(clusterID int) *LogCluster
	// Put inserts or updates a cluster. Returns the evicted cluster if any (LRU mode).
	Put(cluster *LogCluster) *LogCluster
	Remove(clusterID int)
	Len() int
	Values() []*LogCluster
}

ClusterStore is an interface for storing and retrieving log clusters.

type Config 

type Config struct {
	Drain     DrainSection     `yaml:"drain"`
	Snapshot  SnapshotSection  `yaml:"snapshot"`
	Masking   MaskingSection   `yaml:"masking"`
	Limits    LimitsSection    `yaml:"limits"`
	Profiling ProfilingSection `yaml:"profiling"`
}

Config holds all configuration for the TemplateMiner.

func DefaultConfig 

func DefaultConfig() *Config

DefaultConfig returns a Config with safe production defaults.

func LoadConfig 

func LoadConfig(filename string) (*Config, error)

LoadConfig loads a Config from a YAML file. Missing fields are filled with default values.

func (*Config) Clone 

func (c *Config) Clone() *Config

Clone returns a deep copy suitable for use as immutable runtime configuration.

func (*Config) Validate 

func (c *Config) Validate() error

Validate checks configuration values that would otherwise produce invalid matching behavior or unbounded internal operations.

type Drain 

type Drain struct {

	// Configuration
	SimTh                    float64  `json:"-"`
	Depth                    int      `json:"-"`
	MaxChildren              int      `json:"-"`
	MaxClusters              int      `json:"-"`
	ExtraDelimiters          []string `json:"-"`
	ParamStr                 string   `json:"-"`
	ParametrizeNumericTokens bool     `json:"-"`

	// State
	RootNode        *Node        `json:"root_node"`
	IDToCluster     ClusterStore `json:"-"`
	ClustersCounter int          `json:"clusters_counter"`
	// contains filtered or unexported fields
}

Drain is the core log template mining engine. It maintains a prefix tree for fast matching and a collection of log clusters.

func NewDrain 

func NewDrain(cfg DrainConfig) *Drain

NewDrain creates a new Drain engine with the given configuration. It panics for invalid configuration. Use NewDrainChecked when configuration is supplied dynamically and the error must be handled.

func NewDrainChecked 

func NewDrainChecked(cfg DrainConfig) (*Drain, error)

NewDrainChecked creates a new Drain engine and returns configuration errors.

func (*Drain) AddLogMessage 

func (d *Drain) AddLogMessage(content string) (*LogCluster, ChangeType)

AddLogMessage processes a log message and returns the matching cluster and change type.

func (*Drain) ClusterCount 

func (d *Drain) ClusterCount() int

ClusterCount returns the number of active clusters.

func (*Drain) Clusters 

func (d *Drain) Clusters() []*LogCluster

Clusters returns all current log clusters.

func (*Drain) GetContentAsTokens 

func (d *Drain) GetContentAsTokens(content string) []string

GetContentAsTokens splits content into tokens, applying extra delimiters. Extra delimiters are treated as literal strings, matching Drain3 tokenization.

func (*Drain) GetTotalClusterSize 

func (d *Drain) GetTotalClusterSize() int

GetTotalClusterSize returns the sum of all cluster sizes (total messages processed).

func (*Drain) MarshalJSON 

func (d *Drain) MarshalJSON() ([]byte, error)

MarshalJSON serializes the Drain state to JSON.

func (*Drain) Match 

func (d *Drain) Match(content string, fullSearchStrategy SearchStrategy) *LogCluster

Match finds the best matching cluster for a log message without modifying any state.

func (*Drain) StateConfigMismatch 

func (d *Drain) StateConfigMismatch() bool

StateConfigMismatch reports whether the most recently loaded state was created with a different or legacy configuration. The prefix tree is rebuilt automatically in either case.

func (*Drain) UnmarshalState 

func (d *Drain) UnmarshalState(data []byte) error

UnmarshalState restores the Drain state from JSON.

type DrainConfig 

type DrainConfig struct {
	SimTh                    float64
	Depth                    int
	MaxChildren              int
	MaxClusters              int
	ExtraDelimiters          []string
	ParamStr                 string
	ParametrizeNumericTokens bool
}

DrainConfig holds configuration for creating a new Drain instance.

func DefaultDrainConfig 

func DefaultDrainConfig() DrainConfig

DefaultDrainConfig returns a DrainConfig with safe production defaults.

type DrainSection 

type DrainSection struct {
	SimTh                    float64  `yaml:"sim_th"`
	Depth                    int      `yaml:"depth"`
	MaxChildren              int      `yaml:"max_children"`
	MaxClusters              int      `yaml:"max_clusters"`
	ExtraDelimiters          []string `yaml:"extra_delimiters"`
	ParamStr                 string   `yaml:"param_str"`
	ParametrizeNumericTokens *bool    `yaml:"parametrize_numeric_tokens"`
}

DrainSection holds Drain algorithm parameters.

func (*DrainSection) GetParametrizeNumericTokens 

func (d *DrainSection) GetParametrizeNumericTokens() bool

GetParametrizeNumericTokens returns the effective value, defaulting to true if nil.

func (*DrainSection) GetSimTh 

func (d *DrainSection) GetSimTh() float64

GetSimTh returns the configured similarity threshold.

type ExtractedParameter 

type ExtractedParameter struct {
	Value    string
	MaskName string
}

ExtractedParameter represents a single extracted parameter value and its mask name.

type FilePersistence 

type FilePersistence struct {
	FilePath string
}

FilePersistence saves/loads Drain state to/from a file.

func NewFilePersistence 

func NewFilePersistence(filePath string) *FilePersistence

NewFilePersistence creates a new file-based persistence handler.

func (*FilePersistence) ClearState 

func (f *FilePersistence) ClearState() error

ClearState removes the persisted state file. Missing files are ignored.

func (*FilePersistence) LoadState 

func (f *FilePersistence) LoadState() ([]byte, error)

LoadState reads the state from the file. Returns nil, nil if the file does not exist.

func (*FilePersistence) SaveState 

func (f *FilePersistence) SaveState(state []byte) error

SaveState writes the state to the file, atomically.

func (*FilePersistence) StateSize 

func (f *FilePersistence) StateSize() (int64, error)

StateSize returns the persisted file size without reading its contents.

type LimitsSection 

type LimitsSection struct {
	MaxMessageBytes                  int `yaml:"max_message_bytes"`
	MaxStateBytes                    int `yaml:"max_state_bytes"`
	MaxDecompressedStateBytes        int `yaml:"max_decompressed_state_bytes"`
	ParameterExtractionCacheCapacity int `yaml:"parameter_extraction_cache_capacity"`
}

LimitsSection bounds input, persistence, and cache memory usage. A zero value disables the corresponding limit.

type LogCluster 

type LogCluster struct {
	ClusterID         int      `json:"cluster_id"`
	LogTemplateTokens []string `json:"log_template_tokens"`
	Size              int      `json:"size"`
}

LogCluster represents a discovered log template and the count of messages that match it.

func NewLogCluster 

func NewLogCluster(logTemplateTokens []string, clusterID int) *LogCluster

NewLogCluster creates a new log cluster with the given template tokens and ID.

func (*LogCluster) GetTemplate 

func (lc *LogCluster) GetTemplate() string

GetTemplate returns the template as a single string with tokens joined by spaces.

func (*LogCluster) String 

func (lc *LogCluster) String() string

String returns a human-readable representation of the cluster.

type LogClusterCache 

type LogClusterCache struct {
	// contains filtered or unexported fields
}

LogClusterCache is an LRU cache for log clusters. IMPORTANT: Get() does NOT promote the entry (non-destructive). Only Put() and Touch() move entries to the front (most recently used). This matches the Python Drain3 behavior where reading a cluster for matching should not affect eviction order.

func NewLogClusterCache 

func NewLogClusterCache(maxSize int) *LogClusterCache

NewLogClusterCache creates a new LRU cache with the given maximum size.

func (*LogClusterCache) Get 

func (c *LogClusterCache) Get(clusterID int) *LogCluster

Get retrieves a cluster by ID without affecting eviction order.

func (*LogClusterCache) Len 

func (c *LogClusterCache) Len() int

Len returns the number of clusters in the cache.

func (*LogClusterCache) Put 

func (c *LogClusterCache) Put(cluster *LogCluster) *LogCluster

Put inserts or updates a cluster, promoting it to most-recently-used. If the cache is at capacity, the least-recently-used entry is evicted. Returns the evicted cluster, if any.

func (*LogClusterCache) Remove 

func (c *LogClusterCache) Remove(clusterID int)

Remove removes a cluster from the cache.

func (*LogClusterCache) Touch 

func (c *LogClusterCache) Touch(clusterID int)

Touch promotes a cluster to most-recently-used without modifying it.

func (*LogClusterCache) Values 

func (c *LogClusterCache) Values() []*LogCluster

Values returns all clusters in the cache (most recently used first).

type LogMasker 

type LogMasker struct {
	Instructions []*MaskingInstruction
	MaskPrefix   string
	MaskSuffix   string
	// contains filtered or unexported fields
}

LogMasker applies a sequence of masking instructions to log messages.

func NewLogMasker 

func NewLogMasker(instructions []*MaskingInstruction, maskPrefix, maskSuffix string) *LogMasker

NewLogMasker creates a new LogMasker with the given prefix and suffix for mask tokens.

func (*LogMasker) InstructionsByMaskName 

func (m *LogMasker) InstructionsByMaskName(maskName string) []*MaskingInstruction

InstructionsByMaskName returns all instructions that use the given mask name.

func (*LogMasker) Mask 

func (m *LogMasker) Mask(content string) string

Mask applies all masking instructions sequentially to the content.

func (*LogMasker) MaskNames 

func (m *LogMasker) MaskNames() []string

MaskNames returns all unique mask names from the instructions.

func (*LogMasker) MaskWithError 

func (m *LogMasker) MaskWithError(content string) (string, error)

MaskWithError applies all masking instructions and reports regex execution errors, including timeouts. Instructions that fail are skipped so callers can decide whether to accept the partially masked result.

func (*LogMasker) SetTimeout 

func (m *LogMasker) SetTimeout(timeout time.Duration)

SetTimeout sets a regex match timeout for all masking instructions.

type MaskErrorPolicy 

type MaskErrorPolicy string

MaskErrorPolicy controls whether a message is mined when masking fails. 

const (
	// MaskErrorPolicyReject prevents partially masked messages from changing the model.
	MaskErrorPolicyReject MaskErrorPolicy = "reject"
	// MaskErrorPolicyContinue mines the message using masks that completed successfully.
	MaskErrorPolicyContinue MaskErrorPolicy = "continue"
)

type MaskingInstruction 

type MaskingInstruction struct {
	Pattern    string `yaml:"pattern" json:"pattern"`
	MaskWith   string `yaml:"mask_with" json:"mask_with"`
	MaskPrefix string `yaml:"-" json:"-"`
	MaskSuffix string `yaml:"-" json:"-"`
	// contains filtered or unexported fields
}

MaskingInstruction defines a regex pattern and its replacement mask.

func NewMaskingInstruction 

func NewMaskingInstruction(pattern, maskWith string) (*MaskingInstruction, error)

NewMaskingInstruction creates a new MaskingInstruction with a compiled regex.

type MaskingInstructionConfig 

type MaskingInstructionConfig struct {
	Pattern  string `yaml:"pattern"`
	MaskWith string `yaml:"mask_with"`
}

MaskingInstructionConfig is the YAML representation of a masking instruction.

type MaskingSection 

type MaskingSection struct {
	MaskPrefix         string                     `yaml:"mask_prefix"`
	MaskSuffix         string                     `yaml:"mask_suffix"`
	RegexTimeoutMillis int                        `yaml:"regex_timeout_millis"`
	ErrorPolicy        MaskErrorPolicy            `yaml:"error_policy"`
	Instructions       []MaskingInstructionConfig `yaml:"instructions"`
}

MaskingSection holds masking configuration.

type MemoryPersistence 

type MemoryPersistence struct {
	// contains filtered or unexported fields
}

MemoryPersistence is an in-memory PersistenceHandler, useful for testing.

func NewMemoryPersistence 

func NewMemoryPersistence() *MemoryPersistence

NewMemoryPersistence creates a new in-memory persistence handler.

func (*MemoryPersistence) ClearState 

func (m *MemoryPersistence) ClearState() error

ClearState clears any previously saved in-memory state.

func (*MemoryPersistence) LoadState 

func (m *MemoryPersistence) LoadState() ([]byte, error)

LoadState returns the previously saved state, or nil if none.

func (*MemoryPersistence) SaveState 

func (m *MemoryPersistence) SaveState(state []byte) error

SaveState stores state in memory.

type Node 

type Node struct {
	KeyToChildNode map[string]*Node `json:"key_to_child_node"`
	ClusterIDs     []int            `json:"cluster_ids"`
}

Node represents a node in the Drain prefix tree. Internal nodes hold child pointers keyed by token string. Leaf nodes hold a list of cluster IDs for fast matching.

func NewNode 

func NewNode() *Node

NewNode creates a new empty prefix tree node.

type NullProfiler 

type NullProfiler struct{}

NullProfiler is a no-op profiler that does nothing.

func (NullProfiler) EndSection 

func (NullProfiler) EndSection(string)

func (NullProfiler) Report 

func (NullProfiler) Report(bool) string

func (NullProfiler) StartSection 

func (NullProfiler) StartSection(string)

type Option 

type Option func(*Config, *options)

Option configures a TemplateMiner. Use with New().

func WithCompressState 

func WithCompressState(enabled bool) Option

WithCompressState enables or disables zlib compression for persisted state. Default: true.

func WithConfig 

func WithConfig(cfg *Config) Option

WithConfig uses a pre-built Config directly. This overrides all other drain/masking/snapshot/profiling options set before it.

func WithDepth 

func WithDepth(depth int) Option

WithDepth sets the prefix tree depth. Default: 4.

func WithExtraDelimiters 

func WithExtraDelimiters(delimiters ...string) Option

WithExtraDelimiters sets additional characters to split tokens on.

func WithFilePersistence 

func WithFilePersistence(filePath string) Option

WithFilePersistence sets file-based persistence at the given path.

func WithMaskErrorPolicy 

func WithMaskErrorPolicy(policy MaskErrorPolicy) Option

WithMaskErrorPolicy controls whether masking errors reject a message or allow mining to continue with partial masking. Default: MaskErrorPolicyReject.

func WithMaskPrefix 

func WithMaskPrefix(prefix string) Option

WithMaskPrefix sets the prefix for mask tokens. Default: "<".

func WithMaskSuffix 

func WithMaskSuffix(suffix string) Option

WithMaskSuffix sets the suffix for mask tokens. Default: ">".

func WithMasking 

func WithMasking(regexPattern, maskWith string) Option

WithMasking adds a regex masking instruction. The pattern is applied to log messages before template mining, and matches are replaced with <maskWith>. Can be called multiple times to add multiple masks.

Example: 

drain3.WithMasking(`\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}`, "IP")

func WithMaxChildren 

func WithMaxChildren(maxChildren int) Option

WithMaxChildren sets the maximum children per tree node. Default: 100.

func WithMaxClusters 

func WithMaxClusters(maxClusters int) Option

WithMaxClusters sets the maximum number of clusters (0 = unlimited). Default: 0.

func WithMaxDecompressedStateBytes 

func WithMaxDecompressedStateBytes(maxBytes int) Option

WithMaxDecompressedStateBytes limits JSON state size after decompression. Zero disables the limit.

func WithMaxMessageBytes 

func WithMaxMessageBytes(maxBytes int) Option

WithMaxMessageBytes limits input message size. Zero disables the limit.

func WithMaxStateBytes 

func WithMaxStateBytes(maxBytes int) Option

WithMaxStateBytes limits persisted state size before loading and after saving. Zero disables the limit.

func WithParamStr 

func WithParamStr(paramStr string) Option

WithParamStr sets the wildcard placeholder string. Default: "<*>".

func WithParameterExtractionCacheCapacity 

func WithParameterExtractionCacheCapacity(capacity int) Option

WithParameterExtractionCacheCapacity bounds compiled parameter extraction regexes. Zero disables caching.

func WithParametrizeNumericTokens 

func WithParametrizeNumericTokens(enabled bool) Option

WithParametrizeNumericTokens controls whether numeric tokens are routed to the wildcard node in the prefix tree. Default: true.

func WithPersistence 

func WithPersistence(handler PersistenceHandler) Option

WithPersistence sets a persistence handler for saving/loading state.

func WithProfiling 

func WithProfiling(enabled bool) Option

WithProfiling enables the built-in simple profiler.

func WithRegexTimeout 

func WithRegexTimeout(timeout time.Duration) Option

WithRegexTimeout limits the execution time of each masking regular expression. A zero timeout disables timeout checks. Default: 100ms.

func WithSimTh 

func WithSimTh(simTh float64) Option

WithSimTh sets the similarity threshold (0.0–1.0). Default: 0.4.

func WithSnapshotInterval 

func WithSnapshotInterval(minutes int) Option

WithSnapshotInterval sets how often state is auto-saved (in minutes). Default: 5.

type ParameterExtractor 

type ParameterExtractor struct {
	// contains filtered or unexported fields
}

ParameterExtractor extracts variable values from log messages given a template.

func NewParameterExtractor 

func NewParameterExtractor(masker *LogMasker, extraDelimiters []string) *ParameterExtractor

NewParameterExtractor creates a new ParameterExtractor. Extra delimiters are treated as literal strings, consistent with Drain's tokenization.

func NewParameterExtractorWithCapacity 

func NewParameterExtractorWithCapacity(
	masker *LogMasker,
	extraDelimiters []string,
	cacheCapacity int,
) (*ParameterExtractor, error)

NewParameterExtractorWithCapacity creates an extractor with a bounded LRU cache. A capacity of zero disables caching.

func (*ParameterExtractor) ExtractParameters 

func (pe *ParameterExtractor) ExtractParameters(logTemplate, logMessage string, exactMatching bool) []ExtractedParameter

ExtractParameters extracts parameter values from a log message according to a template. If exactMatching is true and a masker is provided, mask-specific regex patterns are used to capture parameter values more accurately. Returns nil if the message does not match the template.

type PersistenceHandler 

type PersistenceHandler interface {
	SaveState(state []byte) error
	LoadState() ([]byte, error)
}

PersistenceHandler defines the interface for saving/loading Drain state.

type Profiler 

type Profiler interface {
	StartSection(sectionName string)
	EndSection(sectionName string)
	Report(reset bool) string
}

Profiler is an interface for tracking timing of sections in the mining pipeline.

type ProfilingSection 

type ProfilingSection struct {
	Enabled   bool `yaml:"enabled"`
	ReportSec int  `yaml:"report_sec"`
}

ProfilingSection holds profiling configuration.

type SearchStrategy 

type SearchStrategy int

SearchStrategy controls how Drain searches for matching clusters. 

const (
	// SearchStrategyNever disables searching — only new clusters are created.
	SearchStrategyNever SearchStrategy = iota
	// SearchStrategyFallback searches existing clusters only when tree search fails.
	SearchStrategyFallback
	// SearchStrategyAlways always searches all clusters for the best match.
	SearchStrategyAlways
)

func (SearchStrategy) String 

func (s SearchStrategy) String() string

type SimpleProfiler 

type SimpleProfiler struct {
	// contains filtered or unexported fields
}

SimpleProfiler tracks cumulative time spent in named sections.

func NewSimpleProfiler 

func NewSimpleProfiler() *SimpleProfiler

NewSimpleProfiler creates a new SimpleProfiler.

func (*SimpleProfiler) EndSection 

func (p *SimpleProfiler) EndSection(sectionName string)

EndSection ends timing a named section and records the elapsed time.

func (*SimpleProfiler) Report 

func (p *SimpleProfiler) Report(reset bool) string

Report returns a formatted string with profiling results. If reset is true, all accumulated stats are cleared.

func (*SimpleProfiler) StartSection 

func (p *SimpleProfiler) StartSection(sectionName string)

StartSection begins timing a named section.

type SnapshotSection 

type SnapshotSection struct {
	SnapshotIntervalMinutes int  `yaml:"snapshot_interval_minutes"`
	CompressState           bool `yaml:"compress_state"`
}

SnapshotSection holds persistence/snapshot configuration.

type StateClearer 

type StateClearer interface {
	ClearState() error
}

StateClearer is an optional interface for persistence backends that can clear corrupted state and continue with a fresh miner state.

type StateSizeProvider 

type StateSizeProvider interface {
	StateSize() (int64, error)
}

StateSizeProvider optionally reports persisted size before loading, allowing callers to reject oversized state without first allocating it.

type TemplateMiner 

type TemplateMiner struct {
	Drain       *Drain
	Config      *Config
	Persistence PersistenceHandler
	Profiler    Profiler
	Masker      *LogMasker
	// contains filtered or unexported fields
}

TemplateMiner is the high-level API that integrates Drain, masking, persistence, and profiling. Its public mining, matching, profiling, and persistence methods are safe for concurrent use.

func New 

func New(opts ...Option) (*TemplateMiner, error)

New creates a new TemplateMiner with functional options.

Example: 

tm, err := drain3.New(
    drain3.WithSimTh(0.5),
    drain3.WithDepth(5),
    drain3.WithMaxClusters(1000),
    drain3.WithMasking(`\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}`, "IP"),
    drain3.WithMasking(`\b\d+\b`, "NUM"),
    drain3.WithFilePersistence("/tmp/drain3_state.json"),
)
Example 
package main

import (
	"errors"
	"fmt"
	"sort"

	"github.com/FEINIAO233/drain3/v2"
)

func requireCluster(result *drain3.AddLogMessageResult) *drain3.LogCluster {
	if result.InputError != nil {
		panic(result.InputError)
	}
	if result.MaskError != nil {
		panic(result.MaskError)
	}
	if result.Cluster == nil {
		panic(errors.New("drain3 returned no cluster"))
	}
	if result.SaveError != nil {
		panic(result.SaveError)
	}
	return result.Cluster
}

func main() {
	// Create with defaults
	tm, err := drain3.New()
	if err != nil {
		panic(err)
	}

	messages := []string{
		"Failed password for invalid user admin from 192.168.1.1 port 22 ssh2",
		"Failed password for invalid user root from 10.0.0.1 port 22 ssh2",
		"Failed password for invalid user test from 172.16.0.1 port 22 ssh2",
		"Accepted password for admin from 192.168.1.1 port 22 ssh2",
		"Accepted password for admin from 10.0.0.1 port 22 ssh2",
	}

	for _, msg := range messages {
		result := tm.AddLogMessage(msg)
		cluster := requireCluster(result)
		fmt.Printf("change=%s cluster=%s\n", result.ChangeType, cluster.GetTemplate())
	}

	fmt.Printf("\nDiscovered %d templates:\n", tm.ClusterCount())
	clusters := tm.Clusters()
	sort.Slice(clusters, func(i, j int) bool {
		return clusters[i].ClusterID < clusters[j].ClusterID
	})
	for _, c := range clusters {
		fmt.Printf("  [size=%d] %s\n", c.Size, c.GetTemplate())
	}
}

Output:
change=cluster_created cluster=Failed password for invalid user admin from 192.168.1.1 port 22 ssh2
change=cluster_template_changed cluster=Failed password for invalid user <*> from <*> port 22 ssh2
change=none cluster=Failed password for invalid user <*> from <*> port 22 ssh2
change=cluster_created cluster=Accepted password for admin from 192.168.1.1 port 22 ssh2
change=cluster_template_changed cluster=Accepted password for admin from <*> port 22 ssh2

Discovered 2 templates:
  [size=3] Failed password for invalid user <*> from <*> port 22 ssh2
  [size=2] Accepted password for admin from <*> port 22 ssh2
Example (WithOptions) 
package main

import (
	"errors"
	"fmt"

	"github.com/FEINIAO233/drain3/v2"
)

func requireCluster(result *drain3.AddLogMessageResult) *drain3.LogCluster {
	if result.InputError != nil {
		panic(result.InputError)
	}
	if result.MaskError != nil {
		panic(result.MaskError)
	}
	if result.Cluster == nil {
		panic(errors.New("drain3 returned no cluster"))
	}
	if result.SaveError != nil {
		panic(result.SaveError)
	}
	return result.Cluster
}

func main() {
	// Create with custom options
	tm, err := drain3.New(
		drain3.WithSimTh(0.5),
		drain3.WithMasking(`\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}`, "IP"),
		drain3.WithMasking(`\b\d+\b`, "NUM"),
	)
	if err != nil {
		panic(err)
	}

	messages := []string{
		"connection from 192.168.1.1 port 22",
		"connection from 10.0.0.1 port 8080",
		"disconnect from 192.168.1.1 port 443",
	}

	for _, msg := range messages {
		result := tm.AddLogMessage(msg)
		fmt.Printf("template: %s\n", requireCluster(result).GetTemplate())
	}
}

Output:
template: connection from <IP> port <NUM>
template: connection from <IP> port <NUM>
template: disconnect from <IP> port <NUM>

func NewTemplateMiner 

func NewTemplateMiner(persistence PersistenceHandler, config *Config) (*TemplateMiner, error)

NewTemplateMiner creates a new TemplateMiner with the given persistence handler and config. If config is nil, DefaultConfig() is used. If persistence is nil, no persistence is performed.

func (*TemplateMiner) AddLogMessage 

func (tm *TemplateMiner) AddLogMessage(message string) *AddLogMessageResult

AddLogMessage processes a log message through the masking and Drain pipeline.

func (*TemplateMiner) ClusterCount 

func (tm *TemplateMiner) ClusterCount() int

ClusterCount returns the number of active clusters.

func (*TemplateMiner) Clusters 

func (tm *TemplateMiner) Clusters() []*LogCluster

Clusters returns all current log clusters.

func (*TemplateMiner) GetProfilerReport 

func (tm *TemplateMiner) GetProfilerReport(reset bool) string

GetProfilerReport returns the profiler report string.

func (*TemplateMiner) LoadState 

func (tm *TemplateMiner) LoadState() error

LoadState restores Drain state from persistence.

func (*TemplateMiner) Match 

func (tm *TemplateMiner) Match(message string, strategy SearchStrategy) *LogCluster

Match finds the best matching cluster for a log message without modifying state.

func (*TemplateMiner) MatchWithError 

func (tm *TemplateMiner) MatchWithError(message string, strategy SearchStrategy) (*LogCluster, error)

MatchWithError finds the best matching cluster and reports masking failures, including regular expression timeouts.

func (*TemplateMiner) NewParameterExtractor 

func (tm *TemplateMiner) NewParameterExtractor() *ParameterExtractor

NewParameterExtractor creates an extractor using this miner's delimiter, masking, and cache-limit configuration.

func (*TemplateMiner) SaveState 

func (tm *TemplateMiner) SaveState() error

SaveState persists the current Drain state.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.