From e1e00b56b62cdc3d37fca28716c362d7dfca5a3d Mon Sep 17 00:00:00 2001
From: Remy Mathieu <remy.mathieu@datadoghq.com>
Date: Fri, 10 Sep 2021 12:14:50 +0200
Subject: [PATCH 1/7] [sender] detect when a fork happened and reset resources
 accordingly

---
 lib/datadog/statsd/forwarder.rb            |  2 +-
 lib/datadog/statsd/message_buffer.rb       |  9 ++-
 lib/datadog/statsd/sender.rb               | 85 ++++++++++++++++------
 lib/datadog/statsd/single_thread_sender.rb | 37 +++++++++-
 spec/statsd/forwarder_spec.rb              |  4 +-
 5 files changed, 106 insertions(+), 31 deletions(-)

diff --git a/lib/datadog/statsd/forwarder.rb b/lib/datadog/statsd/forwarder.rb
index a7d9ad73..07b65450 100644
--- a/lib/datadog/statsd/forwarder.rb
+++ b/lib/datadog/statsd/forwarder.rb
@@ -55,7 +55,7 @@ def initialize(
           overflowing_stategy: buffer_overflowing_stategy,
         )
 
-        @sender = single_thread ? SingleThreadSender.new(buffer) : Sender.new(buffer)
+        @sender = single_thread ? SingleThreadSender.new(buffer) : Sender.new(buffer, logger: logger)
         @sender.start
       end
 
diff --git a/lib/datadog/statsd/message_buffer.rb b/lib/datadog/statsd/message_buffer.rb
index 841862f7..d3d200d1 100644
--- a/lib/datadog/statsd/message_buffer.rb
+++ b/lib/datadog/statsd/message_buffer.rb
@@ -42,13 +42,16 @@ def add(message)
         true
       end
 
+      def reset
+        buffer.clear
+        @message_count = 0
+      end
+
       def flush
         return if buffer.empty?
 
         connection.write(buffer)
-
-        buffer.clear
-        @message_count = 0
+        reset
       end
 
       private
diff --git a/lib/datadog/statsd/sender.rb b/lib/datadog/statsd/sender.rb
index 5818b627..a41c46e9 100644
--- a/lib/datadog/statsd/sender.rb
+++ b/lib/datadog/statsd/sender.rb
@@ -2,43 +2,81 @@
 
 module Datadog
   class Statsd
+    # Sender is using a companion thread to flush and pack messages
+    # in a `MessageBuffer`.
+    # The communication with this thread is done using a `Queue`.
+    # If the thread is dead, it is starting a new one to avoid having a blocked
+    # Sender with no companion thread to communicate with (most of the time, having
+    # a dead companion thread means that a fork just happened and that we are
+    # running in the child process).
     class Sender
       CLOSEABLE_QUEUES = Queue.instance_methods.include?(:close)
 
-      def initialize(message_buffer)
+      def initialize(message_buffer, logger: nil)
         @message_buffer = message_buffer
+        @logger = logger
+        @mx = Mutex.new
       end
 
       def flush(sync: false)
-        # don't try to flush if there is no message_queue instantiated
-        return unless message_queue
+        # don't try to flush if there is no message_queue instantiated or
+        # no companion thread running
+        if !message_queue
+          @logger.debug { "Statsd: can't flush: no message queue ready" } if @logger
+          return
+        end
+        if !sender_thread.alive?
+          @logger.debug { "Statsd: can't flush: no sender_thread alive" } if @logger
+          return
+        end
 
-        message_queue.push(:flush)
+        @mx.synchronize {
+          message_queue.push(:flush)
+        }
 
         rendez_vous if sync
       end
 
       def rendez_vous
-        # Initialize and get the thread's sync queue
-        queue = (Thread.current[:statsd_sync_queue] ||= Queue.new)
-        # tell sender-thread to notify us in the current
-        # thread's queue
-        message_queue.push(queue)
-        # wait for the sender thread to send a message
-        # once the flush is done
-        queue.pop
+        @mx.synchronize {
+          # Initialize and get the thread's sync queue
+          queue = (Thread.current[:statsd_sync_queue] ||= Queue.new)
+          # tell sender-thread to notify us in the current
+          # thread's queue
+          message_queue.push(queue)
+          # wait for the sender thread to send a message
+          # once the flush is done
+          queue.pop
+        }
       end
 
       def add(message)
         raise ArgumentError, 'Start sender first' unless message_queue
 
+        # if the thread does not exist, we assume we are running in a forked process,
+        # empty the message queue and message buffers (these messages belong to
+        # the parent process) and spawn a new companion thread.
+        if !sender_thread.alive?
+          @logger.debug { "Statsd: companion thread is dead, re-creating one" } if @logger
+          @mx.synchronize {
+            # a call from another thread has already re-created
+            # the companion thread before this one acquired the lock
+            break if sender_thread.alive?
+
+            message_queue.close if CLOSEABLE_QUEUES
+            @message_queue = nil
+            message_buffer.reset
+            start
+          }
+        end
+
         message_queue << message
       end
 
       def start
         raise ArgumentError, 'Sender already started' if message_queue
 
-        # initialize message queue for background thread
+        # initialize a new message queue for the background thread
         @message_queue = Queue.new
         # start background thread
         @sender_thread = Thread.new(&method(:send_loop))
@@ -46,26 +84,29 @@ def start
 
       if CLOSEABLE_QUEUES
         def stop(join_worker: true)
-          message_queue = @message_queue
-          message_queue.close if message_queue
+          @mx.synchronize {
+            message_queue = @message_queue
+            message_queue.close if message_queue
 
-          sender_thread = @sender_thread
-          sender_thread.join if sender_thread && join_worker
+            sender_thread = @sender_thread
+            sender_thread.join if sender_thread && join_worker
+          }
         end
       else
         def stop(join_worker: true)
-          message_queue = @message_queue
-          message_queue << :close if message_queue
+          @mx.synchronize {
+            message_queue = @message_queue
+            message_queue << :close if message_queue
 
-          sender_thread = @sender_thread
-          sender_thread.join if sender_thread && join_worker
+            sender_thread = @sender_thread
+            sender_thread.join if sender_thread && join_worker
+          }
         end
       end
 
       private
 
       attr_reader :message_buffer
-
       attr_reader :message_queue
       attr_reader :sender_thread
 
diff --git a/lib/datadog/statsd/single_thread_sender.rb b/lib/datadog/statsd/single_thread_sender.rb
index 5bcbc041..d82dc676 100644
--- a/lib/datadog/statsd/single_thread_sender.rb
+++ b/lib/datadog/statsd/single_thread_sender.rb
@@ -2,17 +2,35 @@
 
 module Datadog
   class Statsd
+    # The SingleThreadSender is a sender synchronously buffering messages
+    # in a `MessageBuffer`.
+    # It is using current Process.PID to check it is the result of a recent fork
+    # and it is reseting the MessageBuffer if that's the case.
     class SingleThreadSender
-      def initialize(message_buffer)
+      def initialize(message_buffer, logger: nil)
         @message_buffer = message_buffer
+        @logger = logger
+        @mx = Mutex.new
+        # store the pid for which this sender has been created
+        update_fork_pid
       end
 
       def add(message)
-        @message_buffer.add(message)
+        @mx.synchronize {
+          # we have just forked, meaning we have messages in the buffer that we should
+          # not send, they belong to the parent process, let's clear the buffer.
+          if forked?
+            @message_buffer.reset
+            update_fork_pid
+          end
+          @message_buffer.add(message)
+        }
       end
 
       def flush(*)
-        @message_buffer.flush()
+        @mx.synchronize {
+          @message_buffer.flush()
+        }
       end
 
       # Compatibility with `Sender`
@@ -26,6 +44,19 @@ def stop()
       # Compatibility with `Sender`
       def rendez_vous()
       end
+
+      private
+
+      # below are "fork management" methods to be able to clean the MessageBuffer
+      # if it detects that it is running in a unknown PID.
+
+      def forked?
+        Process.pid != @fork_pid
+      end
+
+      def update_fork_pid
+        @fork_pid = Process.pid
+      end
     end
   end
 end
diff --git a/spec/statsd/forwarder_spec.rb b/spec/statsd/forwarder_spec.rb
index 96d553af..08adba63 100644
--- a/spec/statsd/forwarder_spec.rb
+++ b/spec/statsd/forwarder_spec.rb
@@ -105,7 +105,7 @@
       it 'builds the sender' do
         expect(Datadog::Statsd::Sender)
           .to receive(:new)
-          .with(message_buffer)
+          .with(message_buffer, logger: logger)
           .exactly(1)
 
         subject
@@ -283,7 +283,7 @@
       it 'builds the sender' do
         expect(Datadog::Statsd::Sender)
           .to receive(:new)
-          .with(message_buffer)
+          .with(message_buffer, logger: logger)
           .exactly(1)
 
         subject

From 975657545a6371ead2d1bdd6c49c134df6b1ef14 Mon Sep 17 00:00:00 2001
From: Remy Mathieu <remy.mathieu@datadoghq.com>
Date: Fri, 17 Sep 2021 16:01:59 +0200
Subject: [PATCH 2/7] [statsd] address multiple feedback.

---
 lib/datadog/statsd/forwarder.rb      |  6 ++--
 lib/datadog/statsd/message_buffer.rb |  2 +-
 lib/datadog/statsd/sender.rb         | 53 +++++++++++++---------------
 3 files changed, 28 insertions(+), 33 deletions(-)

diff --git a/lib/datadog/statsd/forwarder.rb b/lib/datadog/statsd/forwarder.rb
index 07b65450..57194bd5 100644
--- a/lib/datadog/statsd/forwarder.rb
+++ b/lib/datadog/statsd/forwarder.rb
@@ -49,13 +49,12 @@ def initialize(
           raise ArgumentError, "buffer_max_payload_size is not high enough to use telemetry (tags=(#{global_tags.inspect}))"
         end
 
-        @buffer = MessageBuffer.new(@connection,
+        buffer = MessageBuffer.new(@connection,
           max_payload_size: buffer_max_payload_size,
           max_pool_size: buffer_max_pool_size || DEFAULT_BUFFER_POOL_SIZE,
           overflowing_stategy: buffer_overflowing_stategy,
         )
-
-        @sender = single_thread ? SingleThreadSender.new(buffer) : Sender.new(buffer, logger: logger)
+        @sender = (single_thread ? SingleThreadSender : Sender).new(buffer, logger: logger)
         @sender.start
       end
 
@@ -99,7 +98,6 @@ def close
       end
 
       private
-      attr_reader :buffer
       attr_reader :sender
       attr_reader :connection
 
diff --git a/lib/datadog/statsd/message_buffer.rb b/lib/datadog/statsd/message_buffer.rb
index d3d200d1..6ade90aa 100644
--- a/lib/datadog/statsd/message_buffer.rb
+++ b/lib/datadog/statsd/message_buffer.rb
@@ -19,7 +19,7 @@ def initialize(connection,
         @overflowing_stategy = overflowing_stategy
 
         @buffer = String.new
-        @message_count = 0
+        reset
       end
 
       def add(message)
diff --git a/lib/datadog/statsd/sender.rb b/lib/datadog/statsd/sender.rb
index a41c46e9..4155d0db 100644
--- a/lib/datadog/statsd/sender.rb
+++ b/lib/datadog/statsd/sender.rb
@@ -19,9 +19,12 @@ def initialize(message_buffer, logger: nil)
       end
 
       def flush(sync: false)
+        # keep a copy around in case another thread is calling #stop while this method is running
+        current_message_queue = message_queue
+
         # don't try to flush if there is no message_queue instantiated or
         # no companion thread running
-        if !message_queue
+        if !current_message_queue
           @logger.debug { "Statsd: can't flush: no message queue ready" } if @logger
           return
         end
@@ -30,24 +33,22 @@ def flush(sync: false)
           return
         end
 
-        @mx.synchronize {
-          message_queue.push(:flush)
-        }
-
+        current_message_queue.push(:flush)
         rendez_vous if sync
       end
 
       def rendez_vous
-        @mx.synchronize {
-          # Initialize and get the thread's sync queue
-          queue = (Thread.current[:statsd_sync_queue] ||= Queue.new)
-          # tell sender-thread to notify us in the current
-          # thread's queue
-          message_queue.push(queue)
-          # wait for the sender thread to send a message
-          # once the flush is done
-          queue.pop
-        }
+        # could happen if #start hasn't be called
+        return unless message_queue
+
+        # Initialize and get the thread's sync queue
+        queue = (Thread.current[:statsd_sync_queue] ||= Queue.new)
+        # tell sender-thread to notify us in the current
+        # thread's queue
+        message_queue.push(queue)
+        # wait for the sender thread to send a message
+        # once the flush is done
+        queue.pop
       end
 
       def add(message)
@@ -57,11 +58,11 @@ def add(message)
         # empty the message queue and message buffers (these messages belong to
         # the parent process) and spawn a new companion thread.
         if !sender_thread.alive?
-          @logger.debug { "Statsd: companion thread is dead, re-creating one" } if @logger
           @mx.synchronize {
             # a call from another thread has already re-created
             # the companion thread before this one acquired the lock
             break if sender_thread.alive?
+            @logger.debug { "Statsd: companion thread is dead, re-creating one" } if @logger
 
             message_queue.close if CLOSEABLE_QUEUES
             @message_queue = nil
@@ -84,23 +85,19 @@ def start
 
       if CLOSEABLE_QUEUES
         def stop(join_worker: true)
-          @mx.synchronize {
-            message_queue = @message_queue
-            message_queue.close if message_queue
+          message_queue = @message_queue
+          message_queue.close if message_queue
 
-            sender_thread = @sender_thread
-            sender_thread.join if sender_thread && join_worker
-          }
+          sender_thread = @sender_thread
+          sender_thread.join if sender_thread && join_worker
         end
       else
         def stop(join_worker: true)
-          @mx.synchronize {
-            message_queue = @message_queue
-            message_queue << :close if message_queue
+          message_queue = @message_queue
+          message_queue << :close if message_queue
 
-            sender_thread = @sender_thread
-            sender_thread.join if sender_thread && join_worker
-          }
+          sender_thread = @sender_thread
+          sender_thread.join if sender_thread && join_worker
         end
       end
 

From 38758f4ca9e6c51a0e54847ee49f5959d0d8813a Mon Sep 17 00:00:00 2001
From: Remy Mathieu <remy.mathieu@datadoghq.com>
Date: Mon, 20 Sep 2021 10:42:41 +0200
Subject: [PATCH 3/7] [telemetry] reset on message buffer re-creation.

---
 lib/datadog/statsd/connection.rb     | 7 +++++++
 lib/datadog/statsd/message_buffer.rb | 5 +++++
 lib/datadog/statsd/sender.rb         | 1 +
 3 files changed, 13 insertions(+)

diff --git a/lib/datadog/statsd/connection.rb b/lib/datadog/statsd/connection.rb
index 7b643248..40cf7a23 100644
--- a/lib/datadog/statsd/connection.rb
+++ b/lib/datadog/statsd/connection.rb
@@ -8,8 +8,15 @@ def initialize(telemetry: nil, logger: nil)
         @logger = logger
       end
 
+      def reset_telemetry
+        telemetry.reset
+      end
+
       # Close the underlying socket
       def close
+        # NOTE(remy): we do not want to automatically reset the telemetry object
+        # here because the retry mechanism may automatically re-create the connection
+        # in this case, we want to keep the data for the telemetry
         begin
           @socket && @socket.close if instance_variable_defined?(:@socket)
         rescue StandardError => boom
diff --git a/lib/datadog/statsd/message_buffer.rb b/lib/datadog/statsd/message_buffer.rb
index 6ade90aa..e8b026c8 100644
--- a/lib/datadog/statsd/message_buffer.rb
+++ b/lib/datadog/statsd/message_buffer.rb
@@ -47,6 +47,10 @@ def reset
         @message_count = 0
       end
 
+      def reset_telemetry
+        connection.reset_telemetry
+      end
+
       def flush
         return if buffer.empty?
 
@@ -55,6 +59,7 @@ def flush
       end
 
       private
+
       attr :max_payload_size
       attr :max_pool_size
 
diff --git a/lib/datadog/statsd/sender.rb b/lib/datadog/statsd/sender.rb
index 4155d0db..9e6e7bce 100644
--- a/lib/datadog/statsd/sender.rb
+++ b/lib/datadog/statsd/sender.rb
@@ -67,6 +67,7 @@ def add(message)
             message_queue.close if CLOSEABLE_QUEUES
             @message_queue = nil
             message_buffer.reset
+            message_buffer.reset_telemetry
             start
           }
         end

From a3375b21be583f059327a1726f0404c179ea81d0 Mon Sep 17 00:00:00 2001
From: Remy Mathieu <remy.mathieu@datadoghq.com>
Date: Mon, 20 Sep 2021 11:14:31 +0200
Subject: [PATCH 4/7] [telemetry] reset the telemetry on a fork in the
 single_thread_sender

---
 README.md                                  | 12 ++++++------
 lib/datadog/statsd/single_thread_sender.rb |  1 +
 2 files changed, 7 insertions(+), 6 deletions(-)

diff --git a/README.md b/README.md
index 64162533..93cc3efd 100644
--- a/README.md
+++ b/README.md
@@ -128,13 +128,13 @@ In order to use DogStatsD metrics, events, and Service Checks the Agent must be
 
 After the client is created, you can start sending custom metrics to Datadog. See the dedicated [Metric Submission: DogStatsD documentation](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby) to see how to submit all supported metric types to Datadog with working code examples:
 
-* [Submit a COUNT metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#count).
-* [Submit a GAUGE metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#gauge).
-* [Submit a SET metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#set)
-* [Submit a HISTOGRAM metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#histogram)
-* [Submit a DISTRIBUTION metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#distribution)
+* [Submit a COUNT metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#count).
+* [Submit a GAUGE metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#gauge).
+* [Submit a SET metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#set)
+* [Submit a HISTOGRAM metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#histogram)
+* [Submit a DISTRIBUTION metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#distribution)
 
-Some options are suppported when submitting metrics, like [applying a Sample Rate to your metrics](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-submission-options) or [tagging your metrics with your custom tags](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-tagging). Find all the available functions to report metrics in the [DogStatsD-ruby rubydoc](https://www.rubydoc.info/github/DataDog/dogstatsd-ruby/master/Datadog/Statsd).
+Some options are suppported when submitting metrics, like [applying a Sample Rate to your metrics](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-submission-options) or [tagging your metrics with your custom tags](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-tagging). Find all the available functions to report metrics in the [DogStatsD-ruby rubydoc](https://www.rubydoc.info/github/DataDog/dogstatsd-ruby/master/Datadog/Statsd).
 
 ### Events
 
diff --git a/lib/datadog/statsd/single_thread_sender.rb b/lib/datadog/statsd/single_thread_sender.rb
index d82dc676..0efb886b 100644
--- a/lib/datadog/statsd/single_thread_sender.rb
+++ b/lib/datadog/statsd/single_thread_sender.rb
@@ -21,6 +21,7 @@ def add(message)
           # not send, they belong to the parent process, let's clear the buffer.
           if forked?
             @message_buffer.reset
+            @message_buffer.reset_telemetry
             update_fork_pid
           end
           @message_buffer.add(message)

From dc24b67f198805ffe62642af5991549bfaa31d62 Mon Sep 17 00:00:00 2001
From: Remy Mathieu <remy.mathieu@datadoghq.com>
Date: Mon, 20 Sep 2021 14:08:23 +0200
Subject: [PATCH 5/7] [sender] cleaner interface to reset the message buffer.

---
 README.md                            | 12 ++++++------
 lib/datadog/statsd/message_buffer.rb | 15 ++++++++-------
 lib/datadog/statsd/sender.rb         |  1 -
 3 files changed, 14 insertions(+), 14 deletions(-)

diff --git a/README.md b/README.md
index 93cc3efd..64162533 100644
--- a/README.md
+++ b/README.md
@@ -128,13 +128,13 @@ In order to use DogStatsD metrics, events, and Service Checks the Agent must be
 
 After the client is created, you can start sending custom metrics to Datadog. See the dedicated [Metric Submission: DogStatsD documentation](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby) to see how to submit all supported metric types to Datadog with working code examples:
 
-* [Submit a COUNT metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#count).
-* [Submit a GAUGE metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#gauge).
-* [Submit a SET metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#set)
-* [Submit a HISTOGRAM metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#histogram)
-* [Submit a DISTRIBUTION metric](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#distribution)
+* [Submit a COUNT metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#count).
+* [Submit a GAUGE metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#gauge).
+* [Submit a SET metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#set)
+* [Submit a HISTOGRAM metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#histogram)
+* [Submit a DISTRIBUTION metric](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#distribution)
 
-Some options are suppported when submitting metrics, like [applying a Sample Rate to your metrics](https://docs.datadoghq.com/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-submission-options) or [tagging your metrics with your custom tags](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-tagging). Find all the available functions to report metrics in the [DogStatsD-ruby rubydoc](https://www.rubydoc.info/github/DataDog/dogstatsd-ruby/master/Datadog/Statsd).
+Some options are suppported when submitting metrics, like [applying a Sample Rate to your metrics](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-submission-options) or [tagging your metrics with your custom tags](https://docs.datadoghq.com/developers/metrics/dogstatsd_metrics_submission/?tab=ruby#metric-tagging). Find all the available functions to report metrics in the [DogStatsD-ruby rubydoc](https://www.rubydoc.info/github/DataDog/dogstatsd-ruby/master/Datadog/Statsd).
 
 ### Events
 
diff --git a/lib/datadog/statsd/message_buffer.rb b/lib/datadog/statsd/message_buffer.rb
index e8b026c8..0be43d66 100644
--- a/lib/datadog/statsd/message_buffer.rb
+++ b/lib/datadog/statsd/message_buffer.rb
@@ -19,7 +19,7 @@ def initialize(connection,
         @overflowing_stategy = overflowing_stategy
 
         @buffer = String.new
-        reset
+        clear_buffer
       end
 
       def add(message)
@@ -43,11 +43,7 @@ def add(message)
       end
 
       def reset
-        buffer.clear
-        @message_count = 0
-      end
-
-      def reset_telemetry
+        clear_buffer
         connection.reset_telemetry
       end
 
@@ -55,7 +51,7 @@ def flush
         return if buffer.empty?
 
         connection.write(buffer)
-        reset
+        clear_buffer
       end
 
       private
@@ -74,6 +70,11 @@ def should_flush?(message_size)
         false
       end
 
+      def clear_buffer
+        buffer.clear
+        @message_count = 0
+      end
+
       def preemptive_flush?
         @message_count == max_pool_size || buffer.bytesize > bytesize_threshold
       end
diff --git a/lib/datadog/statsd/sender.rb b/lib/datadog/statsd/sender.rb
index 9e6e7bce..4155d0db 100644
--- a/lib/datadog/statsd/sender.rb
+++ b/lib/datadog/statsd/sender.rb
@@ -67,7 +67,6 @@ def add(message)
             message_queue.close if CLOSEABLE_QUEUES
             @message_queue = nil
             message_buffer.reset
-            message_buffer.reset_telemetry
             start
           }
         end

From 073f6858216094aa8548a00650683cedc640aac7 Mon Sep 17 00:00:00 2001
From: Remy Mathieu <remy.mathieu@datadoghq.com>
Date: Mon, 27 Sep 2021 14:02:44 +0200
Subject: [PATCH 6/7] sender: mention that multithread calls to close/add can
 lead to undefined behavior

---
 lib/datadog/statsd/sender.rb | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/lib/datadog/statsd/sender.rb b/lib/datadog/statsd/sender.rb
index 4155d0db..a99b6957 100644
--- a/lib/datadog/statsd/sender.rb
+++ b/lib/datadog/statsd/sender.rb
@@ -84,6 +84,9 @@ def start
       end
 
       if CLOSEABLE_QUEUES
+        # when calling stop, make sure that no other threads is trying
+        # to close the sender nor trying to continue to `#add` more message
+        # into the sender.
         def stop(join_worker: true)
           message_queue = @message_queue
           message_queue.close if message_queue
@@ -92,6 +95,9 @@ def stop(join_worker: true)
           sender_thread.join if sender_thread && join_worker
         end
       else
+        # when calling stop, make sure that no other threads is trying
+        # to close the sender nor trying to continue to `#add` more message
+        # into the sender.
         def stop(join_worker: true)
           message_queue = @message_queue
           message_queue << :close if message_queue

From 362acc7818d3af2b6a16883c96ca110c633bdf2f Mon Sep 17 00:00:00 2001
From: Remy Mathieu <remy.mathieu@datadoghq.com>
Date: Tue, 28 Sep 2021 16:32:30 +0200
Subject: [PATCH 7/7] [sender] remove a redundant call in the
 single_thread_sender

---
 lib/datadog/statsd/single_thread_sender.rb | 1 -
 1 file changed, 1 deletion(-)

diff --git a/lib/datadog/statsd/single_thread_sender.rb b/lib/datadog/statsd/single_thread_sender.rb
index 0efb886b..d82dc676 100644
--- a/lib/datadog/statsd/single_thread_sender.rb
+++ b/lib/datadog/statsd/single_thread_sender.rb
@@ -21,7 +21,6 @@ def add(message)
           # not send, they belong to the parent process, let's clear the buffer.
           if forked?
             @message_buffer.reset
-            @message_buffer.reset_telemetry
             update_fork_pid
           end
           @message_buffer.add(message)