add occurences and watch window to trigger processor
@@ -7,6 +7,7 @@ import (
"io/ioutil"
"log"
"os"
+ "time"
"github.com/antonmedv/expr"
"github.com/antonmedv/expr/vm"
@@ -23,9 +24,13 @@ const (
// Trigger triggers an action when certain conditions are met
type Trigger struct {
Expression string `mapstructure:"expression,omitempty"`
+ MaxOccurrences int `mapstructure:"max-occurrences,omitempty"`
+ Window time.Duration `mapstructure:"window,omitempty"`
Action map[string]interface{} `mapstructure:"action,omitempty"`
Debug bool `mapstructure:"debug,omitempty"`
+ numOccurrences int
+ lastOccurrence time.Time
prg *vm.Program
action actions.Action
@@ -59,11 +64,16 @@ func (p *Trigger) Init(cfg interface{}, logger *log.Logger) error {
if err != nil {
return err
}
+ err = p.setDefaults()
+ if err != nil {
+ return err
+ }
p.logger.Printf("%q initalized: %+v", processorType, p)
return nil
}
func (p *Trigger) Apply(es ...*formatters.EventMsg) []*formatters.EventMsg {
+ now := time.Now()
for _, e := range es {
if e == nil {
continue
@@ -87,14 +97,27 @@ func (p *Trigger) Apply(es ...*formatters.EventMsg) []*formatters.EventMsg {
switch res := res.(type) {
case bool:
if res {
+ p.numOccurrences++
+ // within the window
+ if p.lastOccurrence.Add(p.Window).After(now) {
+ // max occurrences reached
+ if p.MaxOccurrences <= p.numOccurrences {
+ // reset numOccurrences
+ p.numOccurrences = 0
+ // run the action
go func() {
res, err := p.action.Run(e)
if err != nil {
+ p.logger.Printf("trigger action failed: %+v", err)
return
}
p.logger.Printf("result: %+v", res)
}()
}
+ continue
+ }
+ p.lastOccurrence = now
+ }
}
}
return nil
@@ -130,3 +153,10 @@ func (p *Trigger) String() string {
}
return string(b)
}
+
+func (p *Trigger) setDefaults() error {
+ if p.MaxOccurrences <= 0 {
+ p.MaxOccurrences = 1
+ }
+ return nil
+}
