[*] release logrus.Entry immediately in BrokerHook.Fire() (#1303)

* [*] release `logrus.Entry` immediately in `BrokerHook.Fire()`

Format synchronously in `Fire()` while the entry is owned by the calling
goroutine, send `MessageType` through the channel, and `send()` delivers
out without `logrus.Entry` copy and without `lastError` channel

* add tests

Author: pashagolub

internal/log/log_broker_hook.go

@@ -19,9 +19,8 @@ type MessageChanType chan MessageType
 type BrokerHook struct {
 	highLoadTimeout time.Duration     // wait this amount of time before skip log entry
 	subscribers     []MessageChanType //
-	input           chan *logrus.Entry
+	input           chan MessageType
 	ctx             context.Context
-	lastError       chan error
 	level           string
 	mu              *sync.Mutex
 	formatter       logrus.Formatter
@@ -34,8 +33,7 @@ const highLoadLimit = 200 * time.Millisecond
 func NewBrokerHook(ctx context.Context, level string) *BrokerHook {
 	l := &BrokerHook{
 		highLoadTimeout: highLoadLimit,
-		input:           make(chan *logrus.Entry, cacheLimit),
-		lastError:       make(chan error),
+		input:           make(chan MessageType, cacheLimit),
 		ctx:             ctx,
 		level:           level,
 		mu:              new(sync.Mutex),
@@ -79,18 +77,20 @@ func (hook *BrokerHook) Fire(entry *logrus.Entry) error {
 	if hook.ctx.Err() != nil {
 		return nil
 	}
+	hook.mu.Lock()
+	f := hook.formatter
+	hook.mu.Unlock()
+	raw, err := f.Format(entry)
+	if err != nil {
+		return err
+	}
 	select {
-	case hook.input <- entry:
+	case hook.input <- MessageType(raw):
 		// entry sent
 	case <-time.After(hook.highLoadTimeout):
 		// entry dropped due to a huge load, check stdout or file for detailed log
 	}
-	select {
-	case err := <-hook.lastError:
-		return err
-	default:
-		return nil
-	}
+	return nil
 }
 
 // Levels returns the available logging levels
@@ -119,38 +119,29 @@ func (hook *BrokerHook) Levels() []logrus.Level {
 
 // poll checks for incoming messages and caches them internally
 // until either a maximum amount is reached, or a timeout occurs.
-func (hook *BrokerHook) poll(input <-chan *logrus.Entry) {
+func (hook *BrokerHook) poll(input <-chan MessageType) {
 	for {
 		select {
 		case <-hook.ctx.Done(): //check context with high priority
 			return
-		case entry := <-input:
-			hook.send(entry)
+		case msg := <-input:
+			hook.send(msg)
 		}
 	}
 }
 
-// send sends cached messages to the postgres server
-func (hook *BrokerHook) send(entry *logrus.Entry) {
+// send sends out a pre-formatted message to all subscribers
+func (hook *BrokerHook) send(msg MessageType) {
 	hook.mu.Lock()
 	defer hook.mu.Unlock()
 	if len(hook.subscribers) == 0 {
-		return // Nothing to do here.
+		return
 	}
-	msg, err := hook.formatter.Format(entry)
 	for _, subscriber := range hook.subscribers {
 		select {
-		case subscriber <- MessageType(msg):
+		case subscriber <- msg:
 		default:
 			//no time to wait
 		}
 	}
-	if err != nil {
-		select {
-		case hook.lastError <- err:
-			//error sent to the logger
-		default:
-			//there is unprocessed error already
-		}
-	}
 }

internal/log/log_broker_hook_test.go

@@ -2,13 +2,32 @@ package log
 
 import (
 	"context"
+	"sync"
 	"testing"
+	"testing/synctest"
+	"time"
 
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 )
 
+type errorFormatter struct{}
+
+func (f *errorFormatter) Format(*logrus.Entry) ([]byte, error) {
+	return nil, assert.AnError
+}
+
+func TestFormatterError(t *testing.T) {
+	hook := NewBrokerHook(t.Context(), "info")
+	hook.SetBrokerFormatter(&errorFormatter{})
+	entry := logrus.NewEntry(logrus.New())
+	entry.Message = "hello broker"
+	entry.Level = logrus.InfoLevel
+	assert.Error(t, hook.Fire(entry))
+}
+
 func TestRemoveSubscriber(t *testing.T) {
-	hook := NewBrokerHook(context.Background(), "info")
+	hook := NewBrokerHook(t.Context(), "info")
 	msgChan1 := make(MessageChanType)
 	msgChan2 := make(MessageChanType)
 	hook.AddSubscriber(msgChan1)
@@ -23,3 +42,96 @@ func TestRemoveSubscriber(t *testing.T) {
 	hook.RemoveSubscriber(msgChan2)
 	assert.Equal(t, 0, len(hook.subscribers))
 }
+
+func TestFire_CancelledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	hook := NewBrokerHook(ctx, "info")
+	err := hook.Fire(logrus.NewEntry(logrus.New()))
+	assert.NoError(t, err)
+}
+
+func TestFire_DeliverToSubscriber(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		ctx, cancel := context.WithCancel(t.Context())
+		hook := NewBrokerHook(ctx, "info")
+		msgChan := make(MessageChanType, 1)
+		hook.AddSubscriber(msgChan)
+
+		entry := logrus.NewEntry(logrus.New())
+		entry.Message = "hello broker"
+		entry.Level = logrus.InfoLevel
+		assert.NoError(t, hook.Fire(entry))
+
+		// Wait for the poll goroutine to drain input and deliver to msgChan.
+		synctest.Wait()
+
+		select {
+		case msg := <-msgChan:
+			assert.Contains(t, string(msg), "hello broker")
+		default:
+			t.Fatal("message not delivered")
+		}
+
+		cancel()        // signal poll to exit
+		synctest.Wait() // wait for poll goroutine to return
+	})
+}
+
+func TestFire_HighLoad(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		// Construct without starting poll so the input channel stays full.
+		hook := &BrokerHook{
+			highLoadTimeout: time.Millisecond,
+			input:           make(chan MessageType, 1),
+			ctx:             t.Context(),
+			level:           "info",
+			mu:              &sync.Mutex{},
+			formatter:       defaultFormatter,
+		}
+		hook.input <- "fill" // fill to capacity; no poll goroutine to drain it
+
+		entry := logrus.NewEntry(logrus.New())
+		entry.Message = "to be dropped"
+		// The test goroutine is the only goroutine in the bubble.
+		// Fire blocks on time.After; fake time advances and takes the drop branch.
+		assert.NoError(t, hook.Fire(entry))
+	})
+}
+
+func TestSend_NoSubscribersIsNoop(t *testing.T) {
+	hook := NewBrokerHook(t.Context(), "info")
+	// No subscribers — send must return without blocking or panicking.
+	hook.send("msg")
+}
+
+func TestSend_FullSubscriberDrops(t *testing.T) {
+	hook := NewBrokerHook(t.Context(), "info")
+	// Unbuffered channel with no reader — send must take the default branch and not block.
+	msgChan := make(MessageChanType)
+	hook.AddSubscriber(msgChan)
+	hook.send("dropped")
+}
+
+func TestSetBrokerFormatter_NilResetsDefault(t *testing.T) {
+	hook := NewBrokerHook(t.Context(), "info")
+	hook.SetBrokerFormatter(&logrus.JSONFormatter{})
+	hook.SetBrokerFormatter(nil)
+	assert.Equal(t, defaultFormatter, hook.formatter)
+}
+
+func TestLevels(t *testing.T) {
+	cases := []struct {
+		level    string
+		expected []logrus.Level
+	}{
+		{"none", []logrus.Level{}},
+		{"debug", logrus.AllLevels},
+		{"info", []logrus.Level{logrus.PanicLevel, logrus.FatalLevel, logrus.ErrorLevel, logrus.WarnLevel, logrus.InfoLevel}},
+		{"error", []logrus.Level{logrus.PanicLevel, logrus.FatalLevel, logrus.ErrorLevel}},
+	}
+	for _, tc := range cases {
+		hook := NewBrokerHook(t.Context(), tc.level)
+		assert.Equal(t, tc.expected, hook.Levels(), "level=%s", tc.level)
+	}
+}