deduplicate

README

Deduplicate Processor Plugin

The deduplicate processor plugin uses Redis to filter duplicates using configured idempotency_key.

Processor adds ::duplicate label to event with with true value if duplicate was found, otherwise, if event has no configured label or if an error occurs during a request to Redis, it will be false.

Configuration

[[processors]]
  [processors.deduplicate]
    # event label whose value will be used for deduplication
    idempotency_key = "unique_key"

    # maximum number of attempts to execute duplicate search op
    retry_attempts = 0 # zero for endless attempts

    # interval between retries
    retry_after = "5s"

    # Redis connection config
    [processors.deduplicate.redis]
      # if true, one Redis client is shared among the processors in set
      # otherwise, each plugin uses a personal client
      shared = true

      # list of Redis nodes
      servers = [ "localhost:6379" ]

      # Redis credentials
      username = ""
      password = ""

      # operations timeout
      timeout = "30s"

      # keys TTL
      ttl = "1h"      

      # Redis keyspace
      # keys are stored in %keyspace%:%idempotency_key value% format
      keyspace = "neptunus:deduplicate"

      # connection pool settings
      conns_max_open = 2
      conns_max_idle = 1
      conns_max_life_time = "10m"
      conns_max_idle_time = "10m"

Documentation

Index

• type Deduplicate
  • func (p *Deduplicate) Close() error
  • func (p *Deduplicate) Init() error
  • func (p *Deduplicate) Run()
  • func (p *Deduplicate) SetId(id uint64)
• type Redis

Types

type Deduplicate

type Deduplicate struct {
	*core.BaseProcessor `mapstructure:"-"`
	IdempotencyKey      string `mapstructure:"idempotency_key"`
	Redis               Redis  `mapstructure:"redis"`
	*retryer.Retryer    `mapstructure:",squash"`
	// contains filtered or unexported fields
}

func (*Deduplicate) Close

func (p *Deduplicate) Close() error

func (*Deduplicate) Init

func (p *Deduplicate) Init() error

func (*Deduplicate) Run

func (p *Deduplicate) Run()

func (*Deduplicate) SetId

func (p *Deduplicate) SetId(id uint64)

type Redis

type Redis struct {
	Shared           bool          `mapstructure:"shared"`
	Servers          []string      `mapstructure:"servers"`
	Username         string        `mapstructure:"username"`
	Password         string        `mapstructure:"password"`
	Keyspace         string        `mapstructure:"keyspace"`
	TTL              time.Duration `mapstructure:"ttl"`
	Timeout          time.Duration `mapstructure:"timeout"`
	ConnsMaxIdleTime time.Duration `mapstructure:"conns_max_idle_time"`
	ConnsMaxLifetime time.Duration `mapstructure:"conns_max_life_time"`
	ConnsMaxOpen     int           `mapstructure:"conns_max_open"`
	ConnsMaxIdle     int           `mapstructure:"conns_max_idle"`

	*tls.TLSClientConfig `mapstructure:",squash"`
}