Fix races & refactor journalctl plugin

ozontech · vadimalekseev · Jul 31, 2023 · Jul 25, 2023 · Jul 25, 2023 · Jul 26, 2023
commit f06730eb5da5cb3932e80f51e0d14aae927603b0
diff --git a/plugin/input/journalctl/journalctl.go b/plugin/input/journalctl/journalctl.go
@@ -3,22 +3,27 @@
 package journalctl
 
 import (
+	"sync/atomic"
+
 	"github.com/ozontech/file.d/fd"
 	"github.com/ozontech/file.d/metric"
 	"github.com/ozontech/file.d/offset"
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/zap"
 )
 
 /*{ introduction
 Reads `journalctl` output.
 }*/
 
 type Plugin struct {
-	params  *pipeline.InputPluginParams
-	config  *Config
-	reader  *journalReader
-	offInfo *offsetInfo
+	params        *pipeline.InputPluginParams
+	config        *Config
+	reader        *journalReader
+	offInfo       atomic.Pointer[offsetInfo]
+	currentOffset int64
+	logger        *zap.Logger
 
 	//  plugin metrics
 
@@ -51,8 +56,6 @@ type Config struct {
 type offsetInfo struct {
 	Offset int64  `json:"offset"`
 	Cursor string `json:"cursor"`
-
-	current int64
 }
 
 func (o *offsetInfo) set(cursor string) {
@@ -61,8 +64,8 @@ func (o *offsetInfo) set(cursor string) {
 }
 
 func (p *Plugin) Write(bytes []byte) (int, error) {
-	p.params.Controller.In(0, "journalctl", p.offInfo.current, bytes, false)
-	p.offInfo.current++
+	p.params.Controller.In(0, "journalctl", p.currentOffset, bytes, false)
+	p.currentOffset++
 	return len(bytes), nil
 }
 
@@ -80,24 +83,26 @@ func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig) {
 func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.InputPluginParams) {
 	p.params = params
 	p.config = config.(*Config)
+	p.logger = params.Logger.Desugar()
 	p.registerMetrics(params.MetricCtl)
 
-	p.offInfo = &offsetInfo{}
-	if err := offset.LoadYAML(p.config.OffsetsFile, p.offInfo); err != nil {
+	offInfo := &offsetInfo{}
+	if err := offset.LoadYAML(p.config.OffsetsFile, offInfo); err != nil {
 		p.offsetErrorsMetric.WithLabelValues().Inc()
-		p.params.Logger.Error("can't load offset file: %s", err.Error())
+		p.logger.Error("can't load offset file", zap.Error(err))
 	}
+	p.offInfo.Store(offInfo)
 
 	readConfig := &journalReaderConfig{
 		output:   p,
-		cursor:   p.offInfo.Cursor,
+		cursor:   offInfo.Cursor,
 		maxLines: p.config.MaxLines,
-		logger:   p.params.Logger,
+		logger:   p.logger,
 	}
 	p.reader = newJournalReader(readConfig, p.readerErrorsMetric)
 	p.reader.args = append(p.reader.args, p.config.JournalArgs...)
 	if err := p.reader.start(); err != nil {
-		p.params.Logger.Fatal("failure during start: %s", err.Error())
+		p.logger.Fatal("failure during start", zap.Error(err))
 	}
 }
 
@@ -111,21 +116,24 @@ func (p *Plugin) Stop() {
 	err := p.reader.stop()
 	if err != nil {
 		p.journalCtlStopErrorMetric.WithLabelValues().Inc()
-		p.params.Logger.Error("can't stop journalctl cmd: %s", err.Error())
+		p.logger.Error("can't stop journalctl cmd", zap.Error(err))
 	}
 
-	if err := offset.SaveYAML(p.config.OffsetsFile, p.offInfo); err != nil {
+	offsets := *p.offInfo.Load()
+	if err := offset.SaveYAML(p.config.OffsetsFile, offsets); err != nil {
 		p.offsetErrorsMetric.WithLabelValues().Inc()
-		p.params.Logger.Error("can't save offset file: %s", err.Error())
+		p.logger.Error("can't save offset file", zap.Error(err))
 	}
 }
 
 func (p *Plugin) Commit(event *pipeline.Event) {
-	p.offInfo.set(pipeline.CloneString(event.Root.Dig("__CURSOR").AsString()))
+	offInfo := *p.offInfo.Load()
+	offInfo.set(pipeline.CloneString(event.Root.Dig("__CURSOR").AsString()))
+	p.offInfo.Store(&offInfo)
 
-	if err := offset.SaveYAML(p.config.OffsetsFile, p.offInfo); err != nil {
+	if err := offset.SaveYAML(p.config.OffsetsFile, offInfo); err != nil {
 		p.offsetErrorsMetric.WithLabelValues().Inc()
-		p.params.Logger.Error("can't save offset file: %s", err.Error())
+		p.logger.Error("can't save offset file", zap.Error(err))
 	}
 }
 

diff --git a/plugin/input/journalctl/reader.go b/plugin/input/journalctl/reader.go
@@ -6,20 +6,17 @@ import (
 	"os/exec"
 	"strings"
 
-	"github.com/ozontech/file.d/logger"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
 
-//nolint:unused
 type journalReaderConfig struct {
 	output   io.Writer
 	cursor   string
-	logger   *zap.SugaredLogger
+	logger   *zap.Logger
 	maxLines int
 }
 
-//nolint:unused
 type journalReader struct {
 	config *journalReaderConfig
 	cmd    *exec.Cmd
@@ -29,7 +26,6 @@ type journalReader struct {
 	readerErrorsMetric *prometheus.CounterVec
 }
 
-//nolint:unused
 func (r *journalReader) readLines(rd io.Reader, config *journalReaderConfig) {
 	reader := bufio.NewReaderSize(rd, 1024*1024*10) // max message size
 	totalLines := 0
@@ -39,7 +35,7 @@ func (r *journalReader) readLines(rd io.Reader, config *journalReaderConfig) {
 	if config.cursor != "" {
 		_, _, err := reader.ReadLine()
 		if err != nil {
-			logger.Fatalf(err.Error())
+			r.config.logger.Fatal(err.Error())
 		}
 	}
 
@@ -50,13 +46,13 @@ func (r *journalReader) readLines(rd io.Reader, config *journalReaderConfig) {
 		}
 		if err != nil {
 			r.readerErrorsMetric.WithLabelValues().Inc()
-			config.logger.Error(err)
+			r.config.logger.Error(err.Error())
 			continue
 		}
 		_, err = config.output.Write(bytes)
 		if err != nil {
 			r.readerErrorsMetric.WithLabelValues().Inc()
-			config.logger.Error(err)
+			r.config.logger.Error(err.Error())
 		}
 
 		totalLines++
@@ -66,7 +62,6 @@ func (r *journalReader) readLines(rd io.Reader, config *journalReaderConfig) {
 	}
 }
 
-//nolint:deadcode,unused
 func newJournalReader(config *journalReaderConfig, readerErrorsCounter *prometheus.CounterVec) *journalReader {
 	res := &journalReader{
 		config:             config,
@@ -83,9 +78,8 @@ func newJournalReader(config *journalReaderConfig, readerErrorsCounter *promethe
 	return res
 }
 
-//nolint:unused
 func (r *journalReader) start() error {
-	r.config.logger.Infof(`running "journalctl %s"`, strings.Join(r.args, " "))
+	r.config.logger.Info(`running journalctl`, zap.String("args", strings.Join(r.args, " ")))
 	r.cmd = exec.Command("journalctl", r.args...)
 
 	out, err := r.cmd.StdoutPipe()
@@ -102,7 +96,6 @@ func (r *journalReader) start() error {
 	return nil
 }
 
-//nolint:unused
 func (r *journalReader) stop() error {
 	return r.cmd.Process.Kill()
 }