@@ -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)

@@ -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)

new file mode 100644

@@ -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

+}

new file mode 100644

@@ -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()

+}