fix(fds): simplifies sync logic (#134)

My understanding of the current client-server flow is as follows:

- client connects to the server
- it requests push of predefined types (ExportedSvc) as initial push
  request - SotW
- on subscriber(client) connect, server does another SotW push

Not only we are sending the same data twice, but we also need to make
sure that handlers we define are aligned with initial push requests.
If we add new handler, we need to explictly define it as part of
initial push. It also needs to be defined (again) in onSubscribe
on the FDS server side.

This PR moves the initial sync responsibility to the client, ensuring
each handler automatically requests the data it needs. This avoids
redundant pushes and reduces code duplication.

Signed-off-by: bartoszmajsak <[email protected]>

Signed-off-by: bartoszmajsak <[email protected]>

Author: bartoszmajsak

Diff for: cmd/federation-controller/main.go

@@ -24,7 +24,6 @@ import (
 	"syscall"
 	"time"
 
-	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	routev1client "github.com/openshift/client-go/route/clientset/versioned"
 	istiokube "istio.io/istio/pkg/kube"
 	istiolog "istio.io/istio/pkg/log"
@@ -251,13 +250,8 @@ func startReconciler(ctx context.Context, cfg *config.Federation, serviceLister
 }
 
 func startFederationServer(ctx context.Context, cfg *config.Federation, serviceLister v1.ServiceLister, fdsPushRequests chan xds.PushRequest) {
-	triggerFDSPushOnNewSubscription := func() {
-		fdsPushRequests <- xds.PushRequest{TypeUrl: xds.ExportedServiceTypeUrl}
-	}
-
 	federationServer := adss.NewServer(
 		fdsPushRequests,
-		triggerFDSPushOnNewSubscription,
 		fds.NewExportedServicesGenerator(*cfg, serviceLister),
 	)
 
@@ -316,9 +310,6 @@ func startFDSClient(ctx context.Context, remote config.Remote, meshConfigPushReq
 		PeerName:      remote.Name,
 		DiscoveryAddr: discoveryAddr,
 		Authority:     remote.ServiceFQDN(),
-		InitialDiscoveryRequests: []*discovery.DiscoveryRequest{{
-			TypeUrl: xds.ExportedServiceTypeUrl,
-		}},
 		Handlers: map[string]adsc.ResponseHandler{
 			xds.ExportedServiceTypeUrl: fds.NewImportedServiceHandler(importedServiceStore, meshConfigPushRequests),
 		},

Diff for: internal/pkg/xds/adsc/adsc.go

@@ -35,12 +35,11 @@ const (
 )
 
 type ADSCConfig struct {
-	PeerName                 string
-	DiscoveryAddr            string
-	Authority                string
-	InitialDiscoveryRequests []*discovery.DiscoveryRequest
-	Handlers                 map[string]ResponseHandler
-	ReconnectDelay           time.Duration
+	PeerName       string
+	DiscoveryAddr  string
+	Authority      string
+	Handlers       map[string]ResponseHandler
+	ReconnectDelay time.Duration
 }
 
 type ADSC struct {
@@ -66,17 +65,17 @@ func New(opts *ADSCConfig) (*ADSC, error) {
 }
 
 func (a *ADSC) Run(ctx context.Context) error {
-	var err error
-
 	client := discovery.NewAggregatedDiscoveryServiceClient(a.conn)
-	a.stream, err = client.StreamAggregatedResources(ctx)
-	if err != nil {
-		return err
+
+	var err error
+	if a.stream, err = client.StreamAggregatedResources(ctx); err != nil {
+		return fmt.Errorf("failed setting resource stream: %w", err)
 	}
 
-	for _, r := range a.cfg.InitialDiscoveryRequests {
-		if sendErr := a.Send(r); sendErr != nil {
-			a.log.Errorf("failed sending initial discovery request: %+v", sendErr)
+	for k, _ := range a.cfg.Handlers {
+		discoveryRequest := &discovery.DiscoveryRequest{TypeUrl: k}
+		if sendErr := a.Send(discoveryRequest); sendErr != nil {
+			a.log.Errorf("[%s] failed requesting initial discovery sync: %+v", k, sendErr)
 		}
 	}
 

Diff for: internal/pkg/xds/adss/adss_handler.go

@@ -48,7 +48,6 @@ type adsServer struct {
 	handlers         map[string]RequestHandler
 	subscribers      sync.Map
 	nextSubscriberID atomic.Uint64
-	onNewSubscriber  func()
 }
 
 // subscriber represents a client that is subscribed to XDS resources.
@@ -72,9 +71,6 @@ func (adss *adsServer) StreamAggregatedResources(downstream DiscoveryStream) err
 
 	adss.subscribers.Store(sub.id, sub)
 
-	if adss.onNewSubscriber != nil {
-		adss.onNewSubscriber()
-	}
 	go adss.recvFromStream(int64(sub.id), downstream)
 
 	<-ctx.Done()

Diff for: internal/pkg/xds/adss/grpc_server.go

@@ -31,15 +31,14 @@ type Server struct {
 	pushRequests <-chan xds.PushRequest
 }
 
-func NewServer(pushRequests <-chan xds.PushRequest, onNewSubscriber func(), handlers ...RequestHandler) *Server {
+func NewServer(pushRequests <-chan xds.PushRequest, handlers ...RequestHandler) *Server {
 	grpcServer := grpc.NewServer()
 	handlerMap := make(map[string]RequestHandler)
 	for _, g := range handlers {
 		handlerMap[g.GetTypeUrl()] = g
 	}
 	ads := &adsServer{
-		handlers:        handlerMap,
-		onNewSubscriber: onNewSubscriber,
+		handlers: handlerMap,
 	}
 
 	discovery.RegisterAggregatedDiscoveryServiceServer(grpcServer, ads)