add sample logger for parse_re2 and discard plugins

Author: DSmolonogov

logger/logger.go

@@ -3,15 +3,22 @@ package logger
 import (
 	"os"
 	"strings"
+	"time"
 
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 )
 
 var Instance *zap.SugaredLogger
+var SampleInstance *zap.SugaredLogger
 var Level zap.AtomicLevel
 
-const defaultLevel = zap.InfoLevel
+const (
+	defaultLevel      = zap.InfoLevel
+	defaultTick       = time.Second
+	defaultFirst      = 10
+	defaultThereAfter = 100
+)
 
 func init() {
 	var level zapcore.Level
@@ -32,27 +39,44 @@ func init() {
 
 	Level = zap.NewAtomicLevelAt(level)
 
+	core := zapcore.NewCore(
+		zapcore.NewConsoleEncoder(zapcore.EncoderConfig{
+			TimeKey:        "ts",
+			LevelKey:       "level",
+			NameKey:        "Instance",
+			CallerKey:      "caller",
+			MessageKey:     "message",
+			StacktraceKey:  "stacktrace",
+			LineEnding:     zapcore.DefaultLineEnding,
+			EncodeLevel:    zapcore.LowercaseLevelEncoder,
+			EncodeTime:     zapcore.ISO8601TimeEncoder,
+			EncodeDuration: zapcore.SecondsDurationEncoder,
+			EncodeCaller:   zapcore.ShortCallerEncoder,
+		}),
+		zapcore.AddSync(os.Stdout),
+		Level,
+	)
+
+	sampleCore := zapcore.NewSamplerWithOptions(
+		core,
+		defaultTick,
+		defaultFirst,
+		defaultThereAfter,
+	)
+
+	// logger initialization
 	Instance = zap.New(
-		zapcore.NewCore(
-			zapcore.NewConsoleEncoder(zapcore.EncoderConfig{
-				TimeKey:        "ts",
-				LevelKey:       "level",
-				NameKey:        "Instance",
-				CallerKey:      "caller",
-				MessageKey:     "message",
-				StacktraceKey:  "stacktrace",
-				LineEnding:     zapcore.DefaultLineEnding,
-				EncodeLevel:    zapcore.LowercaseLevelEncoder,
-				EncodeTime:     zapcore.ISO8601TimeEncoder,
-				EncodeDuration: zapcore.SecondsDurationEncoder,
-				EncodeCaller:   zapcore.ShortCallerEncoder,
-			}),
-			zapcore.AddSync(os.Stdout),
-			Level,
-		),
+		core,
 	).Sugar().Named("fd")
 
 	Instance.Infof("Logger initialized with level: %s", level)
+
+	// sample logger initialization
+	SampleInstance = zap.New(
+		sampleCore,
+	).Sugar().Named("fd")
+
+	SampleInstance.Infof("SampleLogger initialized with level: %s", level)
 }
 
 func Debug(args ...any) {

pipeline/pipeline.go

@@ -99,6 +99,7 @@ type Pipeline struct {
 
 	// some debugging stuff
 	logger          *zap.SugaredLogger
+	sampleLogger    *zap.SugaredLogger
 	eventLogEnabled bool
 	eventLog        []string
 	eventLogMu      *sync.Mutex
@@ -143,6 +144,7 @@ func New(name string, settings *Settings, registry *prometheus.Registry) *Pipeli
 	pipeline := &Pipeline{
 		Name:           name,
 		logger:         logger.Instance.Named(name),
+		sampleLogger:   logger.SampleInstance.Named(name),
 		settings:       settings,
 		useSpread:      false,
 		disableStreams: false,
@@ -266,7 +268,7 @@ func (p *Pipeline) Start() {
 	p.logger.Infof("stating processors, count=%d", len(p.Procs))
 	for _, processor := range p.Procs {
 		processor.registerMetrics(p.metricsCtl)
-		processor.start(p.actionParams, p.logger)
+		processor.start(p.actionParams, p.logger, p.sampleLogger)
 	}
 
 	p.logger.Infof("starting input plugin %q", p.inputInfo.Type)
@@ -571,7 +573,7 @@ func (p *Pipeline) expandProcs() {
 		proc := p.newProc()
 		p.Procs = append(p.Procs, proc)
 		proc.registerMetrics(p.metricsCtl)
-		proc.start(p.actionParams, p.logger)
+		proc.start(p.actionParams, p.logger, p.sampleLogger)
 	}
 
 	p.procCount.Swap(to)

pipeline/plugin.go

@@ -52,8 +52,9 @@ type PluginDefaultParams struct {
 
 type ActionPluginParams struct {
 	*PluginDefaultParams
-	Controller ActionPluginController
-	Logger     *zap.SugaredLogger
+	Controller   ActionPluginController
+	Logger       *zap.SugaredLogger
+	SampleLogger *zap.SugaredLogger
 }
 
 type OutputPluginParams struct {

pipeline/processor.go

@@ -94,13 +94,14 @@ func NewProcessor(
 	return processor
 }
 
-func (p *processor) start(params *PluginDefaultParams, logger *zap.SugaredLogger) {
+func (p *processor) start(params *PluginDefaultParams, logger *zap.SugaredLogger, sampleLogger *zap.SugaredLogger) {
 	for i, action := range p.actions {
 		actionInfo := p.actionInfos[i]
 		action.Start(actionInfo.PluginStaticInfo.Config, &ActionPluginParams{
 			PluginDefaultParams: params,
 			Controller:          p,
 			Logger:              logger.Named("action").Named(actionInfo.Type),
+			SampleLogger:        sampleLogger.Named("action").Named(actionInfo.Type),
 		})
 	}
 

plugin/action/discard/README.idoc.md

@@ -1,4 +1,5 @@
 # Discard plugin
 @introduction
 
-> No config params
+### Config params
+@config-params|description

plugin/action/discard/README.md

@@ -13,6 +13,12 @@ pipelines:
     ...
 ```
 
-> No config params
+### Config params
+**`is_logging`** *`bool`* *`default=false`* 
+
+Field that includes logging (with sampling).
+
+<br>
+
 
 <br>*Generated using [__insane-doc__](https://github.com/vitkovskii/insane-doc)*

plugin/action/discard/discard.go

@@ -4,6 +4,7 @@ import (
 	"github.com/ozontech/file.d/fd"
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/plugin"
+	"go.uber.org/zap"
 )
 
 /*{ introduction
@@ -23,10 +24,19 @@ pipelines:
 }*/
 
 type Plugin struct {
+	config       *Config
+	sampleLogger *zap.SugaredLogger
 	plugin.NoMetricsPlugin
 }
 
-type Config struct{}
+// ! config-params
+// ^ config-params
+type Config struct {
+	// > @3@4@5@6
+	// >
+	// > Field that includes logging (with sampling).
+	IsLogging bool `json:"is_logging" default:"false"` // *
+}
 
 func init() {
 	fd.DefaultPluginRegistry.RegisterAction(&pipeline.PluginStaticInfo{
@@ -39,12 +49,17 @@ func factory() (pipeline.AnyPlugin, pipeline.AnyConfig) {
 	return &Plugin{}, &Config{}
 }
 
-func (p *Plugin) Start(_ pipeline.AnyConfig, _ *pipeline.ActionPluginParams) {
+func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginParams) {
+	p.config = config.(*Config)
+	p.sampleLogger = params.SampleLogger
 }
 
 func (p *Plugin) Stop() {
 }
 
-func (p *Plugin) Do(_ *pipeline.Event) pipeline.ActionResult {
+func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult {
+	if p.config.IsLogging {
+		p.sampleLogger.Info("discarded event: ", zap.Stringer("json", event))
+	}
 	return pipeline.ActionDiscard
 }

plugin/action/mask/mask_test.go

@@ -179,7 +179,8 @@ func TestMaskAddExtraField(t *testing.T) {
 			PipelineName:     "test_pipeline",
 			PipelineSettings: &pipeline.Settings{},
 		},
-		Logger: zap.L().Sugar(),
+		Logger:       zap.L().Sugar(),
+		SampleLogger: zap.L().Sugar(),
 	})
 	plugin.config.Masks[0].Re_ = regexp.MustCompile(plugin.config.Masks[0].Re)
 

plugin/action/parse_re2/README.md

@@ -14,5 +14,11 @@ A prefix to add to decoded object keys.
 
 <br>
 
+**`is_logging`** *`bool`* *`default=false`* 
+
+Field that includes logging (with sampling).
+
+<br>
+
 
 <br>*Generated using [__insane-doc__](https://github.com/vitkovskii/insane-doc)*

plugin/action/parse_re2/parse_re2.go

@@ -8,16 +8,17 @@ import (
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/plugin"
 	insaneJSON "github.com/vitkovskii/insane-json"
+	"go.uber.org/zap"
 )
 
 /*{ introduction
 It parses string from the event field using re2 expression with named subgroups and merges the result with the event root.
 }*/
 
 type Plugin struct {
-	config *Config
-
-	re *regexp.Regexp
+	config       *Config
+	sampleLogger *zap.SugaredLogger
+	re           *regexp.Regexp
 	plugin.NoMetricsPlugin
 }
 
@@ -39,6 +40,11 @@ type Config struct {
 	// >
 	// > A prefix to add to decoded object keys.
 	Prefix string `json:"prefix" default:""` // *
+
+	// > @3@4@5@6
+	// >
+	// > Field that includes logging (with sampling).
+	IsLogging bool `json:"is_logging" default:"false"` // *
 }
 
 func init() {
@@ -52,8 +58,9 @@ func factory() (pipeline.AnyPlugin, pipeline.AnyConfig) {
 	return &Plugin{}, &Config{}
 }
 
-func (p *Plugin) Start(config pipeline.AnyConfig, _ *pipeline.ActionPluginParams) {
+func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginParams) {
 	p.config = config.(*Config)
+	p.sampleLogger = params.SampleLogger
 
 	p.re = regexp.MustCompile(p.config.Re2)
 }
@@ -70,6 +77,9 @@ func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult {
 	sm := p.re.FindSubmatch(jsonNode.AsBytes())
 
 	if len(sm) == 0 {
+		if p.config.IsLogging {
+			p.sampleLogger.Info("event is not parsed: ", zap.Stringer("json", event))
+		}
 		return pipeline.ActionPass
 	}
 

plugin/input/k8s/multiline_action_test.go

@@ -17,11 +17,15 @@ func TestMultilineAction_Do(t *testing.T) {
 	config := &Config{
 		SplitEventSize: predictionLookahead * 4,
 	}
-	plugin.Start(config, &pipeline.ActionPluginParams{Logger: zap.S(), PluginDefaultParams: &pipeline.PluginDefaultParams{
-		PipelineSettings: &pipeline.Settings{
-			MaxEventSize: 20,
+	plugin.Start(config, &pipeline.ActionPluginParams{
+		Logger:       zap.S(),
+		SampleLogger: zap.S(),
+		PluginDefaultParams: &pipeline.PluginDefaultParams{
+			PipelineSettings: &pipeline.Settings{
+				MaxEventSize: 20,
+			},
 		},
-	}})
+	})
 
 	item := &metaItem{
 		nodeName:      "node_1",
@@ -93,9 +97,13 @@ func TestMultilineAction_Do_shouldSplit(t *testing.T) {
 	config := &Config{
 		SplitEventSize: predictionLookahead * 4,
 	}
-	plugin.Start(config, &pipeline.ActionPluginParams{Logger: zap.S(), PluginDefaultParams: &pipeline.PluginDefaultParams{
-		PipelineSettings: &pipeline.Settings{},
-	}})
+	plugin.Start(config, &pipeline.ActionPluginParams{
+		Logger:       zap.S(),
+		SampleLogger: zap.S(),
+		PluginDefaultParams: &pipeline.PluginDefaultParams{
+			PipelineSettings: &pipeline.Settings{},
+		},
+	})
 
 	item := &metaItem{
 		nodeName:      "node_1",