add log-drop package (#15670)

* add log-drop package

* refactor to extract level

* extract metrics

* Apply suggestions from code review

Co-authored-by: Dan Upton <daniel@floppy.co>

* fix compile errors

* change to implement a log sink

* fix tests to remove sleep

* rename and add go docs

* fix expending variadic

Co-authored-by: Dan Upton <daniel@floppy.co>

agent/log-drop/log-drop.go

@@ -0,0 +1,66 @@
+package logdrop
+
+import (
+	"context"
+	"github.com/hashicorp/go-hclog"
+)
+
+// SinkAdapter mimic the interface from hclog.SinkAdapter
+//
+//go:generate mockery --name SinkAdapter --inpackage
+type SinkAdapter interface {
+	Accept(name string, level hclog.Level, msg string, args ...interface{})
+}
+
+type Log struct {
+	n string
+	s string
+	i []interface{}
+	l hclog.Level
+}
+
+type logDropSink struct {
+	sink   SinkAdapter
+	logCh  chan Log
+	name   string
+	dropFn func(l Log)
+}
+
+// Accept consume a log and push it into a channel,
+// if the channel is filled it will call dropFn
+func (r *logDropSink) Accept(name string, level hclog.Level, msg string, args ...interface{}) {
+	r.pushLog(Log{n: name, l: level, s: msg, i: args})
+}
+
+func (r *logDropSink) pushLog(l Log) {
+	select {
+	case r.logCh <- l:
+	default:
+		r.dropFn(l)
+	}
+}
+
+func (r *logDropSink) logConsumer(ctx context.Context) {
+	for {
+		select {
+		case l := <-r.logCh:
+			r.sink.Accept(l.n, l.l, l.s, l.i)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// NewLogDropSink create a log SinkAdapter that wrap another SinkAdapter
+// It also create a go routine for consuming logs, the given context need to be canceled
+// to properly deallocate the SinkAdapter.
+func NewLogDropSink(ctx context.Context, name string, depth int, sink SinkAdapter, dropFn func(l Log)) hclog.SinkAdapter {
+	r := &logDropSink{
+		sink:   sink,
+		logCh:  make(chan Log, depth),
+		name:   name,
+		dropFn: dropFn,
+	}
+	go r.logConsumer(ctx)
+	return r
+}

agent/log-drop/log-drop_test.go

@@ -0,0 +1,51 @@
+package logdrop
+
+import (
+	"context"
+	"github.com/hashicorp/consul/sdk/testutil/retry"
+	"github.com/hashicorp/go-hclog"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"testing"
+	"time"
+)
+
+func TestNewLogDrop(t *testing.T) {
+	mockLogger := NewMockSinkAdapter(t)
+	mockLogger.On("Accept", "test Accept", hclog.Info, "hello", []interface{}{"test", 0}).Return()
+	ld := NewLogDropSink(context.Background(), "test", 10, mockLogger, func(_ Log) {})
+	require.NotNil(t, ld)
+	ld.Accept("test Accept", hclog.Info, "hello", "test", 0)
+	retry.Run(t, func(r *retry.R) {
+		mockLogger.AssertNumberOfCalls(r, "Accept", 1)
+	})
+
+}
+
+func TestLogDroppedWhenChannelFilled(t *testing.T) {
+	mockLogger := NewMockSinkAdapter(t)
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	block := make(chan interface{})
+	mockLogger.On("Accept", "test", hclog.Debug, "hello", []interface{}(nil)).Run(func(args mock.Arguments) {
+		<-block
+	})
+
+	var called = make(chan interface{})
+	ld := NewLogDropSink(ctx, "test", 1, mockLogger, func(l Log) {
+		close(called)
+	})
+	for i := 0; i < 2; i++ {
+		ld.Accept("test", hclog.Debug, "hello")
+	}
+
+	select {
+	case <-called:
+		close(block)
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("timeout waiting for drop func to be called")
+	}
+	retry.Run(t, func(r *retry.R) {
+		mockLogger.AssertNumberOfCalls(r, "Accept", 1)
+	})
+}

agent/log-drop/mock_SinkAdapter.go

@@ -0,0 +1,33 @@
+// Code generated by mockery v2.12.2. DO NOT EDIT.
+
+package logdrop
+
+import (
+	hclog "github.com/hashicorp/go-hclog"
+	mock "github.com/stretchr/testify/mock"
+
+	testing "testing"
+)
+
+// MockSinkAdapter is an autogenerated mock type for the SinkAdapter type
+type MockSinkAdapter struct {
+	mock.Mock
+}
+
+// Accept provides a mock function with given fields: name, level, msg, args
+func (_m *MockSinkAdapter) Accept(name string, level hclog.Level, msg string, args ...interface{}) {
+	var _ca []interface{}
+	_ca = append(_ca, name, level, msg)
+	_ca = append(_ca, args...)
+	_m.Called(_ca...)
+}
+
+// NewMockSinkAdapter creates a new instance of MockSinkAdapter. It also registers the testing.TB interface on the mock and a cleanup function to assert the mocks expectations.
+func NewMockSinkAdapter(t testing.TB) *MockSinkAdapter {
+	mock := &MockSinkAdapter{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

sdk/testutil/retry/retry.go

@@ -37,6 +37,10 @@ type R struct {
 	output []string
 }
 
+func (r *R) Logf(format string, args ...interface{}) {
+	r.log(fmt.Sprintf(format, args...))
+}
+
 func (r *R) Helper() {}
 
 var runFailed = struct{}{}