Refact pkg/parser/node (#2953)
* extract method processFilter() * extract method processWhitelist() * lint (whitespace, errors)
This commit is contained in:
parent
d2c4bc55fc
commit
c4473839c4
2 changed files with 156 additions and 84 deletions
|
@ -3,7 +3,7 @@
|
|||
linters-settings:
|
||||
cyclop:
|
||||
# lower this after refactoring
|
||||
max-complexity: 53
|
||||
max-complexity: 48
|
||||
|
||||
gci:
|
||||
sections:
|
||||
|
@ -22,7 +22,7 @@ linters-settings:
|
|||
|
||||
gocyclo:
|
||||
# lower this after refactoring
|
||||
min-complexity: 49
|
||||
min-complexity: 48
|
||||
|
||||
funlen:
|
||||
# Checks the number of lines in a function.
|
||||
|
|
|
@ -65,26 +65,27 @@ type Node struct {
|
|||
}
|
||||
|
||||
func (n *Node) validate(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
||||
|
||||
// stage is being set automagically
|
||||
if n.Stage == "" {
|
||||
return fmt.Errorf("stage needs to be an existing stage")
|
||||
return errors.New("stage needs to be an existing stage")
|
||||
}
|
||||
|
||||
/* "" behaves like continue */
|
||||
if n.OnSuccess != "continue" && n.OnSuccess != "next_stage" && n.OnSuccess != "" {
|
||||
return fmt.Errorf("onsuccess '%s' not continue,next_stage", n.OnSuccess)
|
||||
}
|
||||
|
||||
if n.Filter != "" && n.RunTimeFilter == nil {
|
||||
return fmt.Errorf("non-empty filter '%s' was not compiled", n.Filter)
|
||||
}
|
||||
|
||||
if n.Grok.RunTimeRegexp != nil || n.Grok.TargetField != "" {
|
||||
if n.Grok.TargetField == "" && n.Grok.ExpValue == "" {
|
||||
return fmt.Errorf("grok requires 'expression' or 'apply_on'")
|
||||
return errors.New("grok requires 'expression' or 'apply_on'")
|
||||
}
|
||||
|
||||
if n.Grok.RegexpName == "" && n.Grok.RegexpValue == "" {
|
||||
return fmt.Errorf("grok needs 'pattern' or 'name'")
|
||||
return errors.New("grok needs 'pattern' or 'name'")
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -93,6 +94,7 @@ func (n *Node) validate(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
if static.ExpValue == "" {
|
||||
return fmt.Errorf("static %d : when method is set, expression must be present", idx)
|
||||
}
|
||||
|
||||
if _, ok := ectx.Registered[static.Method]; !ok {
|
||||
log.Warningf("the method '%s' doesn't exist or the plugin has not been initialized", static.Method)
|
||||
}
|
||||
|
@ -100,6 +102,7 @@ func (n *Node) validate(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
if static.Meta == "" && static.Parsed == "" && static.TargetByName == "" {
|
||||
return fmt.Errorf("static %d : at least one of meta/event/target must be set", idx)
|
||||
}
|
||||
|
||||
if static.Value == "" && static.RunTimeValue == nil {
|
||||
return fmt.Errorf("static %d value or expression must be set", idx)
|
||||
}
|
||||
|
@ -110,15 +113,19 @@ func (n *Node) validate(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
if stash.Name == "" {
|
||||
return fmt.Errorf("stash %d : name must be set", idx)
|
||||
}
|
||||
|
||||
if stash.Value == "" {
|
||||
return fmt.Errorf("stash %s : value expression must be set", stash.Name)
|
||||
}
|
||||
|
||||
if stash.Key == "" {
|
||||
return fmt.Errorf("stash %s : key expression must be set", stash.Name)
|
||||
}
|
||||
|
||||
if stash.TTL == "" {
|
||||
return fmt.Errorf("stash %s : ttl must be set", stash.Name)
|
||||
}
|
||||
|
||||
if stash.Strategy == "" {
|
||||
stash.Strategy = "LRU"
|
||||
}
|
||||
|
@ -127,23 +134,23 @@ func (n *Node) validate(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
stash.MaxMapSize = 100
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[string]interface{}) (bool, error) {
|
||||
var NodeState bool
|
||||
var NodeHasOKGrok bool
|
||||
func (n *Node) processFilter(cachedExprEnv map[string]interface{}) (bool, error) {
|
||||
clog := n.Logger
|
||||
if n.RunTimeFilter == nil {
|
||||
clog.Tracef("Node has not filter, enter")
|
||||
return true, nil
|
||||
}
|
||||
|
||||
cachedExprEnv := expressionEnv
|
||||
|
||||
clog.Tracef("Event entering node")
|
||||
if n.RunTimeFilter != nil {
|
||||
// Evaluate node's filter
|
||||
output, err := exprhelpers.Run(n.RunTimeFilter, cachedExprEnv, clog, n.Debug)
|
||||
if err != nil {
|
||||
clog.Warningf("failed to run filter : %v", err)
|
||||
clog.Debugf("Event leaving node : ko")
|
||||
|
||||
return false, nil
|
||||
}
|
||||
|
||||
|
@ -156,26 +163,26 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
default:
|
||||
clog.Warningf("Expr '%s' returned non-bool, abort : %T", n.Filter, output)
|
||||
clog.Debugf("Event leaving node : ko")
|
||||
|
||||
return false, nil
|
||||
}
|
||||
NodeState = true
|
||||
} else {
|
||||
clog.Tracef("Node has not filter, enter")
|
||||
NodeState = true
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
if n.Name != "" {
|
||||
NodesHits.With(prometheus.Labels{"source": p.Line.Src, "type": p.Line.Module, "name": n.Name}).Inc()
|
||||
}
|
||||
exprErr := error(nil)
|
||||
func (n *Node) processWhitelist(cachedExprEnv map[string]interface{}, p *types.Event) (bool, error) {
|
||||
var exprErr error
|
||||
|
||||
isWhitelisted := n.CheckIPsWL(p)
|
||||
if !isWhitelisted {
|
||||
isWhitelisted, exprErr = n.CheckExprWL(cachedExprEnv, p)
|
||||
}
|
||||
|
||||
if exprErr != nil {
|
||||
// Previous code returned nil if there was an error, so we keep this behavior
|
||||
return false, nil //nolint:nilerr
|
||||
}
|
||||
|
||||
if isWhitelisted && !p.Whitelisted {
|
||||
p.Whitelisted = true
|
||||
p.WhitelistReason = n.Whitelist.Reason
|
||||
|
@ -185,13 +192,46 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
for k := range p.Overflow.Sources {
|
||||
ips = append(ips, k)
|
||||
}
|
||||
clog.Infof("Ban for %s whitelisted, reason [%s]", strings.Join(ips, ","), n.Whitelist.Reason)
|
||||
|
||||
n.Logger.Infof("Ban for %s whitelisted, reason [%s]", strings.Join(ips, ","), n.Whitelist.Reason)
|
||||
|
||||
p.Overflow.Whitelisted = true
|
||||
}
|
||||
}
|
||||
|
||||
return isWhitelisted, nil
|
||||
}
|
||||
|
||||
func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[string]interface{}) (bool, error) {
|
||||
var NodeHasOKGrok bool
|
||||
|
||||
clog := n.Logger
|
||||
|
||||
cachedExprEnv := expressionEnv
|
||||
|
||||
clog.Tracef("Event entering node")
|
||||
|
||||
NodeState, err := n.processFilter(cachedExprEnv)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
if !NodeState {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
if n.Name != "" {
|
||||
NodesHits.With(prometheus.Labels{"source": p.Line.Src, "type": p.Line.Module, "name": n.Name}).Inc()
|
||||
}
|
||||
|
||||
isWhitelisted, err := n.processWhitelist(cachedExprEnv, p)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
// Process grok if present, should be exclusive with nodes :)
|
||||
gstr := ""
|
||||
|
||||
if n.Grok.RunTimeRegexp != nil {
|
||||
clog.Tracef("Processing grok pattern : %s : %p", n.Grok.RegexpName, n.Grok.RunTimeRegexp)
|
||||
// for unparsed, parsed etc. set sensible defaults to reduce user hassle
|
||||
|
@ -211,6 +251,7 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
clog.Warningf("failed to run RunTimeValue : %v", err)
|
||||
NodeState = false
|
||||
}
|
||||
|
||||
switch out := output.(type) {
|
||||
case string:
|
||||
gstr = out
|
||||
|
@ -229,10 +270,12 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
} else {
|
||||
groklabel = n.Grok.RegexpName
|
||||
}
|
||||
|
||||
grok := n.Grok.RunTimeRegexp.Parse(gstr)
|
||||
if len(grok) > 0 {
|
||||
/*tag explicitly that the *current* node had a successful grok pattern. it's important to know success state*/
|
||||
NodeHasOKGrok = true
|
||||
|
||||
clog.Debugf("+ Grok '%s' returned %d entries to merge in Parsed", groklabel, len(grok))
|
||||
// We managed to grok stuff, merged into parse
|
||||
for k, v := range grok {
|
||||
|
@ -250,7 +293,6 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
clog.Debugf("+ Grok '%s' didn't return data on '%s'", groklabel, gstr)
|
||||
NodeState = false
|
||||
}
|
||||
|
||||
} else {
|
||||
clog.Tracef("! No grok pattern : %p", n.Grok.RunTimeRegexp)
|
||||
}
|
||||
|
@ -258,12 +300,16 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
// Process the stash (data collection) if : a grok was present and succeeded, or if there is no grok
|
||||
if NodeHasOKGrok || n.Grok.RunTimeRegexp == nil {
|
||||
for idx, stash := range n.Stash {
|
||||
var value string
|
||||
var key string
|
||||
var (
|
||||
key string
|
||||
value string
|
||||
)
|
||||
|
||||
if stash.ValueExpression == nil {
|
||||
clog.Warningf("Stash %d has no value expression, skipping", idx)
|
||||
continue
|
||||
}
|
||||
|
||||
if stash.KeyExpression == nil {
|
||||
clog.Warningf("Stash %d has no key expression, skipping", idx)
|
||||
continue
|
||||
|
@ -307,7 +353,9 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
clog.Debugf("Event leaving node : ko")
|
||||
return false, err
|
||||
}
|
||||
|
||||
clog.Tracef("\tsub-node (%s) ret : %v (strategy:%s)", leaf.rn, ret, n.OnSuccess)
|
||||
|
||||
if ret {
|
||||
NodeState = true
|
||||
/* if child is successful, stop processing */
|
||||
|
@ -333,7 +381,9 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
if n.Name != "" {
|
||||
NodesHitsKo.With(prometheus.Labels{"source": p.Line.Src, "type": p.Line.Module, "name": n.Name}).Inc()
|
||||
}
|
||||
|
||||
clog.Debugf("Event leaving node : ko")
|
||||
|
||||
return NodeState, nil
|
||||
}
|
||||
|
||||
|
@ -360,6 +410,7 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
if NodeState {
|
||||
clog.Debugf("Event leaving node : ok")
|
||||
log.Tracef("node is successful, check strategy")
|
||||
|
||||
if n.OnSuccess == "next_stage" {
|
||||
idx := stageidx(p.Stage, ctx.Stages)
|
||||
// we're at the last stage
|
||||
|
@ -375,15 +426,16 @@ func (n *Node) process(p *types.Event, ctx UnixParserCtx, expressionEnv map[stri
|
|||
} else {
|
||||
clog.Debugf("Event leaving node : ko")
|
||||
}
|
||||
|
||||
clog.Tracef("Node successful, continue")
|
||||
|
||||
return NodeState, nil
|
||||
}
|
||||
|
||||
func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
||||
var err error
|
||||
var valid bool
|
||||
|
||||
valid = false
|
||||
valid := false
|
||||
|
||||
dumpr := spew.ConfigState{MaxDepth: 1, DisablePointerAddresses: true}
|
||||
n.rn = seed.Generate()
|
||||
|
@ -393,10 +445,11 @@ func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
/* if the node has debugging enabled, create a specific logger with debug
|
||||
that will be used only for processing this node ;) */
|
||||
if n.Debug {
|
||||
var clog = log.New()
|
||||
clog := log.New()
|
||||
if err = types.ConfigureLogger(clog); err != nil {
|
||||
log.Fatalf("While creating bucket-specific logger : %s", err)
|
||||
}
|
||||
|
||||
clog.SetLevel(log.DebugLevel)
|
||||
n.Logger = clog.WithFields(log.Fields{
|
||||
"id": n.rn,
|
||||
|
@ -425,12 +478,15 @@ func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
/* handle pattern_syntax and groks */
|
||||
for _, pattern := range n.SubGroks {
|
||||
n.Logger.Tracef("Adding subpattern '%s' : '%s'", pattern.Key, pattern.Value)
|
||||
|
||||
if err = pctx.Grok.Add(pattern.Key.(string), pattern.Value.(string)); err != nil {
|
||||
if errors.Is(err, grokky.ErrAlreadyExist) {
|
||||
n.Logger.Warningf("grok '%s' already registred", pattern.Key)
|
||||
continue
|
||||
}
|
||||
|
||||
n.Logger.Errorf("Unable to compile subpattern %s : %v", pattern.Key, err)
|
||||
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
@ -438,28 +494,36 @@ func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
/* load grok by name or compile in-place */
|
||||
if n.Grok.RegexpName != "" {
|
||||
n.Logger.Tracef("+ Regexp Compilation '%s'", n.Grok.RegexpName)
|
||||
|
||||
n.Grok.RunTimeRegexp, err = pctx.Grok.Get(n.Grok.RegexpName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("unable to find grok '%s' : %v", n.Grok.RegexpName, err)
|
||||
}
|
||||
|
||||
if n.Grok.RunTimeRegexp == nil {
|
||||
return fmt.Errorf("empty grok '%s'", n.Grok.RegexpName)
|
||||
}
|
||||
|
||||
n.Logger.Tracef("%s regexp: %s", n.Grok.RegexpName, n.Grok.RunTimeRegexp.String())
|
||||
|
||||
valid = true
|
||||
} else if n.Grok.RegexpValue != "" {
|
||||
if strings.HasSuffix(n.Grok.RegexpValue, "\n") {
|
||||
n.Logger.Debugf("Beware, pattern ends with \\n : '%s'", n.Grok.RegexpValue)
|
||||
}
|
||||
|
||||
n.Grok.RunTimeRegexp, err = pctx.Grok.Compile(n.Grok.RegexpValue)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to compile grok '%s': %v", n.Grok.RegexpValue, err)
|
||||
}
|
||||
|
||||
if n.Grok.RunTimeRegexp == nil {
|
||||
// We shouldn't be here because compilation succeeded, so regexp shouldn't be nil
|
||||
return fmt.Errorf("grok compilation failure: %s", n.Grok.RegexpValue)
|
||||
}
|
||||
|
||||
n.Logger.Tracef("%s regexp : %s", n.Grok.RegexpValue, n.Grok.RunTimeRegexp.String())
|
||||
|
||||
valid = true
|
||||
}
|
||||
|
||||
|
@ -482,6 +546,7 @@ func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
return err
|
||||
}
|
||||
}
|
||||
|
||||
valid = true
|
||||
}
|
||||
|
||||
|
@ -526,14 +591,18 @@ func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
if !n.LeavesNodes[idx].Debug && n.Debug {
|
||||
n.LeavesNodes[idx].Debug = true
|
||||
}
|
||||
|
||||
if !n.LeavesNodes[idx].Profiling && n.Profiling {
|
||||
n.LeavesNodes[idx].Profiling = true
|
||||
}
|
||||
|
||||
n.LeavesNodes[idx].Stage = n.Stage
|
||||
|
||||
err = n.LeavesNodes[idx].compile(pctx, ectx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
valid = true
|
||||
}
|
||||
|
||||
|
@ -546,6 +615,7 @@ func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
return err
|
||||
}
|
||||
}
|
||||
|
||||
valid = true
|
||||
}
|
||||
|
||||
|
@ -554,13 +624,15 @@ func (n *Node) compile(pctx *UnixParserCtx, ectx EnricherCtx) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
valid = valid || whitelistValid
|
||||
|
||||
if !valid {
|
||||
/* node is empty, error force return */
|
||||
n.Logger.Error("Node is empty or invalid, abort")
|
||||
n.Stage = ""
|
||||
return fmt.Errorf("Node is empty")
|
||||
|
||||
return errors.New("Node is empty")
|
||||
}
|
||||
|
||||
if err := n.validate(pctx, ectx); err != nil {
|
||||
|
|
Loading…
Reference in a new issue