AMLII-2207 - alter dogstatsd packet buffers to automatically flush on…

… close

comp/dogstatsd/packets/buffer.go

@@ -89,6 +89,9 @@ func (pb *Buffer) flush() {
 
 // Close closes the packet buffer
 func (pb *Buffer) Close() {
+	pb.m.Lock()
+	pb.flush()
+	pb.m.Unlock()
 	close(pb.closeChannel)
 	if pb.listenerID != "" {
 		pb.telemetryStore.tlmBufferSize.Delete(pb.listenerID)

comp/dogstatsd/packets/buffer_test.go

@@ -24,7 +24,7 @@ func TestBufferTelemetry(t *testing.T) {
 	// We need a high enough duration to avoid the buffer to flush
 	// And cause the program to deadlock on the packetChannel
 	duration := 10 * time.Second
-	packetChannel := make(chan Packets)
+	packetChannel := make(chan Packets, 1)
 	buffer := NewBuffer(3, duration, packetChannel, "test_buffer", telemetryStore)
 	defer buffer.Close()
 
@@ -127,3 +127,22 @@ func TestBufferTelemetryFull(t *testing.T) {
 
 	assert.Equal(t, float64(1), channelSizeMetrics[0].Value())
 }
+
+func TestBufferFlushOnClose(t *testing.T) {
+	telemetryComponent := fxutil.Test[telemetry.Component](t, telemetryimpl.MockModule())
+	telemetryStore := NewTelemetryStore(nil, telemetryComponent)
+	duration := 10 * time.Hour
+	packetChannel := make(chan Packets, 1)
+	buffer := NewBuffer(0, duration, packetChannel, "test_buffer", telemetryStore)
+	packet := &Packet{
+		Contents:   []byte("test"),
+		Buffer:     []byte("test read"),
+		Origin:     "test origin",
+		ListenerID: "1",
+		Source:     0,
+	}
+
+	buffer.Append(packet)
+	buffer.Close()
+	assert.Equal(t, 1, len(packetChannel))
+}

releasenotes/notes/stream-socket-race-fix-255b6e7a10f7fde0.yaml

@@ -0,0 +1,13 @@
+# Each section from every release note are combined when the
+# CHANGELOG.rst is rendered. So the text needs to be worded so that
+# it does not depend on any information only available in another
+# section. This may mean repeating some details, but each section
+# must be readable independently of the other.
+#
+# Each section note must be formatted as reStructuredText.
+---
+fixes:
+  - |
+    Fixed race condition in stream UDS clients of Dogstatsd that
+    allowed for the loss of recieved data.
+