leakybucket

package
v0.3.3 Latest Latest
Warning

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

 Go to latest Published: Sep 29, 2020 License: MIT Imports: 26 Imported by: 1

README

Leakybuckets

Bucket concepts

Leakybucket is used for decision making. Under certain conditions enriched events are poured in these buckets. When these buckets are full, we raise a new event. After this event is raised the bucket is destroyed. There are many types of buckets, and we welcome any new useful design of buckets.

Usually the bucket configuration generates the creation of many buckets. They are differenciated by a field called stackkey. When two events arrives with the same stackkey they go in the same matching bucket.

The very purpose of these buckets is to detect clients that exceed a certain rate of attemps to do something (ssh connection, http authentication failure, etc...). Thus, the most use stackkey field is often the source_ip.

Standard leaky buckets

Default buckets have two main configuration options:

  • capacity: number of events the bucket can hold. When the capacity is reached and a new event is poured, a new event is raised. We call this type of event overflow. This is an int.
  • leakspeed: duration needed for an event to leak. When an event leaks, it disappear from the bucket.

## Trigger

It's a special type of bucket with a zero capacity. Thus, when an event is poured in a trigger, it always raises an overflow.

Uniq

It's a bucket working as the standard leaky bucket except for one thing: a filter returns a property for each event and only one occurence of this property is allowed in the bucket, thus the bucket is called uniq.

Counter

It's a special type of bucket with an infinite capacity and an infinite leakspeed (it never overflows, neither leaks). Nevertheless, the event is raised after a fixed duration. The option is called duration.

Available configuration options for buckets

Fields for standard buckets
  • type: mandatory field. Must be one of "leaky", "trigger", "uniq" or "counter"
  • name: mandatory field, but the value is totally open. Nevertheless this value will tag the events raised by the bucket.
  • filter: mandatory field. It's a filter that is run when the decision to make an event match the bucket or not. The filter have to return a boolean. As a filter implementation we use https://github.com/antonmedv/expr
  • capacity: [mandatory for now, shouldn't be mandatory in the final version] it's the size of the bucket. When pouring in a bucket already with size events, it overflows.
  • leakspeed: leakspeed is a time duration (has to be parseable by https://golang.org/pkg/time/#ParseDuration). After each interval an event is leaked from the bucket.
  • stackkey: mandatory field. This field is used to differenciate on which bucket ongoing events will be poured. When an unknows stackkey is seen in an event a new bucekt is created.
  • on_overflow: optional field, that tells the what to do when the bucket is returning the overflow event. As of today, the possibility are these: "ban,1h", "Reprocess", "Delete". Reprocess is used to send the raised event back in the event pool to be matched agains buckets
Fields for special buckets
Uniq

Uniq has an extra field uniq_filter which is too use the filter implementation from https://github.com/antonmedv/expr. The filter must return a string. All strins returned by this filter in the same buckets have to be different. Thus, if a string is seen twice it is dismissed.

Trigger

Capacity and leakspeed are not relevant for this kind of bucket.

Counter

It's a special kind of bucket that raise an event and is destroyed after a fixed duration. The configuration field used is duration and must be parseable by https://golang.org/pkg/time/#ParseDuration. Nevertheless, this kind of bucket is often used with an infinite leakspeed and an infinite capacity [capacity set to -1 for now].

Add exemples here

# ssh bruteforce
- type: leaky
  name: ssh_bruteforce
  filter: "Meta.log_type == 'ssh_failed-auth'"
  leakspeed: "10s"
  capacity: 5
  stackkey: "source_ip"
  on_overflow: ban,1h

# reporting of src_ip,dest_port seen
- type: counter
  name: counter
  filter: "Meta.service == 'tcp' && Event.new_connection == 'true'"
  distinct: "Meta.source_ip + ':' + Meta.dest_port"
  duration: 5m
  capacity: -1

- type: trigger
  name: "New connection"
  filter: "Meta.service == 'tcp' && Event.new_connection == 'true'"
  on_overflow: Reprocess

Note on leakybuckets implementation

[This is not dry enough to have many details here, but:]

The bucket code is triggered by InfiniBucketify in main.go. There's one struct called buckets which is for now a map[string]interface{} that holds all buckets. The key of this map is derivated from the filter configured for the bucket and its stackkey. This looks like complicated, but in fact it allows us to use only one structs. This is done in buckets.go.

On top of that the implementation define only the standard leaky bucket. A goroutine is launched for every buckets (bucket.go). This goroutine manages the life of the bucket.

For special buckets, hooks are defined at initialization time in manager.go. Hooks are called when relevant by the bucket gorourine when events are poured and/or when bucket overflows.

Documentation

Index

Constants

View Source 
const (
	LIVE = iota
	TIMEMACHINE
)

Variables

View Source 
var BucketsCurrentCount = prometheus.NewGaugeVec(
	prometheus.GaugeOpts{
		Name: "cs_buckets",
		Help: "Number of buckets that currently exist.",
	},
	[]string{"name"},
)
View Source 
var BucketsInstanciation = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "cs_bucket_created_total",
		Help: "Total buckets were instanciated.",
	},
	[]string{"name"},
)
View Source 
var BucketsOverflow = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "cs_bucket_overflowed_total",
		Help: "Total buckets overflowed.",
	},
	[]string{"name"},
)
View Source 
var BucketsPour = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "cs_bucket_poured_total",
		Help: "Total events were poured in bucket.",
	},
	[]string{"source", "name"},
)
View Source 
var BucketsUnderflow = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "cs_bucket_underflowed_total",
		Help: "Total buckets underflowed.",
	},
	[]string{"name"},
)
View Source 
var LeakyRoutineCount int64

Functions

func DumpBucketsStateAt 

func DumpBucketsStateAt(deadline time.Time, buckets *Buckets) (string, error)

func FormatOverflow 

func FormatOverflow(l *Leaky, queue *Queue) types.SignalOccurence

func GarbageCollectBuckets 

func GarbageCollectBuckets(deadline time.Time, buckets *Buckets) error

The leaky routines lifecycle are based on "real" time. But when we are running in time-machine mode, the reference time is in logs and not "real" time. Thus we need to garbage collect them to avoid a skyrocketing memory usage.

func GetKey 

func GetKey(bucketCfg BucketFactory, stackkey string) string

func LeakRoutine 

func LeakRoutine(l *Leaky)
for now mimic a leak routine

LeakRoutine us the life of a bucket. It dies when the bucket underflows or overflows

func LoadBucket 

func LoadBucket(g *BucketFactory, dataFolder string) error

Init recursively process yaml files from a directory and loads them as BucketFactory

func LoadBucketsState 

func LoadBucketsState(file string, buckets *Buckets, holders []BucketFactory) error

func Pour 

func Pour(l *Leaky, msg types.Event)

func PourItemToHolders 

func PourItemToHolders(parsed types.Event, holders []BucketFactory, buckets *Buckets) (bool, error)

func ShutdownAllBuckets added in v0.2.0 

func ShutdownAllBuckets(buckets *Buckets) error

func TimeMachinePour 

func TimeMachinePour(l *Leaky, msg types.Event)

func ValidateFactory 

func ValidateFactory(b *BucketFactory) error

Types

type Blackhole 

type Blackhole struct {
	DumbProcessor
	// contains filtered or unexported fields
}

func NewBlackhole 

func NewBlackhole(g *BucketFactory) (*Blackhole, error)

func (*Blackhole) OnBucketOverflow 

func (bl *Blackhole) OnBucketOverflow(b *BucketFactory) func(*Leaky, types.SignalOccurence, *Queue) (types.SignalOccurence, *Queue)

type BucketFactory 

type BucketFactory struct {
	FormatVersion string            `yaml:"format"`
	Author        string            `yaml:"author"`
	Description   string            `yaml:"description"`
	References    []string          `yaml:"references"`
	Type          string            `yaml:"type"`                //Type can be : leaky, counter, trigger. It determines the main bucket characteristics
	Name          string            `yaml:"name"`                //Name of the bucket, used later in log and user-messages. Should be unique
	Capacity      int               `yaml:"capacity"`            //Capacity is applicable to leaky buckets and determines the "burst" capacity
	LeakSpeed     string            `yaml:"leakspeed"`           //Leakspeed is a float representing how many events per second leak out of the bucket
	Duration      string            `yaml:"duration"`            //Duration allows 'counter' buckets to have a fixed life-time
	Filter        string            `yaml:"filter"`              //Filter is an expr that determines if an event is elligible for said bucket. Filter is evaluated against the Event struct
	GroupBy       string            `yaml:"groupby,omitempty"`   //groupy is an expr that allows to determine the partitions of the bucket. A common example is the source_ip
	Distinct      string            `yaml:"distinct"`            //Distinct, when present, adds a `Pour()` processor that will only pour uniq items (based on uniq_filter expr result)
	Debug         bool              `yaml:"debug"`               //Debug, when set to true, will enable debugging for _this_ scenario specifically
	Labels        map[string]string `yaml:"labels"`              //Labels is K:V list aiming at providing context the overflow
	Blackhole     string            `yaml:"blackhole,omitempty"` //Blackhole is a duration that, if present, will prevent same bucket partition to overflow more often than $duration

	Reprocess      bool                      `yaml:"reprocess"`       //Reprocess, if true, will for the bucket to be re-injected into processing chain
	CacheSize      int                       `yaml:"cache_size"`      //CacheSize, if > 0, limits the size of in-memory cache of the bucket
	Profiling      bool                      `yaml:"profiling"`       //Profiling, if true, will make the bucket record pours/overflows/etc.
	OverflowFilter string                    `yaml:"overflow_filter"` //OverflowFilter if present, is a filter that must return true for the overflow to go through
	BucketName     string                    `yaml:"-"`
	Filename       string                    `yaml:"-"`
	RunTimeFilter  *vm.Program               `json:"-"`
	ExprDebugger   *exprhelpers.ExprDebugger `yaml:"-" json:"-"` // used to debug expression by printing the content of each variable of the expression
	RunTimeGroupBy *vm.Program               `json:"-"`
	Data           []*types.DataSource       `yaml:"data,omitempty"`
	// contains filtered or unexported fields
}

BucketFactory struct holds all fields for any bucket configuration. This is to have a generic struct for buckets. This can be seen as a bucket factory.

func Init 

func Init(cfg map[string]string) ([]BucketFactory, chan types.Event, error)

Init recursively process yaml files from a directory and loads them as BucketFactory

func LoadBucketDir 

func LoadBucketDir(dir string, dataFolder string) ([]BucketFactory, chan types.Event, error)

func LoadBuckets 

func LoadBuckets(files []string, dataFolder string) ([]BucketFactory, chan types.Event, error)

type Buckets 

type Buckets struct {
	Bucket_map sync.Map
}

Buckets is the struct used to hold buckets in the context of main.go the idea is to have one struct to rule them all

func NewBuckets 

func NewBuckets() *Buckets

NewBuckets create the Buckets struct

type DumbProcessor 

type DumbProcessor struct {
}

func (*DumbProcessor) OnBucketInit 

func (d *DumbProcessor) OnBucketInit(b *BucketFactory) error

func (*DumbProcessor) OnBucketOverflow 

func (d *DumbProcessor) OnBucketOverflow(b *BucketFactory) func(*Leaky, types.SignalOccurence, *Queue) (types.SignalOccurence, *Queue)

func (*DumbProcessor) OnBucketPour 

func (d *DumbProcessor) OnBucketPour(b *BucketFactory) func(types.Event, *Leaky) *types.Event

type HiddenKey 

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

type Leaky 

type Leaky struct {
	Name string
	Mode int //LIVE or TIMEMACHINE
	//the limiter is what holds the proper "leaky aspect", it determines when/if we can pour objects
	Limiter         rate.RateLimiter `json:"-"`
	SerializedState rate.Lstate
	//Queue is used to held the cache of objects in the bucket, it is used to know 'how many' objects we have in buffer.
	Queue *Queue
	//Leaky buckets are receiving message through a chan
	In chan types.Event `json:"-"`
	//Leaky buckets are pushing their overflows through a chan
	Out chan *Queue `json:"-"`
	// shared for all buckets (the idea is to kill this afterwards)
	AllOut     chan types.Event `json:"-"`
	KillSwitch chan bool        `json:"-"`
	//max capacity (for burst)
	Capacity int
	//CacheRatio is the number of elements that should be kept in memory (compared to capacity)
	CacheSize int
	//the unique identifier of the bucket (a hash)
	Mapkey string
	// chan for signaling
	Signal       chan bool `json:"-"`
	Reprocess    bool
	Uuid         string
	First_ts     time.Time
	Last_ts      time.Time
	Ovflw_ts     time.Time
	Total_count  int
	Leakspeed    time.Duration
	BucketConfig *BucketFactory
	Duration     time.Duration
	Pour         func(*Leaky, types.Event) `json:"-"`
	//Profiling when set to true enables profiling of bucket
	Profiling bool
	// contains filtered or unexported fields
}

Leaky represents one instance of a bucket

func FromFactory 

func FromFactory(g BucketFactory) *Leaky

func NewLeaky 

func NewLeaky(g BucketFactory) *Leaky

Newleaky creates a new leaky bucket from a BucketFactory Events created by the bucket (overflow, bucket empty) are sent to a chan defined by BucketFactory The leaky bucket implementation is based on rate limiter (see https://godoc.org/golang.org/x/time/rate) There's a trick to have an event said when the bucket gets empty to allow its destruction

func NewTimeMachine 

func NewTimeMachine(g BucketFactory) *Leaky

type OverflowFilter 

type OverflowFilter struct {
	Filter        string
	FilterRuntime *vm.Program
	DumbProcessor
}

func NewOverflowFilter 

func NewOverflowFilter(g *BucketFactory) (*OverflowFilter, error)

func (*OverflowFilter) OnBucketOverflow 

func (u *OverflowFilter) OnBucketOverflow(Bucket *BucketFactory) func(*Leaky, types.SignalOccurence, *Queue) (types.SignalOccurence, *Queue)

type Processor 

type Processor interface {
	OnBucketInit(Bucket *BucketFactory) error
	OnBucketPour(Bucket *BucketFactory) func(types.Event, *Leaky) *types.Event
	OnBucketOverflow(Bucket *BucketFactory) func(*Leaky, types.SignalOccurence, *Queue) (types.SignalOccurence, *Queue)
}

type Queue 

type Queue struct {
	Queue []types.Event
	L     int //capacity
}

Queue holds a limited size queue

func NewQueue 

func NewQueue(l int) *Queue

NewQueue create a new queue with a size of l

func (*Queue) Add 

func (q *Queue) Add(m types.Event)

Add an event in the queue. If it has already l elements, the first element is dropped before adding the new m element

func (*Queue) GetQueue 

func (q *Queue) GetQueue() []types.Event

GetQueue returns the entire queue

type Trigger 

type Trigger struct {
	DumbProcessor
}

func (*Trigger) OnBucketPour 

func (t *Trigger) OnBucketPour(b *BucketFactory) func(types.Event, *Leaky) *types.Event

type Uniq 

type Uniq struct {
	DistinctCompiled *vm.Program
}

func (*Uniq) OnBucketInit 

func (u *Uniq) OnBucketInit(Bucket *BucketFactory) error

func (*Uniq) OnBucketOverflow 

func (u *Uniq) OnBucketOverflow(Bucket *BucketFactory) func(*Leaky, types.SignalOccurence, *Queue) (types.SignalOccurence, *Queue)

func (*Uniq) OnBucketPour 

func (u *Uniq) OnBucketPour(Bucket *BucketFactory) func(types.Event, *Leaky) *types.Event

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.