Repository: moby/moby (language: go)

Issue:
docker stats hangs
**Description**

Docker is working for a few weeks, then it is somehow not responding anymore. Some commands work, others not.
e.g. "docker stats" is hanging. "docker pa -a" works.

I searched through the various issues here that describe different hangs, but found nothing that solves my issue.

Restarting the docker daemon fixes the issue for a few days...

**Steps to reproduce the issue:**
1. We just use docker and are starting / stopping containers.

**Describe the results you received:**
Docker hangs for whatever reason.

**Describe the results you expected:**
Docker should work as expected.

**Additional information you deem important (e.g. issue happens only occasionally):**
In other threads there was the hint of doing a SIGUSR1-Signal to dockerd and posting the result. Also, I have done an strace on the stuck "docker stats" command.
It can be found here: https://git.jar.media/snippets/8

**Output of `docker version`:**

```
Docker version 18.09.7, build 2d0083d
```

**Output of `docker info`:**

```
Containers: 58
 Running: 58
 Paused: 0
 Stopped: 0
Images: 776
Server Version: 18.09.7
Storage Driver: overlay2
 Backing Filesystem: extfs
 Supports d_type: true
 Native Overlay Diff: true
Logging Driver: json-file
Cgroup Driver: cgroupfs
Plugins:
 Volume: local
 Network: bridge host macvlan null overlay
 Log: awslogs fluentd gcplogs gelf journald json-file local logentries splunk syslog
Swarm: inactive
Runtimes: runc
Default Runtime: runc
Init Binary: docker-init
containerd version: 894b81a4b802e4eb2a91d1ce216b8817763c29fb
runc version: 425e105d5a03fabd737a126ad93d62a9eeede87f
init version: fec3683
Security Options:
 seccomp
  Profile: default
Kernel Version: 4.9.0-9-amd64
Operating System: Debian GNU/Linux 9 (stretch)
OSType: linux
Architecture: x86_64
CPUs: 48
Total Memory: 125.8GiB
Name: h19
ID: OJOK:YAOX:OTVB:ALUM:SZUE:ZKJD:T26K:OPEF:XP6O:33GQ:2GRT:KC36
Docker Root Dir: /var/lib/docker
Debug Mode (client): false
Debug Mode (server): false
Registry: https://index.docker.io/v1/
Labels:
Experimental: false
Insecure Registries:
 127.0.0.0/8
Live Restore Enabled: false
Product License: Community Engine

WARNING: No swap limit support
```

**Additional environment details (AWS, VirtualBox, physical, etc.):**
No additional info.


Discussion / hints:
The strace likely won't have much details (as it would only strike the CLI, which is likely not where the issue is)

I attached the files here for easier finding;
[goroutine-stacks-2019-07-15T073325_0200.log.zip](https://github.com/moby/moby/files/3391815/goroutine-stacks-2019-07-15T073325_0200.log.zip)
[strace_of_docker_stats.txt](https://github.com/moby/moby/files/3391816/strace_of_docker_stats.txt)
@cpuguy83 @kolyshkin ptal
@jarmediagmbh is there anything useful in the daemon logs? Could you describe a bit what kind of load / containers you're running? (long-lived / short-lived containers? do you use health-checks? do container produce a lot of logs? do you use bind-mounts?)
**is there anything useful in the daemon logs?** 
- No. Before opening the issue here we tried so many different things. Of course we also analyzed several log files (syslog, daemonlog, dmesg etc.). We did not find suspicious things, just the normal "a new network interface appeared/disappeared etc.".

**Could you describe a bit what kind of load / containers you're running? (long-lived / short-lived containers?** 
We have about 60 Containers running. Each container will be deleted and recreated each morning (we have a very special usecase.. ;) ). The containers offer development environments for web development, the load should not be sooo huge.

**do you use health-checks?**
no.

**do container produce a lot of logs?** 
no.

**do you use bind-mounts?**
Yes. There are several mounts between a container and the host system.

> The containers offer development environments for web development, the load should not be sooo huge.

All containers are running "interactively" (so a shell session attached to it?)
Wow. I have found a suspicous thing: Besides our dev-containers we are using zalenium (https://github.com/zalando/zalenium) on the same docker host. It automatically generated containers with google chrome running in it, so you can remotely and programmatically control the chrome-instances e.g. for test automation for websites.

After I did a "docker-compose down" on the zalenium directories (zalenium consists of a control-container and several chrome-instance-containers), docker stats did work again. However, I am sure, that the last time we had this issue, we removed all containers from the host and docker stats did not work aswell.
> > The containers offer development environments for web development, the load should not be sooo huge.
> 
> All containers are running "interactively" (so a shell session attached to it?)

No, we ssh into the conatiners to work on them. There is no "docker exec -it [CONTAINER] /bin/bash", if you ment that.
> No, we ssh into the conatiners to work on them. There is no "docker exec -it [CONTAINER] /bin/bash", if you ment that.

Thanks; yes that's what I meant (`docker exec` or `docker run -it`) 👍 
In the past we had a lot of problems using docker exec / docker run, because the commands sometimes did not terminate. Thus, even our control-scripts use ssh into the containers and not "docker exec".
Starting and stopping a few tens (~30) containers simultaneously and continuously with periodic docker stats API invocation, the stats API gets hang at a random time and often makes deadlocks for my container termination routines which calls the stats API to sync the final stats to an internal database.
For testing, each container run just a simple "sleep" command for a few seconds, so there is no computational overheads of containers, though we are controlling the containers via our own "job scheduler" implementation.
Maybe there is a race condition issue inside the stats API implementation when the container terminates.

FYI: I have an alternative stats implementation which reads and parses the sysfs directly, and using it instead of the docker stats API resolves the deadlock issue completely, under heavy-load tests. So I'm pretty sure that the problem resides inside the docker daemon somewhere....
We have the same issue, where docker stats suddenly stops responding.

## Issue description
- 3 virtual instances all have the same issue (occurred at different times).
- Running ~30 containers.
- Starting and stopping a few times a day.
- The containers might start on different instance each time.
- Running without issues for at least a week.

Docker stats suddenly stops responding on all containers without an apparent cause.

This never resolves (has to be terminated):
```bash
docker stats --no-stream dd7e99646619
```

This will error with "timeout waiting for stats":
```bash
docker stats dd7e99646619
```

## Docker info & version

All instances are the same:

[docker-info.txt](https://github.com/moby/moby/files/15074922/docker-info.txt)
[docker-version.txt](https://github.com/moby/moby/files/15074930/docker-version.txt)

## Logs

These are the docker daemon logs from around the time the stats endpoint stopped responding.
**These are actually the only logs that are there +- a few hours**

### Instance A

Issue occurred around 10:00 and 13:00.

[journalctl-logs.txt](https://github.com/moby/moby/files/15075226/journalctl-logs.txt)

### Instance B

Issue occurred around 22:00.

[journalctl-logs.txt](https://github.com/moby/moby/files/15077441/journalctl-logs.txt)

### Instance C

Issue occurred around 10:00.

[journalctl-logs.txt](https://github.com/moby/moby/files/15077444/journalctl-logs.txt)

## Metrics

I have all the Prometheus "node" metrics from the instances.
There is no blatantly apparent cause or correlation with the metric data.

I have observed:
- Drop in Disk IOps when the issue arises on two of the 3 instances.
- Some network interfaces start receiving traffic and some stop receiving traffic.
- Drop in memory usage.

## What have you tried

I have tried sending the stats request directly to the docker socket in various combinations like so:
```
curl --unix-socket /var/run/docker.sock http:/v1.24/containers/452bb2e0564d/stats?stream=0
curl --unix-socket /var/run/docker.sock http:/v1.44/containers/452bb2e0564d/stats?stream=0
curl --unix-socket /var/run/docker.sock http:/v1.44/containers/452bb2e0564d/stats?stream=1
```
This request never resolves.

## What fixes the problem

Restarting the docker daemon fixes the issue.

## Important

**I currently have three distinct instances with the problematic docker at hand, have not yet restarted the daemon.
And I can keep it that way for a few days.
So if we have a quick response, I can debug this further directly on the problematic instances**.
On the stuck instances, please grab a goroutine stack dump.

You can get this from the endpoint t "/debug/pprof/goroutine?debug=2"

Also please provide the output of "docker info" and "docker version", assuming those work (they shouldn't generally get wedged).
Thanks for the quick response, here it is.

Instance A:
[goroutinedump.txt](https://github.com/moby/moby/files/15079539/goroutinedump.txt)
[info.txt](https://github.com/moby/moby/files/15079549/info.txt)
[version.txt](https://github.com/moby/moby/files/15079559/version.txt)
[journalctl-logs.txt](https://github.com/moby/moby/files/15079568/journalctl-logs.txt)

Instance B:
[goroutinedump.txt](https://github.com/moby/moby/files/15079570/goroutinedump.txt)
[info.txt](https://github.com/moby/moby/files/15079573/info.txt)
[version.txt](https://github.com/moby/moby/files/15079577/version.txt)
[journalctl-logs.txt](https://github.com/moby/moby/files/15079575/journalctl-logs.txt)

Instance C:
[goroutinedump.txt](https://github.com/moby/moby/files/15079581/goroutinedump.txt)
[info.txt](https://github.com/moby/moby/files/15079584/info.txt)
[version.txt](https://github.com/moby/moby/files/15079589/version.txt)
[journalctl-logs.txt](https://github.com/moby/moby/files/15079586/journalctl-logs.txt)
@zatlodan Thanks, I can see there are a couple of places this is getting wedged trying to close the (cloudwatch) log driver because the log buffer is full. Additionally the same thing is happening in the local log cache (used to allow `docker logs` to work with cloudwatch). Its trying to send the log to cloudwatch but is blocked (because of the above problem) and it is preventing the log driver from being closed.

This is causing the container state to remain locked indefinitely which is causing other issues, namely preventing an image removal from completing because some container is using the image and its trying to take the container lock to see if the container is running... I think this is also another bug, we shouldn't care if the container is running or not here... but there are likely historical reasons for why this is the way it is.
I'll try and take a look at this as well.



Fix patch:
diff --git a/daemon/logger/awslogs/cloudwatchlogs.go b/daemon/logger/awslogs/cloudwatchlogs.go
index 3d90cb4504df8..b6b96d33d5bf8 100644
--- a/daemon/logger/awslogs/cloudwatchlogs.go
+++ b/daemon/logger/awslogs/cloudwatchlogs.go
@@ -8,7 +8,7 @@ import (
 	"regexp"
 	"sort"
 	"strconv"
-	"sync"
+	"sync/atomic"
 	"time"
 	"unicode/utf8"
 
@@ -76,10 +76,11 @@ type logStream struct {
 	forceFlushInterval time.Duration
 	multilinePattern   *regexp.Regexp
 	client             api
-	messages           chan *logger.Message
-	lock               sync.RWMutex
-	closed             bool
-	sequenceToken      *string
+
+	messages *loggerutils.MessageQueue
+	closed   atomic.Bool
+
+	sequenceToken *string
 }
 
 type logStreamConfig struct {
@@ -158,7 +159,7 @@ func New(info logger.Info) (logger.Logger, error) {
 		forceFlushInterval: containerStreamConfig.forceFlushInterval,
 		multilinePattern:   containerStreamConfig.multilinePattern,
 		client:             client,
-		messages:           make(chan *logger.Message, containerStreamConfig.maxBufferedEvents),
+		messages:           loggerutils.NewMessageQueue(containerStreamConfig.maxBufferedEvents),
 	}
 
 	creationDone := make(chan bool)
@@ -168,12 +169,10 @@ func New(info logger.Info) (logger.Logger, error) {
 			maxBackoff := 32
 			for {
 				// If logger is closed we are done
-				containerStream.lock.RLock()
-				if containerStream.closed {
-					containerStream.lock.RUnlock()
+				if containerStream.closed.Load() {
 					break
 				}
-				containerStream.lock.RUnlock()
+
 				err := containerStream.create()
 				if err == nil {
 					break
@@ -426,25 +425,26 @@ func (l *logStream) BufSize() int {
 	return maximumBytesPerEvent
 }
 
+var errClosed = errors.New("awslogs is closed")
+
 // Log submits messages for logging by an instance of the awslogs logging driver
 func (l *logStream) Log(msg *logger.Message) error {
-	l.lock.RLock()
-	defer l.lock.RUnlock()
-	if l.closed {
-		return errors.New("awslogs is closed")
+	// No need to check if we are closed here since the queue will be closed
+	// (i.e. returns false) in this case.
+	ctx := context.TODO()
+	if err := l.messages.Enqueue(ctx, msg); err != nil {
+		if err == loggerutils.ErrQueueClosed {
+			return errClosed
+		}
+		return err
 	}
-	l.messages <- msg
 	return nil
 }
 
 // Close closes the instance of the awslogs logging driver
 func (l *logStream) Close() error {
-	l.lock.Lock()
-	defer l.lock.Unlock()
-	if !l.closed {
-		close(l.messages)
-	}
-	l.closed = true
+	l.closed.Store(true)
+	l.messages.Close()
 	return nil
 }
 
@@ -561,6 +561,8 @@ func (l *logStream) collectBatch(created chan bool) {
 	var eventBuffer []byte
 	var eventBufferTimestamp int64
 	batch := newEventBatch()
+
+	chLogs := l.messages.Receiver()
 	for {
 		select {
 		case t := <-ticker.C:
@@ -576,7 +578,7 @@ func (l *logStream) collectBatch(created chan bool) {
 			}
 			l.publishBatch(batch)
 			batch.reset()
-		case msg, more := <-l.messages:
+		case msg, more := <-chLogs:
 			if !more {
 				// Flush event buffer and release resources
 				l.processEvent(batch, eventBuffer, eventBufferTimestamp)
diff --git a/daemon/logger/loggerutils/queue.go b/daemon/logger/loggerutils/queue.go
new file mode 100644
index 0000000000000..2cd62a6d66186
--- /dev/null
+++ b/daemon/logger/loggerutils/queue.go
@@ -0,0 +1,156 @@
+package loggerutils
+
+import (
+	"context"
+	"sync"
+
+	"github.com/docker/docker/daemon/logger"
+	"github.com/pkg/errors"
+)
+
+// MessageQueue is a queue for log messages.
+//
+// [MessageQueue.Enqueue] will block when the queue is full.
+// To dequeue messages call [MessageQueue.Reciever] and pull messsages off the
+// returned channel.
+//
+// Closing only prevents new messages from being added to the queue.
+// The queue can still be drained after close.
+//
+// The zero value of MessageQueue is safe to use, but does not do any internal
+// buffering (queue size is 0).
+type MessageQueue struct {
+	maxSize int
+
+	mu      sync.Mutex
+	closing bool
+	closed  chan struct{}
+
+	// Blocks multiple calls to [MessageQueue.Close] until the queue is actually closed
+	closeWait chan struct{}
+
+	// We need to be able to safely close the send channel so that [MessageQueue.Dequeue]
+	// can drain the queue without blocking.
+	// This cond var helps deal with that.
+	cond        *sync.Cond
+	sendWaiters int
+
+	ch chan *logger.Message
+}
+
+// NewMessageQueue creates a new queue with the specified size.
+func NewMessageQueue(maxSize int) *MessageQueue {
+	var q MessageQueue
+	q.maxSize = maxSize
+	q.init()
+	return &q
+}
+
+func (q *MessageQueue) init() {
+	if q.cond == nil {
+		q.cond = sync.NewCond(&q.mu)
+	}
+
+	if q.ch == nil {
+		q.ch = make(chan *logger.Message, q.maxSize)
+	}
+
+	if q.closed == nil {
+		q.closed = make(chan struct{})
+	}
+
+	if q.closeWait == nil {
+		q.closeWait = make(chan struct{})
+	}
+}
+
+var ErrQueueClosed = errors.New("queue is closed")
+
+// Enqueue adds the provided message to the queue.
+// Enqueue blocks if the queue is full.
+//
+// The two possible error cases are:
+// 1. The provided context is cancelled
+// 2. [ErrQueueClosed] when the queue has been closed.
+func (q *MessageQueue) Enqueue(ctx context.Context, m *logger.Message) error {
+	q.mu.Lock()
+	q.init()
+
+	// Increment the waiter count
+	// This prevents the send channel from being closed while we are trying to send.
+	q.sendWaiters++
+	q.mu.Unlock()
+
+	defer func() {
+		q.mu.Lock()
+		// Decrement the waiter count and signal to any potential closer to check
+		// the wait count again.
+		// Only bother signaling if this is the last waiter.
+		q.sendWaiters--
+		if q.sendWaiters == 0 {
+			q.cond.Signal()
+		}
+		q.mu.Unlock()
+	}()
+
+	// Before trying to send on the channel, check if we care closed.
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-q.closed:
+		return ErrQueueClosed
+	default:
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-q.closed:
+		return ErrQueueClosed
+	case q.ch <- m:
+		return nil
+	}
+}
+
+// Close prevents any new messages from being added to the queue.
+func (q *MessageQueue) Close() {
+	q.mu.Lock()
+
+	q.init()
+
+	if q.closing {
+		// unlock the mutex here so that the goroutine waiting on the cond var can
+		// take the lock when signaled.
+		q.mu.Unlock()
+		<-q.closeWait
+		return
+	}
+
+	defer q.mu.Unlock()
+
+	// Prevent multiple Close calls from trying to close things.
+	q.closing = true
+
+	close(q.closed)
+
+	// Wait for any senders to finish
+	// Because we closed the channel above, this shouldn't block for a long period.
+	for q.sendWaiters > 0 {
+		q.cond.Wait()
+	}
+
+	close(q.ch)
+	close(q.closeWait)
+}
+
+// Receiver returns a channel that can be used to dequeue messages
+// The channel will be closed when the message queue is closed but may have
+// messages buffered.
+func (q *MessageQueue) Receiver() <-chan *logger.Message {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	q.init()
+
+	return q.ch
+}


Test patch:
diff --git a/daemon/logger/awslogs/cloudwatchlogs_test.go b/daemon/logger/awslogs/cloudwatchlogs_test.go
index 90bfadbb28d97..1ea4befcdfa68 100644
--- a/daemon/logger/awslogs/cloudwatchlogs_test.go
+++ b/daemon/logger/awslogs/cloudwatchlogs_test.go
@@ -356,9 +356,10 @@ func TestCreateAlreadyExists(t *testing.T) {
 func TestLogClosed(t *testing.T) {
 	mockClient := &mockClient{}
 	stream := &logStream{
-		client: mockClient,
-		closed: true,
+		client:   mockClient,
+		messages: loggerutils.NewMessageQueue(0),
 	}
+	stream.Close()
 	err := stream.Log(&logger.Message{})
 	assert.Check(t, err != nil)
 }
@@ -370,7 +371,7 @@ func TestLogBlocking(t *testing.T) {
 	mockClient := &mockClient{}
 	stream := &logStream{
 		client:   mockClient,
-		messages: make(chan *logger.Message),
+		messages: loggerutils.NewMessageQueue(0),
 	}
 
 	errorCh := make(chan error, 1)
@@ -387,14 +388,11 @@ func TestLogBlocking(t *testing.T) {
 		t.Fatal("Expected stream.Log to block: ", err)
 	default:
 	}
+
 	// assuming it is blocked, we can now try to drain the internal channel and
 	// unblock it
-	select {
-	case <-time.After(10 * time.Millisecond):
-		// if we're unable to drain the channel within 10ms, something seems broken
-		t.Fatal("Expected to be able to read from stream.messages but was unable to")
-	case <-stream.messages:
-	}
+	<-stream.messages.Receiver()
+
 	select {
 	case err := <-errorCh:
 		assert.NilError(t, err)
@@ -408,7 +406,7 @@ func TestLogBufferEmpty(t *testing.T) {
 	mockClient := &mockClient{}
 	stream := &logStream{
 		client:   mockClient,
-		messages: make(chan *logger.Message, 1),
+		messages: loggerutils.NewMessageQueue(1),
 	}
 	err := stream.Log(&logger.Message{})
 	assert.NilError(t, err)
@@ -556,7 +554,7 @@ func TestCollectBatchSimple(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	mockClient.putLogEventsFunc = func(ctx context.Context, input *cloudwatchlogs.PutLogEventsInput, opts ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.PutLogEventsOutput, error) {
@@ -575,15 +573,20 @@ func TestCollectBatchSimple(t *testing.T) {
 	close(d)
 	go stream.collectBatch(d)
 
-	stream.Log(&logger.Message{
+	err := stream.Log(&logger.Message{
 		Line:      []byte(logline),
 		Timestamp: time.Time{},
 	})
+	assert.NilError(t, err)
 
 	ticks <- time.Time{}
 	ticks <- time.Time{}
 	stream.Close()
 
+	for len(calls) != 1 {
+		time.Sleep(10 * time.Millisecond)
+	}
+
 	assert.Assert(t, len(calls) == 1)
 	argument := calls[0]
 	assert.Assert(t, argument != nil)
@@ -598,7 +601,7 @@ func TestCollectBatchTicker(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -666,7 +669,7 @@ func TestCollectBatchMultilinePattern(t *testing.T) {
 		logStreamName:    streamName,
 		multilinePattern: multilinePattern,
 		sequenceToken:    aws.String(sequenceToken),
-		messages:         make(chan *logger.Message),
+		messages:         loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -732,7 +735,7 @@ func BenchmarkCollectBatch(b *testing.B) {
 			logGroupName:  groupName,
 			logStreamName: streamName,
 			sequenceToken: aws.String(sequenceToken),
-			messages:      make(chan *logger.Message),
+			messages:      loggerutils.NewMessageQueue(0),
 		}
 		mockClient.putLogEventsFunc = func(ctx context.Context, input *cloudwatchlogs.PutLogEventsInput, opts ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.PutLogEventsOutput, error) {
 			return &cloudwatchlogs.PutLogEventsOutput{
@@ -765,7 +768,7 @@ func BenchmarkCollectBatchMultilinePattern(b *testing.B) {
 			logStreamName:    streamName,
 			multilinePattern: multilinePattern,
 			sequenceToken:    aws.String(sequenceToken),
-			messages:         make(chan *logger.Message),
+			messages:         loggerutils.NewMessageQueue(0),
 		}
 		mockClient.putLogEventsFunc = func(ctx context.Context, input *cloudwatchlogs.PutLogEventsInput, opts ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.PutLogEventsOutput, error) {
 			return &cloudwatchlogs.PutLogEventsOutput{
@@ -796,7 +799,7 @@ func TestCollectBatchMultilinePatternMaxEventAge(t *testing.T) {
 		logStreamName:    streamName,
 		multilinePattern: multilinePattern,
 		sequenceToken:    aws.String(sequenceToken),
-		messages:         make(chan *logger.Message),
+		messages:         loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -870,7 +873,7 @@ func TestCollectBatchMultilinePatternNegativeEventAge(t *testing.T) {
 		logStreamName:    streamName,
 		multilinePattern: multilinePattern,
 		sequenceToken:    aws.String(sequenceToken),
-		messages:         make(chan *logger.Message),
+		messages:         loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -927,7 +930,7 @@ func TestCollectBatchMultilinePatternMaxEventSize(t *testing.T) {
 		logStreamName:    streamName,
 		multilinePattern: multilinePattern,
 		sequenceToken:    aws.String(sequenceToken),
-		messages:         make(chan *logger.Message),
+		messages:         loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -987,7 +990,7 @@ func TestCollectBatchClose(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -1089,7 +1092,7 @@ func TestCollectBatchLineSplit(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -1137,7 +1140,7 @@ func TestCollectBatchLineSplitWithBinary(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -1185,7 +1188,7 @@ func TestCollectBatchMaxEvents(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -1239,7 +1242,7 @@ func TestCollectBatchMaxTotalBytes(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -1320,7 +1323,7 @@ func TestCollectBatchMaxTotalBytesWithBinary(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
@@ -1394,7 +1397,7 @@ func TestCollectBatchWithDuplicateTimestamps(t *testing.T) {
 		logGroupName:  groupName,
 		logStreamName: streamName,
 		sequenceToken: aws.String(sequenceToken),
-		messages:      make(chan *logger.Message),
+		messages:      loggerutils.NewMessageQueue(0),
 	}
 	calls := make([]*cloudwatchlogs.PutLogEventsInput, 0)
 	called := make(chan struct{}, 50)
diff --git a/daemon/logger/loggerutils/queue_test.go b/daemon/logger/loggerutils/queue_test.go
new file mode 100644
index 0000000000000..b40be89963c93
--- /dev/null
+++ b/daemon/logger/loggerutils/queue_test.go
@@ -0,0 +1,87 @@
+package loggerutils
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/docker/docker/daemon/logger"
+	"gotest.tools/v3/assert"
+)
+
+func TestQueue(t *testing.T) {
+	q := NewMessageQueue(2)
+	msg := &logger.Message{Line: []byte("hello")}
+
+	ctx := context.Background()
+	err := q.Enqueue(ctx, msg)
+	assert.Check(t, err)
+
+	recv := q.Receiver()
+	// These pointer values should be the same
+	assert.Equal(t, msg, <-recv)
+
+	err = q.Enqueue(ctx, msg)
+	assert.Check(t, err)
+
+	err = q.Enqueue(ctx, msg)
+	assert.Check(t, err)
+
+	q.Close()
+
+	// We have 2 messages in the queue
+	// Even though this is closed, we should get a true value from dequeue twice.
+	assert.Equal(t, msg, <-recv)
+	assert.Equal(t, msg, <-recv)
+
+	// This should not block and should return false
+	_, more := <-recv
+	assert.Check(t, !more, "expected no more messages in the queue")
+
+	// Test with unbuffered
+	q = &MessageQueue{}
+	recv = q.Receiver()
+
+	chAdd := make(chan error, 1)
+	go func() {
+		chAdd <- q.Enqueue(ctx, msg)
+	}()
+
+	assert.Equal(t, msg, <-recv)
+	assert.Assert(t, <-chAdd)
+
+	ctxC, cancel := context.WithCancel(ctx)
+	cancel()
+
+	err = q.Enqueue(ctxC, msg)
+	assert.ErrorIs(t, err, context.Canceled)
+
+	// Test that blocked senders do not cause a panic on close.
+	// This check is useful because the underlying implementation uses channels
+	// with the send channel eventually getting closed when q.Close is called.
+	go func() {
+		chAdd <- q.Enqueue(ctx, msg)
+	}()
+
+	// Wait for enqueue to be ready (or as close to ready as it can be)
+	for {
+		q.mu.Lock()
+		if q.sendWaiters == 1 {
+			q.mu.Unlock()
+			break
+		}
+		q.mu.Unlock()
+		time.Sleep(time.Millisecond)
+	}
+
+	q.Close()
+
+	select {
+	case <-time.After(5 * time.Second):
+	case err := <-chAdd:
+		assert.ErrorIs(t, err, ErrQueueClosed)
+	}
+
+	// Double-close should not cause any issues
+	q.Close()
+}