multi event output filters and array split filter

f4t · Dec 30, 2020 · 677eeae · 677eeae
1 parent d461391
commit 677eeae
Show file tree

Hide file tree

Showing 17 changed files with 245 additions and 50 deletions.
diff --git a/config/filter.go b/config/filter.go
@@ -16,7 +16,7 @@ var (
 // TypeFilterConfig is interface of filter module
 type TypeFilterConfig interface {
 	TypeCommonConfig
-	Event(context.Context, logevent.LogEvent) (logevent.LogEvent, bool)
+	Event(context.Context, logevent.LogEvent) ([]logevent.LogEvent, bool)
 	CommonFilter(context.Context, logevent.LogEvent) logevent.LogEvent
 }
 
@@ -25,6 +25,7 @@ func (f *FilterConfig) IsConfigured() bool {
 	return len(f.AddTags) != 0 || len(f.AddFields) != 0 || len(f.RemoveTags) != 0 || len(f.RemoveFields) != 0
 }
 
+// CommonFilter applies common inline filters such as add/remove fields/tags
 func (f *FilterConfig) CommonFilter(ctx context.Context, event logevent.LogEvent) logevent.LogEvent {
 
 	event.AddTag(f.AddTags...)
@@ -102,13 +103,26 @@ func (t *Config) startFilters() (err error) {
 				}
 			case event := <-t.chInFilter:
 				var ok bool
+				events := make([]logevent.LogEvent, 0)
+				events = append(events, event)
 				for _, filter := range filters {
-					event, ok = filter.Event(t.ctx, event)
-					if ok {
-						event = filter.CommonFilter(t.ctx, event)
+					eventsOut := make([]logevent.LogEvent, 0)
+					for _, event := range events {
+						var filteredEvents []logevent.LogEvent
+						filteredEvents, ok = filter.Event(t.ctx, event)
+						if ok {
+							for i, evt := range filteredEvents {
+								filteredEvents[i] = filter.CommonFilter(t.ctx, evt)
+							}
+						}
+						eventsOut = append(eventsOut, filteredEvents...)
 					}
+					events = eventsOut
+				}
+
+				for _, evt := range events {
+					t.chFilterOut <- evt
 				}
-				t.chFilterOut <- event
 			}
 		}
 	})

diff --git a/filter/addfield/filteraddfield.go b/filter/addfield/filteraddfield.go
@@ -39,10 +39,14 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	if _, ok := event.Extra[f.Key]; ok {
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
+
 	event.SetValue(f.Key, event.Format(f.Value))
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
diff --git a/filter/cond/filtercond.go b/filter/cond/filtercond.go
@@ -137,34 +137,42 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	if f.expression != nil {
 		ep := EventParameters{Event: &event}
 		ret, err := f.expression.Eval(&ep)
 		if err != nil {
 			goglog.Logger.Error(err)
 			event.AddTag(ErrorTag)
-			return event, false
+			eventsOut = append(eventsOut, event)
+			return eventsOut, false
 		}
 		if r, ok := ret.(bool); ok {
 			if r {
 				for _, filter := range f.filters {
-					event, ok = filter.Event(ctx, event)
+					eventsOut, ok = filter.Event(ctx, event)
 					if ok {
-						event = filter.CommonFilter(ctx, event)
+						for _, evt := range eventsOut {
+							evt = filter.CommonFilter(ctx, evt)
+						}
 					}
 				}
 			} else {
 				for _, filter := range f.elseFilters {
-					event, ok = filter.Event(ctx, event)
+					eventsOut, ok = filter.Event(ctx, event)
 					if ok {
-						event = filter.CommonFilter(ctx, event)
+						for _, evt := range eventsOut {
+							evt = filter.CommonFilter(ctx, evt)
+						}
 					}
 				}
 			}
-			return event, true
+			eventsOut = append(eventsOut, event)
+			return eventsOut, true
 		}
 		goglog.Logger.Warn("filter cond condition returns not a boolean, ignored")
 	}
-	return event, false
+	eventsOut = append(eventsOut, event)
+	return eventsOut, false
 }
diff --git a/filter/date/filterdate.go b/filter/date/filterdate.go
@@ -65,7 +65,8 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	var (
 		timestamp time.Time
 		err       error
@@ -95,15 +96,17 @@ func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (loge
 	if err != nil {
 		event.AddTag(ErrorTag)
 		goglog.Logger.Error(err)
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 	if f.Target == DefaultTarget {
 		event.Timestamp = timestamp.UTC()
 	} else {
 		event.SetValue(f.Target, timestamp.UTC())
 	}
 
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
 
 func convertFloat(value float64) (int64, int64) {

diff --git a/filter/geoip2/filtergeoip2.go b/filter/geoip2/filtergeoip2.go
@@ -84,16 +84,19 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	ipstr := event.GetString(f.IPField)
 	if ipstr == "" {
 		// Passthru if empty
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 	ip := net.ParseIP(ipstr)
 	if f.SkipPrivate && f.privateIP(ip) {
 		// Passthru
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 	var err error
 	var record *geoip2.City
@@ -107,17 +110,20 @@ func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (loge
 				goglog.Logger.Error(err)
 			}
 			event.AddTag(ErrorTag)
-			return event, false
+			eventsOut = append(eventsOut, event)
+			return eventsOut, false
 		}
 		f.cache.Add(ipstr, record)
 	}
 	if record == nil {
 		event.AddTag(ErrorTag)
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 	if record.Location.Latitude == 0 && record.Location.Longitude == 0 {
 		event.AddTag(ErrorTag)
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 
 	if f.FlatFormat {
@@ -172,7 +178,8 @@ func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (loge
 		event.SetValue(f.Key, m)
 	}
 
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
 
 func (f *FilterConfig) privateIP(ip net.IP) bool {

diff --git a/filter/gonx/filtergonx.go b/filter/gonx/filtergonx.go
@@ -72,20 +72,23 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	message := event.GetString(f.Source)
 	reader := gonx.NewParserReader(strings.NewReader(message), f.parser)
 	entry, err := reader.Read()
 	if err != nil {
 		event.AddTag(ErrorTag)
 		goglog.Logger.Errorf("%s: %q", err, message)
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 
 	for _, field := range f.fields {
 		s, _ := entry.Field(field)
 		event.SetValue(field, s)
 	}
 
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
diff --git a/filter/grok/filtergrok.go b/filter/grok/filtergrok.go
@@ -78,7 +78,8 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	message := event.GetString(f.Source)
 	found := false
 	for _, thisMatch := range f.Match {
@@ -105,8 +106,10 @@ func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (loge
 	if !found {
 		event.AddTag(ErrorTag)
 		goglog.Logger.Debugf("grok: no matches for %q", message)
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
diff --git a/filter/json/filterjson.go b/filter/json/filterjson.go
@@ -49,12 +49,14 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	var parsedMessage map[string]interface{}
 	if err := jsoniter.Unmarshal([]byte(event.GetString(f.Source)), &parsedMessage); err != nil {
 		event.AddTag(ErrorTag)
 		goglog.Logger.Error(err)
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 
 	if f.Appendkey != "" {
@@ -79,5 +81,6 @@ func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (loge
 		}
 	}
 
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
diff --git a/filter/mutate/filtermutate.go b/filter/mutate/filtermutate.go
@@ -59,7 +59,8 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	if f.Split[0] != "" {
 		event.SetValue(f.Split[0], strings.Split(event.GetString(f.Split[0]), f.Split[1]))
 	}
@@ -75,7 +76,8 @@ func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (loge
 		event.Remove(f.Rename[0])
 	}
 	// always return true here for configured filter
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
 
 func mergeField(event logevent.LogEvent, destinationName, source string) logevent.LogEvent {

diff --git a/filter/ratelimit/filterratelimit.go b/filter/ratelimit/filterratelimit.go
@@ -70,15 +70,18 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	if event.Extra == nil {
 		event.Extra = map[string]interface{}{}
 	}
 
 	if f.throttle == nil {
-		return event, false
+		eventsOut = append(eventsOut, event)
+		return eventsOut, false
 	}
 
 	<-f.throttle
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }
diff --git a/filter/removefield/filterremovefield.go b/filter/removefield/filterremovefield.go
@@ -49,7 +49,8 @@ func InitHandler(ctx context.Context, raw *config.ConfigRaw) (config.TypeFilterC
 }
 
 // Event the main filter event
-func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (logevent.LogEvent, bool) {
+func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) ([]logevent.LogEvent, bool) {
+	eventsOut := make([]logevent.LogEvent, 0)
 	if event.Extra == nil {
 		event.Extra = map[string]interface{}{}
 	}
@@ -63,5 +64,6 @@ func (f *FilterConfig) Event(ctx context.Context, event logevent.LogEvent) (loge
 	}
 
 	// TODO: remove unset field return false
-	return event, true
+	eventsOut = append(eventsOut, event)
+	return eventsOut, true
 }