341 lines
8.5 KiB
Go
341 lines
8.5 KiB
Go
package config
|
|
|
|
import (
|
|
"bufio"
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// Config holds the complete application configuration.
|
|
type Config struct {
|
|
Service ServiceConfig
|
|
Inputs InputsConfig
|
|
Outputs OutputsConfig
|
|
Correlation CorrelationConfig
|
|
}
|
|
|
|
// ServiceConfig holds service-level configuration.
|
|
type ServiceConfig struct {
|
|
Name string
|
|
Language string
|
|
}
|
|
|
|
// InputsConfig holds input sources configuration.
|
|
type InputsConfig struct {
|
|
UnixSockets []UnixSocketConfig
|
|
}
|
|
|
|
// UnixSocketConfig holds a Unix socket source configuration.
|
|
type UnixSocketConfig struct {
|
|
Name string
|
|
Path string
|
|
Format string
|
|
}
|
|
|
|
// OutputsConfig holds output sinks configuration.
|
|
type OutputsConfig struct {
|
|
File FileOutputConfig
|
|
ClickHouse ClickHouseOutputConfig
|
|
Stdout StdoutOutputConfig
|
|
}
|
|
|
|
// FileOutputConfig holds file sink configuration.
|
|
type FileOutputConfig struct {
|
|
Enabled bool
|
|
Path string
|
|
}
|
|
|
|
// ClickHouseOutputConfig holds ClickHouse sink configuration.
|
|
type ClickHouseOutputConfig struct {
|
|
Enabled bool
|
|
DSN string
|
|
Table string
|
|
BatchSize int
|
|
FlushIntervalMs int
|
|
MaxBufferSize int
|
|
DropOnOverflow bool
|
|
AsyncInsert bool
|
|
TimeoutMs int
|
|
}
|
|
|
|
// StdoutOutputConfig holds stdout sink configuration.
|
|
type StdoutOutputConfig struct {
|
|
Enabled bool
|
|
}
|
|
|
|
// CorrelationConfig holds correlation configuration.
|
|
type CorrelationConfig struct {
|
|
Key []string
|
|
TimeWindow TimeWindowConfig
|
|
OrphanPolicy OrphanPolicyConfig
|
|
}
|
|
|
|
// TimeWindowConfig holds time window configuration.
|
|
type TimeWindowConfig struct {
|
|
Value int
|
|
Unit string
|
|
}
|
|
|
|
// OrphanPolicyConfig holds orphan event policy configuration.
|
|
type OrphanPolicyConfig struct {
|
|
ApacheAlwaysEmit bool
|
|
NetworkEmit bool
|
|
}
|
|
|
|
// Load loads configuration from a text file with directives.
|
|
func Load(path string) (*Config, error) {
|
|
file, err := os.Open(path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to open config file: %w", err)
|
|
}
|
|
defer file.Close()
|
|
|
|
cfg := &Config{
|
|
Service: ServiceConfig{
|
|
Name: "logcorrelator",
|
|
Language: "go",
|
|
},
|
|
Inputs: InputsConfig{
|
|
UnixSockets: make([]UnixSocketConfig, 0),
|
|
},
|
|
Outputs: OutputsConfig{
|
|
File: FileOutputConfig{
|
|
Enabled: true,
|
|
Path: "/var/log/logcorrelator/correlated.log",
|
|
},
|
|
ClickHouse: ClickHouseOutputConfig{
|
|
Enabled: false,
|
|
BatchSize: 500,
|
|
FlushIntervalMs: 200,
|
|
MaxBufferSize: 5000,
|
|
DropOnOverflow: true,
|
|
AsyncInsert: true,
|
|
TimeoutMs: 1000,
|
|
},
|
|
Stdout: StdoutOutputConfig{
|
|
Enabled: false,
|
|
},
|
|
},
|
|
Correlation: CorrelationConfig{
|
|
Key: []string{"src_ip", "src_port"},
|
|
TimeWindow: TimeWindowConfig{
|
|
Value: 1,
|
|
Unit: "s",
|
|
},
|
|
OrphanPolicy: OrphanPolicyConfig{
|
|
ApacheAlwaysEmit: true,
|
|
NetworkEmit: false,
|
|
},
|
|
},
|
|
}
|
|
|
|
scanner := bufio.NewScanner(file)
|
|
lineNum := 0
|
|
|
|
for scanner.Scan() {
|
|
lineNum++
|
|
line := strings.TrimSpace(scanner.Text())
|
|
|
|
// Skip empty lines and comments
|
|
if line == "" || strings.HasPrefix(line, "#") {
|
|
continue
|
|
}
|
|
|
|
if err := parseDirective(cfg, line); err != nil {
|
|
return nil, fmt.Errorf("line %d: %w", lineNum, err)
|
|
}
|
|
}
|
|
|
|
if err := scanner.Err(); err != nil {
|
|
return nil, fmt.Errorf("failed to read config file: %w", err)
|
|
}
|
|
|
|
if err := cfg.Validate(); err != nil {
|
|
return nil, fmt.Errorf("invalid config: %w", err)
|
|
}
|
|
|
|
return cfg, nil
|
|
}
|
|
|
|
func parseDirective(cfg *Config, line string) error {
|
|
parts := strings.Fields(line)
|
|
if len(parts) < 2 {
|
|
return fmt.Errorf("invalid directive: %s", line)
|
|
}
|
|
|
|
directive := parts[0]
|
|
value := strings.Join(parts[1:], " ")
|
|
|
|
switch directive {
|
|
case "service.name":
|
|
cfg.Service.Name = value
|
|
case "service.language":
|
|
cfg.Service.Language = value
|
|
|
|
case "input.unix_socket":
|
|
// Format: input.unix_socket <name> <path> [format]
|
|
if len(parts) < 3 {
|
|
return fmt.Errorf("input.unix_socket requires name and path")
|
|
}
|
|
format := "json"
|
|
if len(parts) >= 4 {
|
|
format = parts[3]
|
|
}
|
|
cfg.Inputs.UnixSockets = append(cfg.Inputs.UnixSockets, UnixSocketConfig{
|
|
Name: parts[1],
|
|
Path: parts[2],
|
|
Format: format,
|
|
})
|
|
|
|
case "output.file.enabled":
|
|
enabled, err := parseBool(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.file.enabled: %w", err)
|
|
}
|
|
cfg.Outputs.File.Enabled = enabled
|
|
case "output.file.path":
|
|
cfg.Outputs.File.Path = value
|
|
|
|
case "output.clickhouse.enabled":
|
|
enabled, err := parseBool(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.clickhouse.enabled: %w", err)
|
|
}
|
|
cfg.Outputs.ClickHouse.Enabled = enabled
|
|
case "output.clickhouse.dsn":
|
|
cfg.Outputs.ClickHouse.DSN = value
|
|
case "output.clickhouse.table":
|
|
cfg.Outputs.ClickHouse.Table = value
|
|
case "output.clickhouse.batch_size":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.clickhouse.batch_size: %w", err)
|
|
}
|
|
cfg.Outputs.ClickHouse.BatchSize = v
|
|
case "output.clickhouse.flush_interval_ms":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.clickhouse.flush_interval_ms: %w", err)
|
|
}
|
|
cfg.Outputs.ClickHouse.FlushIntervalMs = v
|
|
case "output.clickhouse.max_buffer_size":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.clickhouse.max_buffer_size: %w", err)
|
|
}
|
|
cfg.Outputs.ClickHouse.MaxBufferSize = v
|
|
case "output.clickhouse.drop_on_overflow":
|
|
enabled, err := parseBool(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.clickhouse.drop_on_overflow: %w", err)
|
|
}
|
|
cfg.Outputs.ClickHouse.DropOnOverflow = enabled
|
|
case "output.clickhouse.async_insert":
|
|
enabled, err := parseBool(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.clickhouse.async_insert: %w", err)
|
|
}
|
|
cfg.Outputs.ClickHouse.AsyncInsert = enabled
|
|
case "output.clickhouse.timeout_ms":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.clickhouse.timeout_ms: %w", err)
|
|
}
|
|
cfg.Outputs.ClickHouse.TimeoutMs = v
|
|
|
|
case "output.stdout.enabled":
|
|
enabled, err := parseBool(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for output.stdout.enabled: %w", err)
|
|
}
|
|
cfg.Outputs.Stdout.Enabled = enabled
|
|
|
|
case "correlation.key":
|
|
cfg.Correlation.Key = strings.Split(value, ",")
|
|
for i, k := range cfg.Correlation.Key {
|
|
cfg.Correlation.Key[i] = strings.TrimSpace(k)
|
|
}
|
|
case "correlation.time_window.value":
|
|
v, err := strconv.Atoi(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for correlation.time_window.value: %w", err)
|
|
}
|
|
cfg.Correlation.TimeWindow.Value = v
|
|
case "correlation.time_window.unit":
|
|
cfg.Correlation.TimeWindow.Unit = value
|
|
case "correlation.orphan_policy.apache_always_emit":
|
|
enabled, err := parseBool(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for correlation.orphan_policy.apache_always_emit: %w", err)
|
|
}
|
|
cfg.Correlation.OrphanPolicy.ApacheAlwaysEmit = enabled
|
|
case "correlation.orphan_policy.network_emit":
|
|
enabled, err := parseBool(value)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid value for correlation.orphan_policy.network_emit: %w", err)
|
|
}
|
|
cfg.Correlation.OrphanPolicy.NetworkEmit = enabled
|
|
|
|
default:
|
|
return fmt.Errorf("unknown directive: %s", directive)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func parseBool(s string) (bool, error) {
|
|
s = strings.ToLower(s)
|
|
switch s {
|
|
case "true", "yes", "1", "on":
|
|
return true, nil
|
|
case "false", "no", "0", "off":
|
|
return false, nil
|
|
default:
|
|
return false, fmt.Errorf("invalid boolean value: %s", s)
|
|
}
|
|
}
|
|
|
|
// Validate validates the configuration.
|
|
func (c *Config) Validate() error {
|
|
if len(c.Inputs.UnixSockets) < 2 {
|
|
return fmt.Errorf("at least two unix socket inputs are required")
|
|
}
|
|
|
|
if !c.Outputs.File.Enabled && !c.Outputs.ClickHouse.Enabled && !c.Outputs.Stdout.Enabled {
|
|
return fmt.Errorf("at least one output must be enabled")
|
|
}
|
|
|
|
if c.Outputs.ClickHouse.Enabled && c.Outputs.ClickHouse.DSN == "" {
|
|
return fmt.Errorf("clickhouse DSN is required when enabled")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetTimeWindow returns the time window as a duration.
|
|
func (c *CorrelationConfig) GetTimeWindow() time.Duration {
|
|
value := c.TimeWindow.Value
|
|
if value <= 0 {
|
|
value = 1
|
|
}
|
|
|
|
unit := c.TimeWindow.Unit
|
|
if unit == "" {
|
|
unit = "s"
|
|
}
|
|
|
|
switch unit {
|
|
case "ms", "millisecond", "milliseconds":
|
|
return time.Duration(value) * time.Millisecond
|
|
case "s", "second", "seconds":
|
|
return time.Duration(value) * time.Second
|
|
case "m", "minute", "minutes":
|
|
return time.Duration(value) * time.Minute
|
|
default:
|
|
return time.Duration(value) * time.Second
|
|
}
|
|
}
|