From 708e145020640aea7a3196bbabc0f55e571415ed Mon Sep 17 00:00:00 2001 From: Thomas Roehl Date: Fri, 20 Dec 2024 17:07:16 +0100 Subject: [PATCH] natsSink: Use flush timer handling from httpSink and some comments --- sinks/natsSink.go | 82 +++++++++++++++++++++++++++++++++-------------- sinks/natsSink.md | 2 +- 2 files changed, 59 insertions(+), 25 deletions(-) diff --git a/sinks/natsSink.go b/sinks/natsSink.go index b2f0b18..08cb4d1 100644 --- a/sinks/natsSink.go +++ b/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() - }) - } else { - s.flushTimer.Reset(s.flushDelay) + 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 { + + // 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) } } diff --git a/sinks/natsSink.md b/sinks/natsSink.md index 8bafbb0..ef536d4 100644 --- a/sinks/natsSink.md +++ b/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)