Logging improvements

v2/dht.go

@@ -17,6 +17,7 @@
 	"github.com/libp2p/go-libp2p-kad-dht/v2/internal/coord"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/internal/coord/routing"
 	"github.com/libp2p/go-libp2p-kad-dht/v2/kadt"
+	"github.com/libp2p/go-libp2p-kad-dht/v2/tele"
 )
 
 // DHT is an implementation of Kademlia with S/Kademlia modifications.
@@ -95,7 +96,7 @@
 	} else if cfg.ProtocolID == ProtocolIPFS {
 		d.backends, err = d.initAminoBackends()
 		if err != nil {
 			return nil, fmt.Errorf("init amino backends: %w", err)
 		}
 	}
 
@@ -106,13 +107,23 @@
 
 	// instantiate a new Kademlia DHT coordinator.
 	coordCfg := coord.DefaultCoordinatorConfig()
+	coordCfg.Clock = cfg.Clock
+	coordCfg.Logger = cfg.Logger
+	coordCfg.MeterProvider = cfg.MeterProvider
+	coordCfg.TracerProvider = cfg.TracerProvider
+
+	coordCfg.Query.Clock = cfg.Clock
+	coordCfg.Query.Logger = cfg.Logger.With("behaviour", "pooledquery")
+	coordCfg.Query.Tracer = cfg.TracerProvider.Tracer(tele.TracerName)
 	coordCfg.Query.Concurrency = cfg.Query.Concurrency
 	coordCfg.Query.Timeout = cfg.Query.Timeout
 	coordCfg.Query.RequestConcurrency = cfg.Query.RequestConcurrency
 	coordCfg.Query.RequestTimeout = cfg.Query.RequestTimeout
-	coordCfg.Clock = cfg.Clock
-	coordCfg.MeterProvider = cfg.MeterProvider
-	coordCfg.TracerProvider = cfg.TracerProvider
+
+	coordCfg.Routing.Clock = cfg.Clock
+	coordCfg.Routing.Logger = cfg.Logger.With("behaviour", "routing")
+	coordCfg.Routing.Tracer = cfg.TracerProvider.Tracer(tele.TracerName)
+	coordCfg.Routing.Meter = cfg.MeterProvider.Meter(tele.MeterName)
 
 	d.kad, err = coord.NewCoordinator(kadt.PeerID(d.host.ID()), &router{host: h, ProtocolID: cfg.ProtocolID}, d.rt, coordCfg)
 	if err != nil {
@@ -152,18 +163,18 @@
 	)
 
 	if d.cfg.Datastore != nil {
 		dstore = d.cfg.Datastore
 	} else if dstore, err = InMemoryDatastore(); err != nil {
 		return nil, fmt.Errorf("new default datastore: %w", err)
 	}
 
 	// wrap datastore in open telemetry tracing
 	dstore = trace.New(dstore, d.tele.Tracer)
 
 	pbeCfg, err := DefaultProviderBackendConfig()
 	if err != nil {
 		return nil, fmt.Errorf("default provider config: %w", err)
 	}
 	pbeCfg.Logger = d.cfg.Logger
 	pbeCfg.AddressFilter = d.cfg.AddressFilter
 	pbeCfg.Tele = d.tele
@@ -171,26 +182,26 @@
 
 	pbe, err := NewBackendProvider(d.host.Peerstore(), dstore, pbeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("new provider backend: %w", err)
 	}
 
 	rbeCfg, err := DefaultRecordBackendConfig()
 	if err != nil {
 		return nil, fmt.Errorf("default provider config: %w", err)
 	}
 	rbeCfg.Logger = d.cfg.Logger
 	rbeCfg.Tele = d.tele
 	rbeCfg.clk = d.cfg.Clock
 
 	ipnsBe, err := NewBackendIPNS(dstore, d.host.Peerstore(), rbeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("new ipns backend: %w", err)
 	}
 
 	pkBe, err := NewBackendPublicKey(dstore, rbeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("new public key backend: %w", err)
 	}
 
 	return map[string]Backend{
 		namespaceIPNS:      ipnsBe,
@@ -202,11 +213,11 @@
 // Close cleans up all resources associated with this DHT.
 func (d *DHT) Close() error {
 	if err := d.sub.Close(); err != nil {
-		d.log.With("err", err).Debug("failed closing event bus subscription")
+		d.debugErr(err, "failed closing event bus subscription")
 	}
 
 	if err := d.kad.Close(); err != nil {
-		d.log.With("err", err).Debug("failed closing coordinator")
+		d.debugErr(err, "failed closing coordinator")
 	}
 
 	for ns, b := range d.backends {
@@ -216,7 +227,7 @@
 		}
 
 		if err := closer.Close(); err != nil {
-			d.log.Warn("failed closing backend", "namespace", ns, "err", err.Error())
+			d.warnErr(err, "failed closing backend", "namespace", ns)
 		}
 	}
 
@@ -230,7 +241,7 @@
 	if d.cfg.ProtocolID == ProtocolIPFS && d.cfg.Datastore == nil {
 		if pbe, err := typedBackend[*ProvidersBackend](d, namespaceProviders); err == nil {
 			if err := pbe.datastore.Close(); err != nil {
-				d.log.Warn("failed closing in memory datastore", "err", err.Error())
+				d.warnErr(err, "failed closing in memory datastore")
 			}
 		}
 	}
@@ -244,7 +255,7 @@
 			}
 
 			if err := s.Reset(); err != nil {
-				d.log.With("err", err).Debug("failed closing stream")
+				d.debugErr(err, "failed closing stream")
 			}
 		}
 	}
@@ -302,21 +313,41 @@
 			}
 
 			if err := s.Reset(); err != nil {
-				d.log.With("err", err).Debug("failed closing stream")
+				d.debugErr(err, "failed closing stream")
 			}
 		}
 	}
 }
 
-// logErr is a helper method that uses the slogger of the DHT and writes a
-// warning log line with the given message alongside the error. If the error
+// warnErr is a helper method that uses the slogger of the DHT and writes a
+// warning log line with the given message alongside the error. args is a list of
+// key/value pairs or slog.Attrs that will be included with the log message. If the error
 // is nil, this method is a no-op.
-func (d *DHT) logErr(err error, msg string) {
+func (d *DHT) warnErr(err error, msg string, args ...any) {
 	if err == nil {
 		return
 	}
 
-	d.log.Warn(msg, "err", err.Error())
+	if len(args) == 0 {
+		d.log.Warn(msg, tele.LogAttrError(err))
+		return
+	}
+	d.log.With(args...).Warn(msg, tele.LogAttrError(err))
+}
+
+// debugErr is a helper method that uses the slogger of the DHT and writes a
+// debug log line with the given message alongside the error. args is a list of
+// key/value pairs or slog.Attrs that will be included with the log message. If the error
+// is nil, this method is a no-op.
+func (d *DHT) debugErr(err error, msg string, args ...any) {
+	if err == nil {
+		return
+	}
+	if len(args) == 0 {
+		d.log.Debug(msg, tele.LogAttrError(err))
+		return
+	}
+	d.log.With(args...).Debug(msg, tele.LogAttrError(err))
 }
 
 // AddAddresses suggests peers and their associated addresses to be added to the routing table.

v2/internal/coord/coordinator.go

@@ -154,33 +154,33 @@
 
 func NewCoordinator(self kadt.PeerID, rtr coordt.Router[kadt.Key, kadt.PeerID, *pb.Message], rt routing.RoutingTableCpl[kadt.Key, kadt.PeerID], cfg *CoordinatorConfig) (*Coordinator, error) {
 	if cfg == nil {
 		cfg = DefaultCoordinatorConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	// initialize a new telemetry struct
 	tele, err := NewTelemetry(cfg.MeterProvider, cfg.TracerProvider)
 	if err != nil {
 		return nil, fmt.Errorf("init telemetry: %w", err)
 	}
 
 	queryBehaviour, err := NewPooledQueryBehaviour(self, &cfg.Query)
 	if err != nil {
 		return nil, fmt.Errorf("query behaviour: %w", err)
 	}
 
 	routingBehaviour, err := NewRoutingBehaviour(self, rt, &cfg.Routing)
 	if err != nil {
 		return nil, fmt.Errorf("routing behaviour: %w", err)
 	}
 
 	networkBehaviour := NewNetworkBehaviour(rtr, cfg.Logger, tele.Tracer)
 
 	b, err := brdcst.NewPool[kadt.Key, kadt.PeerID, *pb.Message](self, nil)
 	if err != nil {
 		return nil, fmt.Errorf("broadcast: %w", err)
 	}
 
 	brdcstBehaviour := NewPooledBroadcastBehaviour(b, cfg.Logger, tele.Tracer)
 
@@ -215,8 +215,8 @@
 	return nil
 }
 
 func (c *Coordinator) ID() kadt.PeerID {
 	return c.self
 }
 
 func (c *Coordinator) eventLoop(ctx context.Context) {
@@ -233,14 +233,14 @@
 		case <-ctx.Done():
 			// coordinator is closing
 			return
 		case <-c.networkBehaviour.Ready():
 			ev, ok = c.networkBehaviour.Perform(ctx)
 		case <-c.routingBehaviour.Ready():
 			ev, ok = c.routingBehaviour.Perform(ctx)
 		case <-c.queryBehaviour.Ready():
 			ev, ok = c.queryBehaviour.Perform(ctx)
 		case <-c.brdcstBehaviour.Ready():
 			ev, ok = c.brdcstBehaviour.Perform(ctx)
 		}
 
 		if ok {
@@ -256,10 +256,10 @@
 	switch ev := ev.(type) {
 	case NetworkCommand:
 		c.networkBehaviour.Notify(ctx, ev)
 	case QueryCommand:
 		c.queryBehaviour.Notify(ctx, ev)
 	case BrdcstCommand:
 		c.brdcstBehaviour.Notify(ctx, ev)
 	case RoutingCommand:
 		c.routingBehaviour.Notify(ctx, ev)
 	case RoutingNotification:
@@ -267,8 +267,8 @@
 		rn := c.routingNotifier
 		c.routingNotifierMu.RUnlock()
 		rn.Notify(ctx, ev)
 	default:
 		panic(fmt.Sprintf("unexpected event: %T", ev))
 	}
 }
 
@@ -289,8 +289,8 @@
 
 	nh, err := c.networkBehaviour.getNodeHandler(ctx, id)
 	if err != nil {
 		return nil, err
 	}
 	return nh, nil
 }
 
@@ -303,8 +303,8 @@
 	for _, id := range closest {
 		nh, err := c.networkBehaviour.getNodeHandler(ctx, id)
 		if err != nil {
 			return nil, err
 		}
 		nodes = append(nodes, nh)
 	}
 	return nodes, nil
@@ -312,14 +312,14 @@
 
 // GetValue requests that the node return any value associated with the supplied key.
 // If the node does not have a value for the key it returns ErrValueNotFound.
 func (c *Coordinator) GetValue(ctx context.Context, k kadt.Key) (coordt.Value, error) {
 	panic("not implemented")
 }
 
 // PutValue requests that the node stores a value to be associated with the supplied key.
 // If the node cannot or chooses not to store the value for the key it returns ErrValueNotAccepted.
 func (c *Coordinator) PutValue(ctx context.Context, r coordt.Value, q int) error {
 	panic("not implemented")
 }
 
 // QueryClosest starts a query that attempts to find the closest nodes to the target key.
@@ -336,14 +336,15 @@
 func (c *Coordinator) QueryClosest(ctx context.Context, target kadt.Key, fn coordt.QueryFunc, numResults int) ([]kadt.PeerID, coordt.QueryStats, error) {
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.Query")
 	defer span.End()
+	c.cfg.Logger.Debug("starting query for closest nodes", tele.LogAttrKey(target))
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	seeds, err := c.GetClosestNodes(ctx, target, 20)
 	if err != nil {
 		return nil, coordt.QueryStats{}, err
 	}
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
@@ -381,18 +382,22 @@
 func (c *Coordinator) QueryMessage(ctx context.Context, msg *pb.Message, fn coordt.QueryFunc, numResults int) (coordt.QueryStats, error) {
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.QueryMessage")
 	defer span.End()
+	if msg == nil {
+		return coordt.QueryStats{}, fmt.Errorf("no message supplied for query")
+	}
+	c.cfg.Logger.Debug("starting query with message", tele.LogAttrKey(msg.Target()), slog.String("type", msg.Type.String()))
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	if numResults < 1 {
 		numResults = 20 // TODO: parameterize
 	}
 
 	seeds, err := c.GetClosestNodes(ctx, msg.Target(), numResults)
 	if err != nil {
 		return coordt.QueryStats{}, err
 	}
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
@@ -421,19 +426,23 @@
 func (c *Coordinator) BroadcastRecord(ctx context.Context, msg *pb.Message) error {
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.BroadcastRecord")
 	defer span.End()
+	if msg == nil {
+		return fmt.Errorf("no message supplied for broadcast")
+	}
+	c.cfg.Logger.Debug("starting broadcast with message", tele.LogAttrKey(msg.Target()), slog.String("type", msg.Type.String()))
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
 	seeds, err := c.GetClosestNodes(ctx, msg.Target(), 20) // TODO: parameterize
 	if err != nil {
 		return err
 	}
 
 	seedIDs := make([]kadt.PeerID, 0, len(seeds))
 	for _, s := range seeds {
 		seedIDs = append(seedIDs, s.ID())
 	}
 
 	waiter := NewWaiter[BehaviourEvent]()
 	queryID := c.newOperationID()
@@ -452,8 +461,8 @@
 
 	_, _, err = c.waitForBroadcast(ctx, waiter)
 
 	return err
 }
 
 func (c *Coordinator) waitForQuery(ctx context.Context, queryID coordt.QueryID, waiter *Waiter[BehaviourEvent], fn coordt.QueryFunc) ([]kadt.PeerID, coordt.QueryStats, error) {
 	var lastStats coordt.QueryStats
@@ -461,28 +470,31 @@
 		select {
 		case <-ctx.Done():
 			return nil, lastStats, ctx.Err()
 		case wev := <-waiter.Chan():
 			ctx, ev := wev.Ctx, wev.Event
 			switch ev := ev.(type) {
 			case *EventQueryProgressed:
+				c.cfg.Logger.Debug("query made progress", "query_id", queryID, tele.LogAttrPeerID(ev.NodeID), slog.Duration("elapsed", c.cfg.Clock.Since(ev.Stats.Start)), slog.Int("requests", ev.Stats.Requests), slog.Int("failures", ev.Stats.Failure))
 				lastStats = coordt.QueryStats{
 					Start:    ev.Stats.Start,
 					Requests: ev.Stats.Requests,
 					Success:  ev.Stats.Success,
 					Failure:  ev.Stats.Failure,
 				}
 				nh, err := c.networkBehaviour.getNodeHandler(ctx, ev.NodeID)
 				if err != nil {
 					// ignore unknown node
+					c.cfg.Logger.Debug("node handler not found", "query_id", queryID, tele.LogAttrError, err)
 					break
 				}
 
 				err = fn(ctx, nh.ID(), ev.Response, lastStats)
 				if errors.Is(err, coordt.ErrSkipRemaining) {
 					// done
+					c.cfg.Logger.Debug("query done", "query_id", queryID)
 					c.queryBehaviour.Notify(ctx, &EventStopQuery{QueryID: queryID})
 					return nil, lastStats, nil
 				}
 				if err != nil {
 					// user defined error that terminates the query
 					c.queryBehaviour.Notify(ctx, &EventStopQuery{QueryID: queryID})
@@ -492,10 +504,11 @@
 			case *EventQueryFinished:
 				// query is done
 				lastStats.Exhausted = true
+				c.cfg.Logger.Debug("query ran to exhaustion", "query_id", queryID, slog.Duration("elapsed", ev.Stats.End.Sub(ev.Stats.Start)), slog.Int("requests", ev.Stats.Requests), slog.Int("failures", ev.Stats.Failure))
 				return ev.ClosestNodes, lastStats, nil
 
 			default:
 				panic(fmt.Sprintf("unexpected event: %T", ev))
 			}
 		}
 	}
@@ -503,8 +516,8 @@
 
 func (c *Coordinator) waitForBroadcast(ctx context.Context, waiter *Waiter[BehaviourEvent]) ([]kadt.PeerID, map[string]struct {
 	Node kadt.PeerID
 	Err  error
 }, error,
 ) {
 	for {
 		select {
@@ -517,16 +530,16 @@
 				return ev.Contacted, ev.Errors, nil
 
 			default:
 				panic(fmt.Sprintf("unexpected event: %T", ev))
 			}
 		}
 	}
 }
 
 // AddNodes suggests new DHT nodes to be added to the routing table.
 // If the routing table is updated as a result of this operation an EventRoutingUpdated notification
 // is emitted on the routing notification channel.
 func (c *Coordinator) AddNodes(ctx context.Context, ids []kadt.PeerID) error {
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.AddNodes")
 	defer span.End()
 	for _, id := range ids {
@@ -540,8 +553,8 @@
 		})
 
 	}
 
 	return nil
 }
 
 // Bootstrap instructs the dht to begin bootstrapping the routing table.
@@ -562,6 +575,7 @@
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.NotifyConnectivity")
 	defer span.End()
 
+	c.cfg.Logger.Debug("peer has connectivity", tele.LogAttrPeerID(id), "source", "notify")
 	c.routingBehaviour.Notify(ctx, &EventNotifyConnectivity{
 		NodeID: id,
 	})
@@ -575,20 +589,21 @@
 	ctx, span := c.tele.Tracer.Start(ctx, "Coordinator.NotifyNonConnectivity")
 	defer span.End()
 
+	c.cfg.Logger.Debug("peer has no connectivity", tele.LogAttrPeerID(id), "source", "notify")
 	c.routingBehaviour.Notify(ctx, &EventNotifyNonConnectivity{
 		NodeID: id,
 	})
 
 	return nil
 }
 
 func (c *Coordinator) newOperationID() coordt.QueryID {
 	next := c.lastQueryID.Add(1)
 	return coordt.QueryID(fmt.Sprintf("%016x", next))
 }
 
 // A BufferedRoutingNotifier is a [RoutingNotifier] that buffers [RoutingNotification] events and provides methods
 // to expect occurrences of specific events. It is designed for use in a test environment.
 type BufferedRoutingNotifier struct {
 	mu       sync.Mutex
 	buffered []RoutingNotification
@@ -636,8 +651,8 @@
 
 // ExpectRoutingUpdated blocks until an [EventRoutingUpdated] event is seen for the specified peer id
 func (w *BufferedRoutingNotifier) ExpectRoutingUpdated(ctx context.Context, id kadt.PeerID) (*EventRoutingUpdated, error) {
 	for {
 		// look in buffered events
 		w.mu.Lock()
 		for i, ev := range w.buffered {
 			if tev, ok := ev.(*EventRoutingUpdated); ok {
@@ -662,8 +677,8 @@
 
 // ExpectRoutingRemoved blocks until an [EventRoutingRemoved] event is seen for the specified peer id
 func (w *BufferedRoutingNotifier) ExpectRoutingRemoved(ctx context.Context, id kadt.PeerID) (*EventRoutingRemoved, error) {
 	for {
 		// look in buffered events
 		w.mu.Lock()
 		for i, ev := range w.buffered {
 			if tev, ok := ev.(*EventRoutingRemoved); ok {
@@ -688,4 +703,4 @@
 
 type nullRoutingNotifier struct{}
 
 func (nullRoutingNotifier) Notify(context.Context, RoutingNotification) {}

v2/internal/coord/event.go

@@ -54,8 +54,8 @@
 	SeedNodes []kadt.PeerID
 }
 
 func (*EventStartBootstrap) behaviourEvent() {}
 func (*EventStartBootstrap) routingCommand() {}
 
 type EventOutboundGetCloserNodes struct {
 	QueryID coordt.QueryID
@@ -64,9 +64,9 @@
 	Notify  Notify[BehaviourEvent]
 }
 
 func (*EventOutboundGetCloserNodes) behaviourEvent()     {}
 func (*EventOutboundGetCloserNodes) nodeHandlerRequest() {}
 func (*EventOutboundGetCloserNodes) networkCommand()     {}
 
 type EventOutboundSendMessage struct {
 	QueryID coordt.QueryID
@@ -75,9 +75,9 @@
 	Notify  Notify[BehaviourEvent]
 }
 
 func (*EventOutboundSendMessage) behaviourEvent()     {}
 func (*EventOutboundSendMessage) nodeHandlerRequest() {}
 func (*EventOutboundSendMessage) networkCommand()     {}
 
 type EventStartMessageQuery struct {
 	QueryID           coordt.QueryID
@@ -88,8 +88,8 @@
 	NumResults        int // the minimum number of nodes to successfully contact before considering iteration complete
 }
 
 func (*EventStartMessageQuery) behaviourEvent() {}
 func (*EventStartMessageQuery) queryCommand()   {}
 
 type EventStartFindCloserQuery struct {
 	QueryID           coordt.QueryID
@@ -99,23 +99,23 @@
 	NumResults        int // the minimum number of nodes to successfully contact before considering iteration complete
 }
 
 func (*EventStartFindCloserQuery) behaviourEvent() {}
 func (*EventStartFindCloserQuery) queryCommand()   {}
 
 type EventStopQuery struct {
 	QueryID coordt.QueryID
 }
 
 func (*EventStopQuery) behaviourEvent() {}
 func (*EventStopQuery) queryCommand()   {}
 
 // EventAddNode notifies the routing behaviour of a potential new peer.
 type EventAddNode struct {
 	NodeID kadt.PeerID
 }
 
 func (*EventAddNode) behaviourEvent() {}
 func (*EventAddNode) routingCommand() {}
 
 // EventGetCloserNodesSuccess notifies a behaviour that a GetCloserNodes request, initiated by an
 // [EventOutboundGetCloserNodes] event has produced a successful response.
@@ -126,8 +126,8 @@
 	CloserNodes []kadt.PeerID
 }
 
 func (*EventGetCloserNodesSuccess) behaviourEvent()      {}
 func (*EventGetCloserNodesSuccess) nodeHandlerResponse() {}
 
 // EventGetCloserNodesFailure notifies a behaviour that a GetCloserNodes request, initiated by an
 // [EventOutboundGetCloserNodes] event has failed to produce a valid response.
@@ -138,8 +138,8 @@
 	Err     error
 }
 
 func (*EventGetCloserNodesFailure) behaviourEvent()      {}
 func (*EventGetCloserNodesFailure) nodeHandlerResponse() {}
 
 // EventSendMessageSuccess notifies a behaviour that a SendMessage request, initiated by an
 // [EventOutboundSendMessage] event has produced a successful response.
@@ -151,8 +151,8 @@
 	CloserNodes []kadt.PeerID
 }
 
 func (*EventSendMessageSuccess) behaviourEvent()      {}
 func (*EventSendMessageSuccess) nodeHandlerResponse() {}
 
 // EventSendMessageFailure notifies a behaviour that a SendMessage request, initiated by an
 // [EventOutboundSendMessage] event has failed to produce a valid response.
@@ -164,8 +164,8 @@
 	Err     error
 }
 
 func (*EventSendMessageFailure) behaviourEvent()      {}
 func (*EventSendMessageFailure) nodeHandlerResponse() {}
 
 // EventQueryProgressed is emitted by the coordinator when a query has received a
 // response from a node.
@@ -176,7 +176,7 @@
 	Stats    query.QueryStats
 }
 
 func (*EventQueryProgressed) behaviourEvent() {}
 
 // EventQueryFinished is emitted by the coordinator when a query has finished, either through
 // running to completion or by being canceled.
@@ -186,23 +186,23 @@
 	ClosestNodes []kadt.PeerID
 }
 
 func (*EventQueryFinished) behaviourEvent() {}
 
 // EventRoutingUpdated is emitted by the coordinator when a new node has been verified and added to the routing table.
 type EventRoutingUpdated struct {
 	NodeID kadt.PeerID
 }
 
 func (*EventRoutingUpdated) behaviourEvent()      {}
 func (*EventRoutingUpdated) routingNotification() {}
 
 // EventRoutingRemoved is emitted by the coordinator when new node has been removed from the routing table.
 type EventRoutingRemoved struct {
 	NodeID kadt.PeerID
 }
 
 func (*EventRoutingRemoved) behaviourEvent()      {}
 func (*EventRoutingRemoved) routingNotification() {}
 
 // EventBootstrapFinished is emitted by the coordinator when a bootstrap has finished, either through
 // running to completion or by being canceled.
@@ -210,8 +210,8 @@
 	Stats query.QueryStats
 }
 
 func (*EventBootstrapFinished) behaviourEvent()      {}
 func (*EventBootstrapFinished) routingNotification() {}
 
 // EventNotifyConnectivity notifies a behaviour that a peer's connectivity and support for finding closer nodes
 // has been confirmed such as from a successful query response or an inbound query. This should not be used for
@@ -221,8 +221,8 @@
 	NodeID kadt.PeerID
 }
 
-func (*EventNotifyConnectivity) behaviourEvent()      {}
-func (*EventNotifyConnectivity) routingNotification() {}
+func (*EventNotifyConnectivity) behaviourEvent() {}
+func (*EventNotifyConnectivity) routingCommand() {}
 
 // EventNotifyNonConnectivity notifies a behaviour that a peer does not have connectivity and/or does not support
 // finding closer nodes is known.
@@ -230,11 +230,11 @@
 	NodeID kadt.PeerID
 }
 
 func (*EventNotifyNonConnectivity) behaviourEvent() {}
 func (*EventNotifyNonConnectivity) routingCommand() {}
 
 // EventRoutingPoll notifies a routing behaviour that it may proceed with any pending work.
 type EventRoutingPoll struct{}
 
 func (*EventRoutingPoll) behaviourEvent() {}
 func (*EventRoutingPoll) routingCommand() {}

v2/internal/coord/query.go

@@ -119,10 +119,10 @@
 
 func NewPooledQueryBehaviour(self kadt.PeerID, cfg *PooledQueryConfig) (*PooledQueryBehaviour, error) {
 	if cfg == nil {
 		cfg = DefaultPooledQueryConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	qpCfg := query.DefaultPoolConfig()
 	qpCfg.Clock = cfg.Clock
@@ -133,8 +133,8 @@
 
 	pool, err := query.NewPool[kadt.Key, kadt.PeerID, *pb.Message](self, qpCfg)
 	if err != nil {
 		return nil, fmt.Errorf("query pool: %w", err)
 	}
 
 	h := &PooledQueryBehaviour{
 		cfg:     *cfg,
@@ -200,6 +200,7 @@
 		}
 	case *EventGetCloserNodesFailure:
 		// queue an event that will notify the routing behaviour of a failed node
+		p.cfg.Logger.Debug("peer has no connectivity", tele.LogAttrPeerID(ev.To), "source", "query")
 		p.pending = append(p.pending, &EventNotifyNonConnectivity{
 			ev.To,
 		})
@@ -231,6 +232,7 @@
 		}
 	case *EventSendMessageFailure:
 		// queue an event that will notify the routing behaviour of a failed node
+		p.cfg.Logger.Debug("peer has no connectivity", tele.LogAttrPeerID(ev.To), "source", "query")
 		p.pending = append(p.pending, &EventNotifyNonConnectivity{
 			ev.To,
 		})
@@ -240,8 +242,8 @@
 			QueryID: ev.QueryID,
 			Error:   ev.Err,
 		}
 	default:
 		panic(fmt.Sprintf("unexpected dht event: %T", ev))
 	}
 
 	// attempt to advance the query pool
@@ -285,14 +287,14 @@
 		}
 
 		// attempt to advance the query pool
 		ev, ok := p.advancePool(ctx, &query.EventPoolPoll{})
 		if ok {
 			return ev, true
 		}
 
 		if len(p.pending) == 0 {
 			return nil, false
 		}
 	}
 }
 
@@ -319,9 +321,9 @@
 			Message: st.Message,
 			Notify:  p,
 		}, true
 	case *query.StatePoolWaitingAtCapacity:
 		// nothing to do except wait for message response or timeout
 	case *query.StatePoolWaitingWithCapacity:
 		// nothing to do except wait for message response or timeout
 	case *query.StatePoolQueryFinished[kadt.Key, kadt.PeerID]:
 		waiter, ok := p.waiters[st.QueryID]
@@ -333,12 +335,12 @@
 			})
 			waiter.Close()
 		}
 	case *query.StatePoolQueryTimeout:
 		// TODO
 	case *query.StatePoolIdle:
 		// nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected pool state: %T", st))
 	}
 
 	return nil, false

v2/internal/coord/routing.go

@@ -309,10 +309,10 @@
 
 func NewRoutingBehaviour(self kadt.PeerID, rt routing.RoutingTableCpl[kadt.Key, kadt.PeerID], cfg *RoutingConfig) (*RoutingBehaviour, error) {
 	if cfg == nil {
 		cfg = DefaultRoutingConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	bootstrapCfg := routing.DefaultBootstrapConfig()
 	bootstrapCfg.Clock = cfg.Clock
@@ -324,8 +324,8 @@
 
 	bootstrap, err := routing.NewBootstrap[kadt.Key](self, bootstrapCfg)
 	if err != nil {
 		return nil, fmt.Errorf("bootstrap: %w", err)
 	}
 
 	includeCfg := routing.DefaultIncludeConfig()
 	includeCfg.Clock = cfg.Clock
@@ -337,8 +337,8 @@
 
 	include, err := routing.NewInclude[kadt.Key, kadt.PeerID](rt, includeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("include: %w", err)
 	}
 
 	probeCfg := routing.DefaultProbeConfig()
 	probeCfg.Clock = cfg.Clock
@@ -350,8 +350,8 @@
 
 	probe, err := routing.NewProbe[kadt.Key](rt, probeCfg)
 	if err != nil {
 		return nil, fmt.Errorf("probe: %w", err)
 	}
 
 	exploreCfg := routing.DefaultExploreConfig()
 	exploreCfg.Clock = cfg.Clock
@@ -363,13 +363,13 @@
 
 	schedule, err := routing.NewDynamicExploreSchedule(cfg.ExploreMaximumCpl, cfg.Clock.Now(), cfg.ExploreInterval, cfg.ExploreIntervalMultiplier, cfg.ExploreIntervalJitter)
 	if err != nil {
 		return nil, fmt.Errorf("explore schedule: %w", err)
 	}
 
 	explore, err := routing.NewExplore[kadt.Key](self, rt, cplutil.GenRandPeerID, schedule, exploreCfg)
 	if err != nil {
 		return nil, fmt.Errorf("explore: %w", err)
 	}
 
 	return ComposeRoutingBehaviour(self, bootstrap, include, probe, explore, cfg)
 }
@@ -385,10 +385,10 @@
 	cfg *RoutingConfig,
 ) (*RoutingBehaviour, error) {
 	if cfg == nil {
 		cfg = DefaultRoutingConfig()
 	} else if err := cfg.Validate(); err != nil {
 		return nil, err
 	}
 
 	r := &RoutingBehaviour{
 		self:      self,
@@ -452,8 +452,8 @@
 		// attempt to advance the probe state machine
 		next, ok := r.advanceProbe(ctx, cmd)
 		if ok {
 			r.pending = append(r.pending, next)
 		}
 
 	case *EventGetCloserNodesSuccess:
 		span.SetAttributes(attribute.String("event", "EventGetCloserNodesSuccess"), attribute.String("queryid", string(ev.QueryID)), attribute.String("nodeid", ev.To.String()))
@@ -483,35 +483,35 @@
 					NodeID: ev.To,
 				}
 			} else {
 				cmd = &routing.EventIncludeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 					NodeID: ev.To,
 					Error:  fmt.Errorf("response did not include any closer nodes"),
 				}
 			}
 			// attempt to advance the include
 			next, ok := r.advanceInclude(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		case ProbeQueryID:
 			var cmd routing.ProbeEvent
 			// require that the node responded with at least one closer node
 			if len(ev.CloserNodes) > 0 {
 				cmd = &routing.EventProbeConnectivityCheckSuccess[kadt.Key, kadt.PeerID]{
 					NodeID: ev.To,
 				}
 			} else {
 				cmd = &routing.EventProbeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 					NodeID: ev.To,
 					Error:  fmt.Errorf("response did not include any closer nodes"),
 				}
 			}
 			// attempt to advance the probe state machine
 			next, ok := r.advanceProbe(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		case routing.ExploreQueryID:
 			for _, info := range ev.CloserNodes {
@@ -525,11 +525,11 @@
 			}
 			next, ok := r.advanceExplore(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		default:
 			panic(fmt.Sprintf("unexpected query id: %s", ev.QueryID))
 		}
 	case *EventGetCloserNodesFailure:
 		span.SetAttributes(attribute.String("event", "EventGetCloserNodesFailure"), attribute.String("queryid", string(ev.QueryID)), attribute.String("nodeid", ev.To.String()))
@@ -543,8 +543,8 @@
 			// attempt to advance the bootstrap
 			next, ok := r.advanceBootstrap(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 		case IncludeQueryID:
 			cmd := &routing.EventIncludeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 				NodeID: ev.To,
@@ -553,18 +553,18 @@
 			// attempt to advance the include state machine
 			next, ok := r.advanceInclude(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 		case ProbeQueryID:
 			cmd := &routing.EventProbeConnectivityCheckFailure[kadt.Key, kadt.PeerID]{
 				NodeID: ev.To,
 				Error:  ev.Err,
 			}
 			// attempt to advance the probe state machine
 			next, ok := r.advanceProbe(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 		case routing.ExploreQueryID:
 			cmd := &routing.EventExploreFindCloserFailure[kadt.Key, kadt.PeerID]{
 				NodeID: ev.To,
@@ -573,18 +573,20 @@
 			// attempt to advance the explore
 			next, ok := r.advanceExplore(ctx, cmd)
 			if ok {
 				r.pending = append(r.pending, next)
 			}
 
 		default:
 			panic(fmt.Sprintf("unexpected query id: %s", ev.QueryID))
 		}
 	case *EventNotifyConnectivity:
 		span.SetAttributes(attribute.String("event", "EventNotifyConnectivity"), attribute.String("nodeid", ev.NodeID.String()))
 		// ignore self
 		if r.self.Equal(ev.NodeID) {
 			break
 		}
+		r.cfg.Logger.Debug("peer has connectivity", tele.LogAttrPeerID(ev.NodeID))
+
 		// tell the include state machine in case this is a new peer that could be added to the routing table
 		cmd := &routing.EventIncludeAddCandidate[kadt.Key, kadt.PeerID]{
 			NodeID: ev.NodeID,
@@ -600,8 +602,8 @@
 		}
 		nextProbe, ok := r.advanceProbe(ctx, cmdProbe)
 		if ok {
 			r.pending = append(r.pending, nextProbe)
 		}
 	case *EventNotifyNonConnectivity:
 		span.SetAttributes(attribute.String("event", "EventNotifyConnectivity"), attribute.String("nodeid", ev.NodeID.String()))
 
@@ -611,13 +613,13 @@
 		}
 		nextProbe, ok := r.advanceProbe(ctx, cmdProbe)
 		if ok {
 			r.pending = append(r.pending, nextProbe)
 		}
 	case *EventRoutingPoll:
 		r.pollChildren(ctx)
 
 	default:
 		panic(fmt.Sprintf("unexpected dht event: %T", ev))
 	}
 
 	if len(r.pending) > 0 {
@@ -660,8 +662,8 @@
 
 		// finally check if any pending events were accumulated in the meantime
 		if len(r.pending) == 0 {
 			return nil, false
 		}
 	}
 }
 
@@ -669,13 +671,13 @@
 func (r *RoutingBehaviour) pollChildren(ctx context.Context) {
 	ev, ok := r.advanceBootstrap(ctx, &routing.EventBootstrapPoll{})
 	if ok {
 		r.pending = append(r.pending, ev)
 	}
 
 	ev, ok = r.advanceInclude(ctx, &routing.EventIncludePoll{})
 	if ok {
 		r.pending = append(r.pending, ev)
 	}
 
 	ev, ok = r.advanceProbe(ctx, &routing.EventProbePoll{})
 	if ok {
@@ -702,16 +704,17 @@
 			Notify:  r,
 		}, true
 
 	case *routing.StateBootstrapWaiting:
 		// bootstrap waiting for a message response, nothing to do
 	case *routing.StateBootstrapFinished:
+		r.cfg.Logger.Debug("bootstrap finished", slog.Duration("elapsed", st.Stats.End.Sub(st.Stats.Start)), slog.Int("requests", st.Stats.Requests), slog.Int("failures", st.Stats.Failure))
 		return &EventBootstrapFinished{
 			Stats: st.Stats,
 		}, true
 	case *routing.StateBootstrapIdle:
 		// bootstrap not running, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected bootstrap state: %T", st))
 	}
 
 	return nil, false
@@ -726,6 +729,7 @@
 	case *routing.StateIncludeConnectivityCheck[kadt.Key, kadt.PeerID]:
 		span.SetAttributes(attribute.String("out_event", "EventOutboundGetCloserNodes"))
 		// include wants to send a find node message to a node
+		r.cfg.Logger.Debug("starting connectivity check", tele.LogAttrPeerID(st.NodeID), "source", "include")
 		return &EventOutboundGetCloserNodes{
 			QueryID: IncludeQueryID,
 			To:      st.NodeID,
@@ -743,19 +747,20 @@
 
 		// return the event to notify outwards too
 		span.SetAttributes(attribute.String("out_event", "EventRoutingUpdated"))
+		r.cfg.Logger.Debug("peer added to routing table", tele.LogAttrPeerID(st.NodeID))
 		return &EventRoutingUpdated{
 			NodeID: st.NodeID,
 		}, true
 	case *routing.StateIncludeWaitingAtCapacity:
 		// nothing to do except wait for message response or timeout
 	case *routing.StateIncludeWaitingWithCapacity:
 		// nothing to do except wait for message response or timeout
 	case *routing.StateIncludeWaitingFull:
 		// nothing to do except wait for message response or timeout
 	case *routing.StateIncludeIdle:
 		// nothing to do except wait for new nodes to be added to queue
 	default:
 		panic(fmt.Sprintf("unexpected include state: %T", st))
 	}
 
 	return nil, false
@@ -768,6 +773,7 @@
 	switch st := st.(type) {
 	case *routing.StateProbeConnectivityCheck[kadt.Key, kadt.PeerID]:
 		// include wants to send a find node message to a node
+		r.cfg.Logger.Debug("starting connectivity check", tele.LogAttrPeerID(st.NodeID), "source", "probe")
 		return &EventOutboundGetCloserNodes{
 			QueryID: ProbeQueryID,
 			To:      st.NodeID,
@@ -778,6 +784,7 @@
 		// a node has failed a connectivity check and been removed from the routing table and the probe list
 
 		// emit an EventRoutingRemoved event to notify clients that the node has been removed
+		r.cfg.Logger.Debug("peer removed from routing table", tele.LogAttrPeerID(st.NodeID))
 		r.pending = append(r.pending, &EventRoutingRemoved{
 			NodeID: st.NodeID,
 		})
@@ -786,17 +793,17 @@
 		r.notify(ctx, &EventAddNode{
 			NodeID: st.NodeID,
 		})
 	case *routing.StateProbeWaitingAtCapacity:
 		// the probe state machine is waiting for responses for checks and the maximum number of concurrent checks has been reached.
 		// nothing to do except wait for message response or timeout
 	case *routing.StateProbeWaitingWithCapacity:
 		// the probe state machine is waiting for responses for checks but has capacity to perform more
 		// nothing to do except wait for message response or timeout
 	case *routing.StateProbeIdle:
 		// the probe state machine is not running any checks.
 		// nothing to do except wait for message response or timeout
 	default:
 		panic(fmt.Sprintf("unexpected include state: %T", st))
 	}
 
 	return nil, false
@@ -809,6 +816,7 @@
 	switch st := bstate.(type) {
 
 	case *routing.StateExploreFindCloser[kadt.Key, kadt.PeerID]:
+		r.cfg.Logger.Debug("starting explore", slog.Int("cpl", st.Cpl), tele.LogAttrPeerID(st.NodeID))
 		return &EventOutboundGetCloserNodes{
 			QueryID: routing.ExploreQueryID,
 			To:      st.NodeID,
@@ -816,18 +824,18 @@
 			Notify:  r,
 		}, true
 
 	case *routing.StateExploreWaiting:
 		// explore waiting for a message response, nothing to do
 	case *routing.StateExploreQueryFinished:
 		// nothing to do except notify via telemetry
 	case *routing.StateExploreQueryTimeout:
 		// nothing to do except notify via telemetry
 	case *routing.StateExploreFailure:
-		r.cfg.Logger.Warn("explore failure", "cpl", st.Cpl, "error", st.Error)
+		r.cfg.Logger.Warn("explore failure", slog.Int("cpl", st.Cpl), tele.LogAttrError(st.Error))
 	case *routing.StateExploreIdle:
 		// bootstrap not running, nothing to do
 	default:
 		panic(fmt.Sprintf("unexpected explore state: %T", st))
 	}
 
 	return nil, false

v2/routing.go

@@ -47,8 +47,8 @@
 	var foundPeer peer.ID
 	fn := func(ctx context.Context, visited kadt.PeerID, msg *pb.Message, stats coordt.QueryStats) error {
 		if peer.ID(visited) == id {
 			foundPeer = peer.ID(visited)
 			return coordt.ErrSkipRemaining
 		}
 		return nil
 	}
@@ -62,7 +62,7 @@
 		return peer.AddrInfo{}, fmt.Errorf("peer record not found")
 	}
 
 	return d.host.Peerstore().PeerInfo(foundPeer), nil
 }
 
 func (d *DHT) Provide(ctx context.Context, c cid.Cid, brdcst bool) error {
@@ -140,10 +140,10 @@
 
 	ps, ok := stored.(*providerSet)
 	if !ok {
 		span.RecordError(err)
 		d.log.Warn("Stored value is not a provider set", slog.String("cid", c.String()), slog.String("type", fmt.Sprintf("%T", stored)))
 		return
 	}
 
 	// send all providers onto the out channel until the desired count
 	// was reached. If no count was specified, continue with network lookup.
@@ -183,8 +183,8 @@
 
 			// actually send the provider information to the user
 			select {
 			case <-ctx.Done():
 				return coordt.ErrSkipRemaining
 			case out <- provider:
 			}
 
@@ -201,10 +201,10 @@
 
 	_, err = d.kad.QueryMessage(ctx, msg, fn, 20) // TODO: parameterize
 	if err != nil {
 		span.RecordError(err)
 		d.log.Warn("Failed querying", slog.String("cid", c.String()), slog.String("err", err.Error()))
 		return
 	}
 }
 
 // PutValue satisfies the [routing.Routing] interface and will add the given
@@ -219,8 +219,8 @@
 	// first parse the routing options
 	rOpt := routing.Options{} // routing config
 	if err := rOpt.Apply(opts...); err != nil {
 		return fmt.Errorf("apply routing options: %w", err)
 	}
 
 	// then always store the given value locally
 	if err := d.putValueLocal(ctx, keyStr, value); err != nil {
@@ -234,19 +234,19 @@
 
 	// construct Kademlia-key. Yes, we hash the complete key string which
 	// includes the namespace prefix.
 	msg := &pb.Message{
 		Type:   pb.Message_PUT_VALUE,
 		Key:    []byte(keyStr),
 		Record: record.MakePutRecord(keyStr, value),
 	}
 
 	// finally, find the closest peers to the target key.
 	err := d.kad.BroadcastRecord(ctx, msg)
 	if err != nil {
 		return fmt.Errorf("query error: %w", err)
 	}
 
 	return nil
 }
 
 // putValueLocal stores a value in the local datastore without querying the network.
@@ -281,11 +281,11 @@
 
 	v, err := d.getValueLocal(ctx, key)
 	if err == nil {
 		return v, nil
 	}
 	if !errors.Is(err, ds.ErrNotFound) {
 		return nil, fmt.Errorf("put value locally: %w", err)
 	}
 
 	req := &pb.Message{
 		Type: pb.Message_GET_VALUE,
@@ -296,16 +296,16 @@
 	var value []byte
 	fn := func(ctx context.Context, id kadt.PeerID, resp *pb.Message, stats coordt.QueryStats) error {
 		if resp == nil {
 			return nil
 		}
 
 		if resp.GetType() != pb.Message_GET_VALUE {
 			return nil
 		}
 
 		if string(resp.GetKey()) != key {
 			return nil
 		}
 
 		value = resp.GetRecord().GetValue()
 
@@ -314,8 +314,8 @@
 
 	_, err = d.kad.QueryMessage(ctx, req, fn, d.cfg.BucketSize)
 	if err != nil {
 		return nil, fmt.Errorf("failed to run query: %w", err)
 	}
 
 	return value, nil
 }
@@ -342,8 +342,8 @@
 
 	rec, ok := val.(*recpb.Record)
 	if !ok {
 		return nil, fmt.Errorf("expected *recpb.Record from backend, got: %T", val)
 	}
 
 	return rec.GetValue(), nil
 }
@@ -356,17 +356,18 @@
 }
 
 func (d *DHT) Bootstrap(ctx context.Context) error {
 	ctx, span := d.tele.Tracer.Start(ctx, "DHT.Bootstrap")
 	defer span.End()
+	d.log.Info("Starting bootstrap")
 
 	seed := make([]kadt.PeerID, len(d.cfg.BootstrapPeers))
 	for i, addrInfo := range d.cfg.BootstrapPeers {
 		seed[i] = kadt.PeerID(addrInfo.ID)
 		// TODO: how to handle TTL if BootstrapPeers become dynamic and don't
 		// point to stable peers or consist of ephemeral peers that we have
 		// observed during a previous run.
 		d.host.Peerstore().AddAddrs(addrInfo.ID, addrInfo.Addrs, peerstore.PermanentAddrTTL)
 	}
 
 	return d.kad.Bootstrap(ctx, seed)
 }

v2/stream.go

@@ -42,18 +42,18 @@
 
 	if err := s.Scope().SetService(ServiceName); err != nil {
 		d.log.LogAttrs(ctx, slog.LevelWarn, "error attaching stream to DHT service", slog.String("err", err.Error()))
-		d.logErr(s.Reset(), "failed to reset stream")
+		d.warnErr(s.Reset(), "failed to reset stream")
 		span.RecordError(err)
 		return
 	}
 
 	if err := d.handleNewStream(ctx, s); err != nil {
 		// If we exited with an error, let the remote peer know.
-		d.logErr(s.Reset(), "failed to reset stream")
+		d.warnErr(s.Reset(), "failed to reset stream")
 		span.RecordError(err)
 	} else {
 		// If we exited without an error, close gracefully.
-		d.logErr(s.Close(), "failed to close stream")
+		d.warnErr(s.Close(), "failed to close stream")
 	}
 }
 

v2/tele/log.go

@@ -2,10 +2,33 @@ package tele
 
 import (
 	logging "github.com/ipfs/go-log/v2"
+	"github.com/libp2p/go-libp2p-kad-dht/v2/kadt"
 	"go.uber.org/zap/exp/zapslog"
 	"golang.org/x/exp/slog"
 )
 
-func DefaultLogger(system string) *slog.Logger {
-	return slog.New(zapslog.NewHandler(logging.Logger(system).Desugar().Core()))
+func DefaultLogger(name string) *slog.Logger {
+	return slog.New(zapslog.NewHandler(logging.Logger(name).Desugar().Core()))
+}
+
+// Attributes that can be used with logging or tracing
+const (
+	AttrKeyError    = "error"
+	AttrKeyPeerID   = "peer_id"
+	AttrKeyKey      = "key"
+	AttrKeyCacheHit = "hit"
+	AttrKeyInEvent  = "in_event"
+	AttrKeyOutEvent = "out_event"
+)
+
+func LogAttrError(err error) slog.Attr {
+	return slog.Attr{Key: AttrKeyError, Value: slog.AnyValue(err)}
+}
+
+func LogAttrPeerID(id kadt.PeerID) slog.Attr {
+	return slog.String(AttrKeyPeerID, id.String())
+}
+
+func LogAttrKey(kk kadt.Key) slog.Attr {
+	return slog.String(AttrKeyKey, kk.HexString())
 }

v2/tele/tele.go

@@ -79,11 +79,11 @@ func AttrInstanceID(instanceID string) attribute.KeyValue {
 }
 
 func AttrPeerID(pid string) attribute.KeyValue {
-	return attribute.String("peer_id", pid)
+	return attribute.String(AttrKeyPeerID, pid)
 }
 
 func AttrCacheHit(hit bool) attribute.KeyValue {
-	return attribute.Bool("hit", hit)
+	return attribute.Bool(AttrKeyCacheHit, hit)
 }
 
 // AttrRecordType is currently only used for the provider backend LRU cache
@@ -101,12 +101,12 @@ func AttrKey(val string) attribute.KeyValue {
 
 // AttrInEvent creates an attribute that records the type of an event
 func AttrInEvent(t any) attribute.KeyValue {
-	return attribute.String("in_event", fmt.Sprintf("%T", t))
+	return attribute.String(AttrKeyInEvent, fmt.Sprintf("%T", t))
 }
 
 // AttrOutEvent creates an attribute that records the type of an event being returned
 func AttrOutEvent(t any) attribute.KeyValue {
-	return attribute.String("out_event", fmt.Sprintf("%T", t))
+	return attribute.String(AttrKeyOutEvent, fmt.Sprintf("%T", t))
 }
 
 // WithAttributes is a function that attaches the provided attributes to the