*: propagate Logger through drpcmanager, drpcstream, drpcwire, and drpcpool

Thread the Logger interface through the entire DRPC stack so that all
packages have access to structured logging:

- drpcwire.Writer: add NewWriterWithLogger constructor, replace
  drpcdebug.Log calls with Logger.Debugf for buffer flush events.

- drpcstream.Options: add Logger field, default to drpc.DefaultLogger.
  Replace drpcdebug.Log calls with Logger.Debugf for stream lifecycle
  events (HANDLE, SEND, FIN, FLUSH, CALL).

- drpcmanager.Options: add Logger field, default to drpc.DefaultLogger.
  Replace drpcdebug.Log calls with Logger.Debugf for connection management
  events (WAIT, TERM, READ, STREAM, CANCEL, BUSY, UNFIN, CLEAN). Propagate
  Logger to Writer and Stream instances it creates.

- drpcpool.Options: add Logger field, default to drpc.DefaultLogger.
  Replace drpcdebug.Log calls with Logger.Debugf for pool operations
  (CLOSE, TAKEN, PUT).

- drpcserver: propagate its Logger to the managers it creates via
  ServeOne, ensuring the entire call stack shares the same logger.

All packages now use Logger.Debugf instead of the build-tag-gated
drpcdebug.Log. Since DefaultLogger.Debugf is a no-op, production behavior
is unchanged. Custom Logger implementations (e.g. CockroachDB's) can
route these to verbose/conditional logging.

Co-Authored-By: roachdev-claude <roachdev-claude-bot@cockroachlabs.com>

Author: cthumuluru-crdb

debug_off.go

@@ -0,0 +1,11 @@
+// Copyright (C) 2019 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+//go:build !drpcdebug
+
+package drpc
+
+// DebugEnabled controls whether debug logging is active. When false (the
+// default), the compiler eliminates debug log callsites entirely so that
+// callbacks passed to log helpers are never allocated or evaluated.
+const DebugEnabled = false

debug_on.go

@@ -0,0 +1,10 @@
+// Copyright (C) 2019 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+//go:build drpcdebug
+
+package drpc
+
+// DebugEnabled controls whether debug logging is active. Build with
+// -tags=drpcdebug to enable debug log evaluation.
+const DebugEnabled = true

drpcclient/dialoptions.go

@@ -33,6 +33,9 @@ type dialOptions struct {
 	// tlsConfig is an optional TLS configuration for secure connections.
 	tlsConfig *tls.Config
 
+	// logger is used to log errors and operational events on the connection.
+	logger drpc.Logger
+
 	// metrics holds optional metrics the conn will populate. No metrics are
 	// recorded if this is nil. When shouldRecord is set, metrics are recorded
 	// only when shouldRecord returns true.
@@ -94,6 +97,13 @@ func WithShouldRecordFunc(shouldRecord func() bool) DialOption {
 	}
 }
 
+// WithLogger returns a DialOption that sets the Logger for the connection.
+func WithLogger(logger drpc.Logger) DialOption {
+	return func(o *dialOptions) {
+		o.logger = logger
+	}
+}
+
 // WithContextDialer returns a DialOption that sets a custom dialer function
 // to be used instead of the default net.Dialer.
 func WithContextDialer(dialer func(context.Context, string) (net.Conn, error)) DialOption {
@@ -160,6 +170,7 @@ func DialContext(
 			},
 			SoftCancel: false,
 		},
+		Logger:       options.logger,
 		ShouldRecord: options.shouldRecord,
 		Metrics:      *options.metrics,
 	}), nil

drpcconn/conn.go

@@ -25,6 +25,10 @@ type Options struct {
 	// Manager controls the options we pass to the manager of this conn.
 	Manager drpcmanager.Options
 
+	// Logger is used to log errors and operational events. If nil,
+	// drpc.DefaultLogger is used.
+	Logger drpc.Logger
+
 	// TODO: (server): deprecate this
 	// CollectStats controls whether the client should collect stats on the
 	// rpcs it creates.
@@ -77,6 +81,9 @@ func NewWithOptions(tr drpc.Transport, opts Options) *Conn {
 		c.stats = make(map[string]*drpcstats.Stats)
 	}
 
+	if opts.Manager.Logger == nil {
+		opts.Manager.Logger = opts.Logger
+	}
 	c.man = drpcmanager.NewWithOptions(c.tr, opts.Manager)
 
 	return c
@@ -123,7 +130,9 @@ func (c *Conn) Close() (err error) { return c.man.Close() }
 
 // Invoke issues the rpc on the transport serializing in, waits for a response, and
 // deserializes it into out. Only one Invoke or Stream may be open at a time.
-func (c *Conn) Invoke(ctx context.Context, rpc string, enc drpc.Encoding, in, out drpc.Message) (err error) {
+func (c *Conn) Invoke(
+	ctx context.Context, rpc string, enc drpc.Encoding, in, out drpc.Message,
+) (err error) {
 	defer func() { err = drpc.ToRPCErr(err) }()
 
 	var metadata []byte
@@ -155,7 +164,14 @@ func (c *Conn) Invoke(ctx context.Context, rpc string, enc drpc.Encoding, in, ou
 	return nil
 }
 
-func (c *Conn) doInvoke(stream *drpcstream.Stream, enc drpc.Encoding, rpc string, data []byte, metadata []byte, out drpc.Message) (err error) {
+func (c *Conn) doInvoke(
+	stream *drpcstream.Stream,
+	enc drpc.Encoding,
+	rpc string,
+	data []byte,
+	metadata []byte,
+	out drpc.Message,
+) (err error) {
 	if len(metadata) > 0 {
 		if err := stream.RawWrite(drpcwire.KindInvokeMetadata, metadata); err != nil {
 			return err
@@ -178,7 +194,9 @@ func (c *Conn) doInvoke(stream *drpcstream.Stream, enc drpc.Encoding, rpc string
 
 // NewStream begins a streaming rpc on the connection. Only one Invoke or Stream may
 // be open at a time.
-func (c *Conn) NewStream(ctx context.Context, rpc string, enc drpc.Encoding) (_ drpc.Stream, err error) {
+func (c *Conn) NewStream(
+	ctx context.Context, rpc string, enc drpc.Encoding,
+) (_ drpc.Stream, err error) {
 	defer func() { err = drpc.ToRPCErr(err) }()
 
 	var metadata []byte

drpcconn/conn_test.go

@@ -183,6 +183,29 @@ func TestConn_NewStreamSendsGrpcAndDrpcMetadata(t *testing.T) {
 	_ = s.CloseSend()
 }
 
+func TestConnDefaultLogger(t *testing.T) {
+	pc, ps := net.Pipe()
+	defer func() { _ = pc.Close() }()
+	defer func() { _ = ps.Close() }()
+
+	conn := NewWithOptions(pc, Options{})
+	defer func() { _ = conn.Close() }()
+	// Verify construction with nil Logger does not panic.
+	_ = conn
+}
+
+func TestConnCustomLogger(t *testing.T) {
+	pc, ps := net.Pipe()
+	defer func() { _ = pc.Close() }()
+	defer func() { _ = ps.Close() }()
+
+	var logger drpc.InMemLogger
+	conn := NewWithOptions(pc, Options{Logger: &logger})
+	defer func() { _ = conn.Close() }()
+	// Verify construction with custom Logger does not panic.
+	_ = conn
+}
+
 func TestConn_encodeMetadata(t *testing.T) {
 	pc, ps := net.Pipe()
 	defer func() { assert.NoError(t, pc.Close()) }()

drpcmanager/manager.go

@@ -15,9 +15,7 @@ import (
 
 	"github.com/zeebo/errs"
 	grpcmetadata "google.golang.org/grpc/metadata"
-
 	"storj.io/drpc"
-	"storj.io/drpc/drpcdebug"
 	"storj.io/drpc/drpcmetadata"
 	"storj.io/drpc/drpcsignal"
 	"storj.io/drpc/drpcstream"
@@ -60,6 +58,10 @@ type Options struct {
 	// handling. When enabled, the server stream will decode incoming metadata
 	// into grpc metadata in the context.
 	GRPCMetadataCompatMode bool
+
+	// Logger is used to log operational events. If nil, drpc.DefaultLogger is
+	// used.
+	Logger drpc.Logger
 }
 
 // Manager handles the logic of managing a transport for a drpc client or
@@ -100,9 +102,13 @@ func New(tr drpc.Transport) *Manager {
 // NewWithOptions returns a new manager for the transport. It uses the provided
 // options to manage details of how it uses it.
 func NewWithOptions(tr drpc.Transport, opts Options) *Manager {
+	if opts.Logger == nil {
+		opts.Logger = drpc.DefaultLogger
+	}
+
 	m := &Manager{
 		tr:   tr,
-		wr:   drpcwire.NewWriter(tr, opts.WriterBufferSize),
+		wr:   drpcwire.NewWriterWithLogger(tr, opts.WriterBufferSize, opts.Logger),
 		rd:   drpcwire.NewReaderWithOptions(tr, opts.Reader),
 		opts: opts,
 
@@ -135,8 +141,8 @@ func NewWithOptions(tr drpc.Transport, opts Options) *Manager {
 func (m *Manager) String() string { return fmt.Sprintf("<man %p>", m) }
 
 func (m *Manager) log(what string, cb func() string) {
-	if drpcdebug.Enabled {
-		drpcdebug.Log(func() (_, _, _ string) { return m.String(), what, cb() })
+	if drpc.DebugEnabled {
+		m.opts.Logger.Debugf("%s %s %s", m.String(), what, cb())
 	}
 }
 
@@ -298,8 +304,11 @@ func (m *Manager) manageReader() {
 //
 
 // newStream creates a stream value with the appropriate configuration for this manager.
-func (m *Manager) newStream(ctx context.Context, sid uint64, kind drpc.StreamKind, rpc string) (*drpcstream.Stream, error) {
+func (m *Manager) newStream(
+	ctx context.Context, sid uint64, kind drpc.StreamKind, rpc string,
+) (*drpcstream.Stream, error) {
 	opts := m.opts.Stream
+	opts.Logger = m.opts.Logger
 	drpcopts.SetStreamKind(&opts.Internal, kind)
 	drpcopts.SetStreamRPC(&opts.Internal, rpc)
 	if cb := drpcopts.GetManagerStatsCB(&m.opts.Internal); cb != nil {
@@ -425,7 +434,9 @@ func (m *Manager) Close() error {
 }
 
 // NewClientStream starts a stream on the managed transport for use by a client.
-func (m *Manager) NewClientStream(ctx context.Context, rpc string) (stream *drpcstream.Stream, err error) {
+func (m *Manager) NewClientStream(
+	ctx context.Context, rpc string,
+) (stream *drpcstream.Stream, err error) {
 	if err := m.acquireSemaphore(ctx); err != nil {
 		return nil, err
 	}
@@ -436,7 +447,9 @@ func (m *Manager) NewClientStream(ctx context.Context, rpc string) (stream *drpc
 // NewServerStream starts a stream on the managed transport for use by a server.
 // It does this by waiting for the client to issue an invoke message and
 // returning the details.
-func (m *Manager) NewServerStream(ctx context.Context) (stream *drpcstream.Stream, rpc string, err error) {
+func (m *Manager) NewServerStream(
+	ctx context.Context,
+) (stream *drpcstream.Stream, rpc string, err error) {
 	if err := m.acquireSemaphore(ctx); err != nil {
 		return nil, "", err
 	}

drpcmanager/manager_debug_test.go

@@ -0,0 +1,66 @@
+// Copyright (C) 2019 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+//go:build drpcdebug
+
+package drpcmanager
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net"
+	"strings"
+	"testing"
+
+	"github.com/zeebo/assert"
+	"storj.io/drpc"
+	"storj.io/drpc/drpctest"
+	"storj.io/drpc/drpcwire"
+)
+
+func TestManagerLoggerPropagation(t *testing.T) {
+	ctx := drpctest.NewTracker(t)
+	defer ctx.Close()
+
+	cconn, sconn := net.Pipe()
+	defer func() { _ = cconn.Close() }()
+	defer func() { _ = sconn.Close() }()
+
+	var logger drpc.InMemLogger
+
+	cman := NewWithOptions(cconn, Options{Logger: &logger})
+	defer func() { _ = cman.Close() }()
+
+	sman := New(sconn)
+	defer func() { _ = sman.Close() }()
+
+	ctx.Run(func(ctx context.Context) {
+		stream, err := cman.NewClientStream(ctx, "rpc")
+		assert.NoError(t, err)
+		defer func() { _ = stream.Close() }()
+
+		assert.NoError(t, stream.RawWrite(drpcwire.KindInvoke, []byte("invoke")))
+		assert.NoError(t, stream.RawWrite(drpcwire.KindMessage, []byte("message")))
+		assert.NoError(t, stream.RawFlush())
+
+		assert.NoError(t, stream.Close())
+	})
+
+	ctx.Run(func(ctx context.Context) {
+		stream, _, err := sman.NewServerStream(ctx)
+		assert.NoError(t, err)
+		defer func() { _ = stream.Close() }()
+
+		_, err = stream.RawRecv()
+		assert.NoError(t, err)
+
+		_, err = stream.RawRecv()
+		assert.That(t, errors.Is(err, io.EOF))
+	})
+
+	ctx.Wait()
+
+	logs := logger.String()
+	assert.That(t, strings.Contains(logs, "DEBUG:"))
+}

drpcmanager/manager_test.go

@@ -14,8 +14,8 @@ import (
 
 	"github.com/zeebo/assert"
 	grpcmetadata "google.golang.org/grpc/metadata"
+	"storj.io/drpc"
 	"storj.io/drpc/drpcmetadata"
-
 	"storj.io/drpc/drpctest"
 	"storj.io/drpc/drpcwire"
 )
@@ -295,3 +295,18 @@ func (b *blockedTransport) wait(p int, rw *bool) (int, error) {
 func (b *blockedTransport) Read(p []byte) (n int, err error)  { return b.wait(len(p), &b.ro) }
 func (b *blockedTransport) Write(p []byte) (n int, err error) { return b.wait(len(p), &b.wo) }
 func (b *blockedTransport) Close() error                      { return nil }
+
+func TestManagerDefaultLogger(t *testing.T) {
+	tr := make(blockingTransport)
+	man := New(tr)
+	defer func() { _ = man.Close() }()
+	assert.Equal(t, man.opts.Logger, drpc.DefaultLogger)
+}
+
+func TestManagerCustomLogger(t *testing.T) {
+	tr := make(blockingTransport)
+	var logger drpc.InMemLogger
+	man := NewWithOptions(tr, Options{Logger: &logger})
+	defer func() { _ = man.Close() }()
+	assert.Equal(t, man.opts.Logger, &logger)
+}

drpcpool/pool.go

@@ -5,12 +5,11 @@ package drpcpool
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"time"
 
 	"github.com/zeebo/errs"
-	"storj.io/drpc/drpcdebug"
+	"storj.io/drpc"
 	"storj.io/drpc/drpcmetrics"
 )
 
@@ -43,6 +42,10 @@ type Options struct {
 
 	// Labels holds optional labels to be attached to all metrics.
 	Labels map[string]string
+
+	// Logger is used to log operational events. If nil, drpc.DefaultLogger is
+	// used.
+	Logger drpc.Logger
 }
 
 // Pool is a connection pool with key type K. It maintains a cache of connections
@@ -58,6 +61,9 @@ type Pool[K comparable, V Conn] struct {
 
 // New constructs a new Pool with the provided Options.
 func New[K comparable, V Conn](opts Options) *Pool[K, V] {
+	if opts.Logger == nil {
+		opts.Logger = drpc.DefaultLogger
+	}
 	pool := Pool[K, V]{
 		opts:    opts,
 		entries: make(map[K]*list[K, V]),
@@ -112,8 +118,8 @@ func (p *Pool[K, V]) updatePoolSize() {
 }
 
 func (p *Pool[K, V]) log(what string, cb func() string) {
-	if drpcdebug.Enabled {
-		drpcdebug.Log(func() (_, _, _ string) { return fmt.Sprintf("<pül %p>", p), what, cb() })
+	if drpc.DebugEnabled {
+		p.opts.Logger.Debugf("<pool %p> %s %s", p, what, cb())
 	}
 }
 

drpcpool/pool_test.go

@@ -646,6 +646,19 @@ func TestPoolMetrics_ShouldRecordDynamic(t *testing.T) {
 	assert.Equal(t, misses.total, 1.0) // unchanged
 }
 
+func TestPoolDefaultLogger(t *testing.T) {
+	pool := New[string, Conn](Options{})
+	defer func() { _ = pool.Close() }()
+	assert.Equal(t, pool.opts.Logger, drpc.DefaultLogger)
+}
+
+func TestPoolCustomLogger(t *testing.T) {
+	var logger drpc.InMemLogger
+	pool := New[string, Conn](Options{Logger: &logger})
+	defer func() { _ = pool.Close() }()
+	assert.Equal(t, pool.opts.Logger, &logger)
+}
+
 func BenchmarkPool(b *testing.B) {
 	ctx := drpctest.NewTracker(b)
 	defer ctx.Close()