add test harness sotw ordered options

Signed-off-by: Alec Holmes <[email protected]>

pkg/server/sotw/v3/xds.go

@@ -103,7 +103,7 @@ func (s *server) process(str stream.Stream, reqCh chan *discovery.DiscoveryReque
 					}()
 
 					// Trigger a different code path specifically for ADS.
-					// We want resource orderingso things don't get sent before they should.
+					// We want resource ordering so things don't get sent before they should.
 					// This is a blocking call and will exit the process function
 					// on successful completion.
 					return s.processADS(&sw, reqCh, defaultTypeURL)

pkg/test/main/main.go

@@ -29,6 +29,8 @@ import (
 	"time"
 
 	"github.com/envoyproxy/go-control-plane/pkg/cache/v3"
+	conf "github.com/envoyproxy/go-control-plane/pkg/server/config"
+	"github.com/envoyproxy/go-control-plane/pkg/server/sotw/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/server/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/test"
 	"github.com/envoyproxy/go-control-plane/pkg/test/resource/v3"
@@ -195,10 +197,17 @@ func main() {
 			},
 		}
 	}
-	srv := server.NewServer(context.Background(), configCache, cb)
+
+	opts := []conf.XDSOption{}
+	if mode == resource.Ads {
+		log.Println("enabling ordered ADS mode...")
+		// Enable resource ordering if we enter ADS mode.
+		opts = append(opts, sotw.WithOrderedADS())
+	}
+	srv := server.NewServer(context.Background(), configCache, cb, opts...)
 	als := &testv3.AccessLogService{}
 
-	if mode != "delta" {
+	if mode != resource.Delta {
 		vhdsHTTPListeners = 0
 	}
 

pkg/test/v3/callbacks.go

@@ -23,40 +23,55 @@ func (cb *Callbacks) Report() {
 	defer cb.mu.Unlock()
 	log.Printf("server callbacks fetches=%d requests=%d\n", cb.Fetches, cb.Requests)
 }
+
 func (cb *Callbacks) OnStreamOpen(_ context.Context, id int64, typ string) error {
 	if cb.Debug {
 		log.Printf("stream %d open for %s\n", id, typ)
 	}
 	return nil
 }
+
 func (cb *Callbacks) OnStreamClosed(id int64) {
 	if cb.Debug {
 		log.Printf("stream %d closed\n", id)
 	}
 }
+
 func (cb *Callbacks) OnDeltaStreamOpen(_ context.Context, id int64, typ string) error {
 	if cb.Debug {
 		log.Printf("delta stream %d open for %s\n", id, typ)
 	}
 	return nil
 }
+
 func (cb *Callbacks) OnDeltaStreamClosed(id int64) {
 	if cb.Debug {
 		log.Printf("delta stream %d closed\n", id)
 	}
 }
-func (cb *Callbacks) OnStreamRequest(int64, *discovery.DiscoveryRequest) error {
+
+func (cb *Callbacks) OnStreamRequest(id int64, req *discovery.DiscoveryRequest) error {
 	cb.mu.Lock()
 	defer cb.mu.Unlock()
 	cb.Requests++
 	if cb.Signal != nil {
 		close(cb.Signal)
 		cb.Signal = nil
 	}
+
+	if cb.Debug {
+		log.Printf("received request on stream %d for %s:%v", id, req.GetTypeUrl(), req.GetResourceNames())
+	}
+
 	return nil
 }
-func (cb *Callbacks) OnStreamResponse(context.Context, int64, *discovery.DiscoveryRequest, *discovery.DiscoveryResponse) {
+
+func (cb *Callbacks) OnStreamResponse(ctx context.Context, id int64, req *discovery.DiscoveryRequest, res *discovery.DiscoveryResponse) {
+	if cb.Debug {
+		log.Printf("responding on stream %d with %s:%s", id, res.GetVersionInfo(), res.GetTypeUrl())
+	}
 }
+
 func (cb *Callbacks) OnStreamDeltaResponse(id int64, req *discovery.DeltaDiscoveryRequest, res *discovery.DeltaDiscoveryResponse) {
 	cb.mu.Lock()
 	defer cb.mu.Unlock()

sample/bootstrap-ads.yaml

@@ -24,7 +24,7 @@ node:
   id: test-id
 static_resources:
   clusters:
-  - connect_timeout: 1s
+  - connect_timeout: 30s
     load_assignment:
       cluster_name: xds_cluster
       endpoints:
@@ -36,7 +36,7 @@ static_resources:
                 port_value: 18000
     http2_protocol_options: {}
     name: xds_cluster
-  - connect_timeout: 1s
+  - connect_timeout: 30s
     load_assignment:
       cluster_name: als_cluster
       endpoints: