ClusterCockpit/cc-metric-collector
1
0
Fork 0
mirror of https://github.com/ClusterCockpit/cc-metric-collector.git synced 2024-12-26 15:29:04 +01:00
Code Issues Packages Projects Releases Wiki Activity

Wait for concurrent flush operations to finish

Browse Source
This commit is contained in:
Holger Obermaier 2023-10-05 16:44:03 +02:00
parent b0f0462995
commit 9dae829f9d
1 changed files with 16 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
sinks/httpSink.go
View File

@ -75,6 +75,9 @@ type HttpSink struct {
// Lock to assure that only one timer is running at a time // Lock to assure that only one timer is running at a time
timerLock sync.Mutex timerLock sync.Mutex
// WaitGroup for concurrent flush operations
flushWaitGroup sync.WaitGroup
config HttpSinkConfig config HttpSinkConfig
} }
@ -206,6 +209,9 @@ func (s *HttpSink) Write(m lp.CCMetric) error {
} }
func (s *HttpSink) Flush() error { func (s *HttpSink) Flush() error {
s.flushWaitGroup.Add(1)
defer s.flushWaitGroup.Done()
// Lock for encoder usage // Lock for encoder usage
// Own lock for as short as possible: the time it takes to clone the buffer. // Own lock for as short as possible: the time it takes to clone the buffer.
s.encoderLock.Lock() s.encoderLock.Lock()
@ -270,10 +276,19 @@ func (s *HttpSink) Flush() error {
} }
func (s *HttpSink) Close() { func (s *HttpSink) Close() {
s.flushTimer.Stop() // Stop existing timer and immediately flush
if s.flushTimer != nil {
if ok := s.flushTimer.Stop(); ok {
s.timerLock.Unlock()
}
}
if err := s.Flush(); err != nil { if err := s.Flush(); err != nil {
cclog.ComponentError(s.name, "flush failed:", err.Error()) cclog.ComponentError(s.name, "flush failed:", err.Error())
} }
// Wait for flush operations to finish
s.flushWaitGroup.Wait()
s.client.CloseIdleConnections() s.client.CloseIdleConnections()
} }