Fix flake SDS test (#11875)

Signed-off-by: Yossi Mesika <[email protected]>

Author: ymesika

internal/sds/pkg/run/run_e2e_test.go

@@ -11,6 +11,7 @@ import (
 	envoy_service_secret_v3 "github.com/envoyproxy/go-control-plane/envoy/service/secret/v3"
 	"github.com/spf13/afero"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/kgateway-dev/kgateway/v2/internal/sds/pkg/run"
 	"github.com/kgateway-dev/kgateway/v2/internal/sds/pkg/server"
@@ -112,8 +113,8 @@ var _ = Describe("SDS Server E2E Test", Serial, func() {
 		// Check that we get a good response
 		Eventually(func() bool {
 			_, err = client.FetchSecrets(ctx, &envoy_service_discovery_v3.DiscoveryRequest{})
-			return err != nil
-		}, "5s", "1s").Should(BeTrue())
+			return err == nil
+		}, "30s", "1s").Should(BeTrue())
 
 		// Cancel the context in order to stop the gRPC server
 		cancel()
@@ -122,7 +123,7 @@ var _ = Describe("SDS Server E2E Test", Serial, func() {
 		Eventually(func() bool {
 			_, err = client.FetchSecrets(ctx, &envoy_service_discovery_v3.DiscoveryRequest{})
 			return err != nil
-		}, "5s", "1s").Should(BeTrue())
+		}, "30s", "1s").Should(BeTrue())
 
 	})
 
@@ -143,7 +144,7 @@ var _ = Describe("SDS Server E2E Test", Serial, func() {
 
 		// Connect with the server
 		var conn *grpc.ClientConn
-		conn, err = grpc.Dial(testServerAddress, grpc.WithInsecure())
+		conn, err = grpc.NewClient(testServerAddress, grpc.WithTransportCredentials(insecure.NewCredentials()))
 		Expect(err).NotTo(HaveOccurred())
 		defer conn.Close()
 		client := envoy_service_secret_v3.NewSecretDiscoveryServiceClient(conn)

internal/sds/pkg/server/server.go

@@ -79,18 +79,32 @@ func (s *Server) Run(ctx context.Context) (<-chan struct{}, error) {
 		return nil, err
 	}
 	slog.Info("sds server listening", "address", s.address)
+
+	// Create channels for synchronization
+	serveStarted := make(chan struct{})
+	serverStopped := make(chan struct{})
+
+	// Start the server in a goroutine
 	go func() {
+		// Signal that Serve is about to be called
+		close(serveStarted)
 		if err = s.grpcServer.Serve(lis); err != nil {
 			log.Fatalf("fatal error in gRPC server: address=%s error=%v", s.address, err)
 		}
 	}()
-	serverStopped := make(chan struct{})
+
+	// Wait for Serve to start before setting up shutdown handler
 	go func() {
+		// Wait for Serve to be called
+		<-serveStarted
+
+		// Now wait for context cancellation
 		<-ctx.Done()
 		slog.Info("stopping sds server", "address", s.address)
 		s.grpcServer.GracefulStop()
 		serverStopped <- struct{}{}
 	}()
+
 	return serverStopped, nil
 }