wait for the first DAG build before starting the XDS server

cmd/contour/serve.go

@@ -30,6 +30,7 @@ import (
 	networking_v1 "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
@@ -63,6 +64,10 @@ import (
 	"github.com/projectcontour/contour/pkg/config"
 )
 
+const (
+	initialDagBuildPollPeriod = 100 * time.Millisecond
+)
+
 // registerServe registers the serve subcommand and flags
 // with the Application provided.
 func registerServe(app *kingpin.Application) (*kingpin.CmdClause, *serveContext) {
@@ -658,12 +663,12 @@ func (s *Server) doServe() error {
 	}
 
 	xdsServer := &xdsServer{
-		log:               s.log,
-		registry:          s.registry,
-		config:            *contourConfiguration.XDSServer,
-		snapshotHandler:   snapshotHandler,
-		resources:         resources,
-		handlerCacheSyncs: s.handlerCacheSyncs,
+		log:             s.log,
+		registry:        s.registry,
+		config:          *contourConfiguration.XDSServer,
+		snapshotHandler: snapshotHandler,
+		resources:       resources,
+		initialDagBuilt: contourHandler.HasBuiltInitialDag,
 	}
 	if err := s.mgr.Add(xdsServer); err != nil {
 		return err
@@ -837,6 +842,7 @@ type xdsServer struct {
 	snapshotHandler   *xdscache.SnapshotHandler
 	resources         []xdscache.ResourceCache
 	handlerCacheSyncs []cache.InformerSynced
+	initialDagBuilt   func() bool
 }
 
 func (x *xdsServer) NeedLeaderElection() bool {
@@ -846,9 +852,13 @@ func (x *xdsServer) NeedLeaderElection() bool {
 func (x *xdsServer) Start(ctx context.Context) error {
 	log := x.log.WithField("context", "xds")
 
-	log.Printf("waiting for the initial list to be delivered to handlers")
-	cache.WaitForCacheSync(ctx.Done(), x.handlerCacheSyncs...)
-	log.Printf("the initial list delivered to handlers")
+	log.Printf("waiting for the initial dag to be built")
+	if err := wait.PollImmediateUntil(initialDagBuildPollPeriod, func() (bool, error) {
+		return x.initialDagBuilt(), nil
+	}, ctx.Done()); err != nil {
+		return fmt.Errorf("failed to wait for initial dag build, %w", err)
+	}
+	log.Printf("the initial dag is built")
 
 	grpcServer := xds.NewServer(x.registry, grpcOptions(log, x.config.TLS)...)
 

internal/contour/handler.go

@@ -60,6 +60,8 @@ type EventHandler struct {
 	seq int
 
 	syncTracker *synctrack.SingleFileTracker
+
+	initialDagBuilt bool
 }
 
 func NewEventHandler(config EventHandlerConfig, upstreamHasSynced cache.InformerSynced) *EventHandler {
@@ -91,6 +93,7 @@ type opDelete struct {
 
 func (e *EventHandler) OnAdd(obj any, isInInitialList bool) {
 	if isInInitialList {
+		time.Sleep(time.Second * 3)
 		e.syncTracker.Start()
 	}
 	e.update <- opAdd{obj: obj, isInInitialList: isInInitialList}
@@ -109,6 +112,10 @@ func (e *EventHandler) NeedLeaderElection() bool {
 	return false
 }
 
+func (e *EventHandler) HasBuiltInitialDag() bool {
+	return e.initialDagBuilt
+}
+
 // Implements leadership.NeedLeaderElectionNotification
 func (e *EventHandler) OnElectedLeader() {
 	// Trigger an update when we are elected leader to ensure resource
@@ -196,6 +203,7 @@ func (e *EventHandler) Start(ctx context.Context) error {
 			e.rebuildDAG()
 			e.incSequence()
 			lastDAGRebuild = time.Now()
+			e.initialDagBuilt = true
 		case <-ctx.Done():
 			// shutdown
 			return nil