natsSink: Use flush timer handling from httpSink and some comments

sinks/natsSink.go

@@ -25,6 +25,7 @@ type NatsSinkConfig struct {
 	User       string `json:"user,omitempty"`
 	Password   string `json:"password,omitempty"`
 	FlushDelay string `json:"flush_delay,omitempty"`
+	flushDelay time.Duration
 	NkeyFile   string `json:"nkey_file,omitempty"`
 	// Timestamp precision
 	Precision string `json:"precision,omitempty"`
@@ -37,9 +38,8 @@ type NatsSink struct {
 	encoderLock sync.Mutex
 	config      NatsSinkConfig
 
-	flushDelay time.Duration
 	flushTimer *time.Timer
-	//timerLock  sync.Mutex
+	timerLock  sync.Mutex
 }
 
 func (s *NatsSink) connect() error {
@@ -75,39 +75,67 @@ func (s *NatsSink) connect() error {
 func (s *NatsSink) Write(m lp.CCMessage) error {
 	msg, err := s.mp.ProcessMessage(m)
 	if err == nil && msg != nil {
+		// Lock for encoder usage
 		s.encoderLock.Lock()
 
-		err = EncoderAdd(&s.encoder, msg)
+		// Add message to encoder
+		err = EncoderAdd(&s.encoder, m)
 
+		// Unlock encoder usage
 		s.encoderLock.Unlock()
+
+		// Check that encoding worked
 		if err != nil {
 			cclog.ComponentError(s.name, "Write:", err.Error())
 			return err
 		}
 	}
 
-	if s.flushDelay == 0 {
-		s.Flush()
-	} else if s.flushTimer == nil {
-		s.flushTimer = time.AfterFunc(s.flushDelay, func() {
-			s.Flush()
-		})
+	if s.config.flushDelay == 0 {
+		// Directly flush if no flush delay is configured
+		return s.Flush()
+	} else if s.timerLock.TryLock() {
+		// Setup flush timer when flush delay is configured
+		// and no other timer is already running
+		if s.flushTimer != nil {
+
+			// Restarting existing flush timer
+			cclog.ComponentDebug(s.name, "Write(): Restarting flush timer")
+			s.flushTimer.Reset(s.config.flushDelay)
 		} else {
-		s.flushTimer.Reset(s.flushDelay)
+
+			// Creating and starting flush timer
+			cclog.ComponentDebug(s.name, "Write(): Starting new flush timer")
+			s.flushTimer = time.AfterFunc(
+				s.config.flushDelay,
+				func() {
+					defer s.timerLock.Unlock()
+					cclog.ComponentDebug(s.name, "Starting flush triggered by flush timer")
+					if err := s.Flush(); err != nil {
+						cclog.ComponentError(s.name, "Flush triggered by flush timer: flush failed:", err)
+					}
+				})
+		}
 	}
 
 	return nil
 }
 
 func (s *NatsSink) Flush() error {
+	// Lock for encoder usage
+	// Own lock for as short as possible: the time it takes to clone the buffer.
 	s.encoderLock.Lock()
+
 	buf := slices.Clone(s.encoder.Bytes())
 	s.encoder.Reset()
+
+	// Unlock encoder usage
 	s.encoderLock.Unlock()
 
 	if len(buf) == 0 {
 		return nil
 	}
+
 	if err := s.client.Publish(s.config.Subject, buf); err != nil {
 		cclog.ComponentError(s.name, "Flush:", err.Error())
 		return err
@@ -116,14 +144,21 @@ func (s *NatsSink) Flush() error {
 }
 
 func (s *NatsSink) Close() {
-	cclog.ComponentDebug(s.name, "Close")
+	// Stop existing timer and immediately flush
+	if s.flushTimer != nil {
+		if ok := s.flushTimer.Stop(); ok {
+			s.timerLock.Unlock()
+		}
+	}
+	cclog.ComponentDebug(s.name, "Close NATS connection")
 	s.client.Close()
 }
 
 func NewNatsSink(name string, config json.RawMessage) (Sink, error) {
 	s := new(NatsSink)
 	s.name = fmt.Sprintf("NatsSink(%s)", name)
-	s.flushDelay = 10 * time.Second
+	s.config.flushDelay = 5 * time.Second
+	s.config.FlushDelay = "5s"
 	s.config.Port = "4222"
 	s.config.Precision = "s"
 	if len(config) > 0 {
@@ -139,21 +174,25 @@ func NewNatsSink(name string, config json.RawMessage) (Sink, error) {
 		len(s.config.Subject) == 0 {
 		return nil, errors.New("not all configuration variables set required by NatsSink")
 	}
+	// Create a new message processor
 	p, err := mp.NewMessageProcessor()
 	if err != nil {
 		return nil, fmt.Errorf("initialization of message processor failed: %v", err.Error())
 	}
 	s.mp = p
+	// Read config related to message processor
 	if len(s.config.MessageProcessor) > 0 {
 		err = s.mp.FromConfigJSON(s.config.MessageProcessor)
 		if err != nil {
 			return nil, fmt.Errorf("failed parsing JSON for message processor: %v", err.Error())
 		}
 	}
-	// Create lookup map to use meta infos as tags in the output metric
+	// Add meta_as_tags list to message processor
 	for _, k := range s.config.MetaAsTags {
 		s.mp.AddMoveMetaToTags("true", k, k)
 	}
+
+	// Setup Influx line protocol encoder
 	precision := influx.Second
 	if len(s.config.Precision) > 0 {
 		switch s.config.Precision {
@@ -168,11 +207,6 @@ func NewNatsSink(name string, config json.RawMessage) (Sink, error) {
 		}
 	}
 
-	// s.meta_as_tags = make(map[string]bool)
-	// for _, k := range s.config.MetaAsTags {
-	// 	s.meta_as_tags[k] = true
-	// }
-	// Setup Influx line protocol
 	s.encoder.SetPrecision(precision)
 	// Setup infos for connection
 	if err := s.connect(); err != nil {
@@ -180,11 +214,11 @@ func NewNatsSink(name string, config json.RawMessage) (Sink, error) {
 	}
 
 	s.flushTimer = nil
-	if len(s.config.FlushDelay) != 0 {
-		var err error
-		s.flushDelay, err = time.ParseDuration(s.config.FlushDelay)
-		if err != nil {
-			return nil, err
+	if len(s.config.FlushDelay) > 0 {
+		t, err := time.ParseDuration(s.config.FlushDelay)
+		if err == nil {
+			s.config.flushDelay = t
+			cclog.ComponentDebug(s.name, "Init(): flushDelay", t)
 		}
 	}
 

sinks/natsSink.md

@@ -31,7 +31,7 @@ The `nats` sink publishes all metrics into a NATS network. The publishing key is
 - `user`: Username for basic authentication
 - `password`: Password for basic authentication
 - `nkey_file`: Path to credentials file with NKEY
-- `flush_delay`: Maximum time until metrics are sent out
+- `flush_delay`: Maximum time until metrics are sent out (default '5s')
 - `precision`: Precision of the timestamp. Valid values are 's', 'ms', 'us' and 'ns'. (default is 's')
 - `process_messages`: Process messages with given rules before progressing or dropping, see [here](../pkg/messageProcessor/README.md)  (optional)
 - `meta_as_tags`: print all meta information as tags in the output (deprecated, optional)