server: add Node to stream closed callbacks (#572)

Signed-off-by: Bing Han <[email protected]>

Author: tony612

docs/server.md

@@ -22,6 +22,7 @@ import (
 	"log"
 	"sync"
 
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 )
 
@@ -46,9 +47,9 @@ func (cb *Callbacks) OnStreamOpen(_ context.Context, id int64, typ string) error
 	}
 	return nil
 }
-func (cb *Callbacks) OnStreamClosed(id int64) {
+func (cb *Callbacks) OnStreamClosed(id int64, node *core.Node) {
 	if cb.Debug {
-		log.Printf("stream %d closed\n", id)
+		log.Printf("stream %d of node %s closed\n", id, node.Id)
 	}
 }
 func (cb *Callbacks) OnDeltaStreamOpen(_ context.Context, id int64, typ string) error {
@@ -57,9 +58,9 @@ func (cb *Callbacks) OnDeltaStreamOpen(_ context.Context, id int64, typ string)
 	}
 	return nil
 }
-func (cb *Callbacks) OnDeltaStreamClosed(id int64) {
+func (cb *Callbacks) OnDeltaStreamClosed(id int64, node *core.Node) {
 	if cb.Debug {
-		log.Printf("delta stream %d closed\n", id)
+		log.Printf("delta stream %d of node %s closed\n", id, node.Id)
 	}
 }
 func (cb *Callbacks) OnStreamRequest(int64, *discovery.DiscoveryRequest) error {

pkg/server/delta/v3/server.go

@@ -26,7 +26,7 @@ type Callbacks interface {
 	// Returning an error will end processing and close the stream. OnStreamClosed will still be called.
 	OnDeltaStreamOpen(context.Context, int64, string) error
 	// OnDeltaStreamClosed is called immediately prior to closing an xDS stream with a stream ID.
-	OnDeltaStreamClosed(int64)
+	OnDeltaStreamClosed(int64, *core.Node)
 	// OnStreamDeltaRequest is called once a request is received on a stream.
 	// Returning an error will end processing and close the stream. OnStreamClosed will still be called.
 	OnStreamDeltaRequest(int64, *discovery.DeltaDiscoveryRequest) error
@@ -63,10 +63,12 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
 	// a collection of stack allocated watches per request type
 	watches := newWatches()
 
+	var node = &core.Node{}
+
 	defer func() {
 		watches.Cancel()
 		if s.callbacks != nil {
-			s.callbacks.OnDeltaStreamClosed(streamID)
+			s.callbacks.OnDeltaStreamClosed(streamID, node)
 		}
 	}()
 
@@ -96,7 +98,6 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
 		}
 	}
 
-	var node = &core.Node{}
 	for {
 		select {
 		case <-s.ctx.Done():

pkg/server/sotw/v3/server.go

@@ -41,7 +41,7 @@ type Callbacks interface {
 	// Returning an error will end processing and close the stream. OnStreamClosed will still be called.
 	OnStreamOpen(context.Context, int64, string) error
 	// OnStreamClosed is called immediately prior to closing an xDS stream with a stream ID.
-	OnStreamClosed(int64)
+	OnStreamClosed(int64, *core.Node)
 	// OnStreamRequest is called once a request is received on a stream.
 	// Returning an error will end processing and close the stream. OnStreamClosed will still be called.
 	OnStreamRequest(int64, *discovery.DiscoveryRequest) error
@@ -87,10 +87,13 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq
 	// a collection of stack allocated watches per request type
 	watches := newWatches()
 
+	// node may only be set on the first discovery request
+	var node = &core.Node{}
+
 	defer func() {
 		watches.close()
 		if s.callbacks != nil {
-			s.callbacks.OnStreamClosed(streamID)
+			s.callbacks.OnStreamClosed(streamID, node)
 		}
 	}()
 
@@ -130,9 +133,6 @@ func (s *server) process(str stream.Stream, reqCh <-chan *discovery.DiscoveryReq
 		}
 	}
 
-	// node may only be set on the first discovery request
-	var node = &core.Node{}
-
 	// recompute dynamic channels for this stream
 	watches.recompute(s.ctx, reqCh)
 

pkg/server/v3/server.go

@@ -26,6 +26,7 @@ import (
 	"github.com/envoyproxy/go-control-plane/pkg/server/sotw/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
 
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	clusterservice "github.com/envoyproxy/go-control-plane/envoy/service/cluster/v3"
 	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	discoverygrpc "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
@@ -68,9 +69,9 @@ type Callbacks interface {
 // CallbackFuncs is a convenience type for implementing the Callbacks interface.
 type CallbackFuncs struct {
 	StreamOpenFunc          func(context.Context, int64, string) error
-	StreamClosedFunc        func(int64)
+	StreamClosedFunc        func(int64, *core.Node)
 	DeltaStreamOpenFunc     func(context.Context, int64, string) error
-	DeltaStreamClosedFunc   func(int64)
+	DeltaStreamClosedFunc   func(int64, *core.Node)
 	StreamRequestFunc       func(int64, *discovery.DiscoveryRequest) error
 	StreamResponseFunc      func(context.Context, int64, *discovery.DiscoveryRequest, *discovery.DiscoveryResponse)
 	StreamDeltaRequestFunc  func(int64, *discovery.DeltaDiscoveryRequest) error
@@ -91,9 +92,9 @@ func (c CallbackFuncs) OnStreamOpen(ctx context.Context, streamID int64, typeURL
 }
 
 // OnStreamClosed invokes StreamClosedFunc.
-func (c CallbackFuncs) OnStreamClosed(streamID int64) {
+func (c CallbackFuncs) OnStreamClosed(streamID int64, node *core.Node) {
 	if c.StreamClosedFunc != nil {
-		c.StreamClosedFunc(streamID)
+		c.StreamClosedFunc(streamID, node)
 	}
 }
 
@@ -107,9 +108,9 @@ func (c CallbackFuncs) OnDeltaStreamOpen(ctx context.Context, streamID int64, ty
 }
 
 // OnDeltaStreamClosed invokes DeltaStreamClosedFunc.
-func (c CallbackFuncs) OnDeltaStreamClosed(streamID int64) {
+func (c CallbackFuncs) OnDeltaStreamClosed(streamID int64, node *core.Node) {
 	if c.DeltaStreamClosedFunc != nil {
-		c.DeltaStreamClosedFunc(streamID)
+		c.DeltaStreamClosedFunc(streamID, node)
 	}
 }
 

pkg/test/v3/callbacks.go

@@ -5,7 +5,9 @@ import (
 	"log"
 	"sync"
 
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/server/v3"
 )
 
 type Callbacks struct {
@@ -18,6 +20,8 @@ type Callbacks struct {
 	mu             sync.Mutex
 }
 
+var _ server.Callbacks = &Callbacks{}
+
 func (cb *Callbacks) Report() {
 	cb.mu.Lock()
 	defer cb.mu.Unlock()
@@ -29,9 +33,9 @@ func (cb *Callbacks) OnStreamOpen(_ context.Context, id int64, typ string) error
 	}
 	return nil
 }
-func (cb *Callbacks) OnStreamClosed(id int64) {
+func (cb *Callbacks) OnStreamClosed(id int64, node *core.Node) {
 	if cb.Debug {
-		log.Printf("stream %d closed\n", id)
+		log.Printf("stream %d of node %s closed\n", id, node.Id)
 	}
 }
 func (cb *Callbacks) OnDeltaStreamOpen(_ context.Context, id int64, typ string) error {
@@ -40,9 +44,9 @@ func (cb *Callbacks) OnDeltaStreamOpen(_ context.Context, id int64, typ string)
 	}
 	return nil
 }
-func (cb *Callbacks) OnDeltaStreamClosed(id int64) {
+func (cb *Callbacks) OnDeltaStreamClosed(id int64, node *core.Node) {
 	if cb.Debug {
-		log.Printf("delta stream %d closed\n", id)
+		log.Printf("delta stream %d of node %s closed\n", id, node.Id)
 	}
 }
 func (cb *Callbacks) OnStreamRequest(int64, *discovery.DiscoveryRequest) error {