stats

README

Stats Processor Plugin

The stats processor calculates count, sum, average, min, max and stores field last value as gauge for each configured field and produces it as an event every period.

Plugin collects and produces stats for each combination of field name and labels values. If incoming event has no any configured label, event will be skipped. If incoming event has no configured field or field type is not a number, field stats will not updated.

Stats stored as child fields in stats key.

This is the format of stats event:

{
  "id": "af002295-7c47-4323-ae5f-f268fad56340",
  "routing_key": "neptunus.generated.metric", # <- configured routing key
  "timestamp": "2023-08-25T22:29:28.9120822+03:00", # <- time of an event creation
  "tags": [],
  "labels": {
    "::line": "3",
    "region": "US/California",
    "::type": "metric", # <- internal label
    "::name": "path.to.one" # <- field name
  },
  "data": { # <- event data
    "stats": { 
      "count": 11,
      "sum": 125,
      "avg": 11.9
    }
  }
}

Configuration

[[processors]]
  [processors.stats]
    # plugin mode, "individual" or "shared"
    # in individual mode each plugin collects and produces it's own stats
    # 
    # in shared mode with multiple processors lines
    # each plugin set uses a shared stats cache
    # and sends stats events to plugins channels using ROUND ROBIN algorithm
    mode = "shared"

    # stats collection, producing and reset interval
    # count, sum and gauge are not reset, other stats are set to zero
    # after stats events are produced
    # if configured value less than 1s, it will be set to 1s 
    period = "1m"

    # routing key with which events will be created
    routing_key = "neptunus.generated.metric"

    # labels of incoming events by which metrics will be grouped
    labels = [ "::line", "region" ]

    # if true, consumed events will be dropped after stats collection
    drop_origin = false

    # "fields" is a "field path -> stats" map
    # plugin expects: "count", "sum", "gauge", "avg", "min", "max"
    # any other value or an empty list will cause an error
    [processors.stats.fields]
      "measurements.count" = ["count", "sum", "avg"]
      temperature = ["gauge", "max", "min"]

Documentation

Index

• type Stats
  • func (p *Stats) Close() error
  • func (p *Stats) Flush()
  • func (p *Stats) Init() error
  • func (p *Stats) Observe(e *core.Event)
  • func (p *Stats) Run()
  • func (p *Stats) SetId(id uint64)

Types

type Stats

type Stats struct {
	*core.BaseProcessor `mapstructure:"-"`
	Period              time.Duration       `mapstructure:"period"`
	Mode                string              `mapstructure:"mode"`
	RoutingKey          string              `mapstructure:"routing_key"`
	Labels              []string            `mapstructure:"labels"`
	DropOrigin          bool                `mapstructure:"drop_origin"`
	Fields              map[string][]string `mapstructure:"fields"`
	// contains filtered or unexported fields
}

func (*Stats) Close

func (p *Stats) Close() error

func (*Stats) Flush

func (p *Stats) Flush()

func (*Stats) Init

func (p *Stats) Init() error

func (*Stats) Observe

func (p *Stats) Observe(e *core.Event)

func (*Stats) Run

func (p *Stats) Run()

func (*Stats) SetId

func (p *Stats) SetId(id uint64)