Adds WatchList to enable testing with WatchList/StreamingList

.gitignore

@@ -29,3 +29,6 @@ tmp/
 
 #.txt files which contain response stats
 *.txt
+
+# VSCode settings
+.vscode/

api/types/load_traffic.go

@@ -77,6 +77,8 @@ type WeightedRequest struct {
 	StaleList *RequestList `json:"staleList,omitempty" yaml:"staleList,omitempty"`
 	// QuorumList means this list request without kube-apiserver cache.
 	QuorumList *RequestList `json:"quorumList,omitempty" yaml:"quorumList,omitempty"`
+	// WatchList lists objects with the watch list feature, a.k.a streaming list.
+	WatchList *RequestWatchList `json:"watchList,omitempty" yaml:"watchList,omitempty"`
 	// StaleGet means this get request with zero resource version.
 	StaleGet *RequestGet `json:"staleGet,omitempty" yaml:"staleGet,omitempty"`
 	// QuorumGet means this get request without kube-apiserver cache.
@@ -111,6 +113,17 @@ type RequestList struct {
 	FieldSelector string `json:"fieldSelector" yaml:"fieldSelector"`
 }
 
+type RequestWatchList struct {
+	// KubeGroupVersionResource identifies the resource URI.
+	KubeGroupVersionResource `yaml:",inline"`
+	// Namespace is object's namespace.
+	Namespace string `json:"namespace" yaml:"namespace"`
+	// Selector defines how to identify a set of objects.
+	Selector string `json:"selector" yaml:"selector"`
+	// FieldSelector defines how to identify a set of objects with field selector.
+	FieldSelector string `json:"fieldSelector" yaml:"fieldSelector"`
+}
+
 // RequestPut defines PUT request for target resource type.
 type RequestPut struct {
 	// KubeGroupVersionResource identifies the resource URI.
@@ -198,6 +211,8 @@ func (r WeightedRequest) Validate() error {
 		return r.StaleList.Validate(true)
 	case r.QuorumList != nil:
 		return r.QuorumList.Validate(false)
+	case r.WatchList != nil:
+		return r.WatchList.Validate()
 	case r.StaleGet != nil:
 		return r.StaleGet.Validate()
 	case r.QuorumGet != nil:
@@ -227,6 +242,13 @@ func (r *RequestList) Validate(stale bool) error {
 	return nil
 }
 
+func (r *RequestWatchList) Validate() error {
+	if err := r.KubeGroupVersionResource.Validate(); err != nil {
+		return fmt.Errorf("kube metadata: %v", err)
+	}
+	return nil
+}
+
 // Validate validates RequestGet type.
 func (r *RequestGet) Validate() error {
 	if err := r.KubeGroupVersionResource.Validate(); err != nil {

api/types/load_traffic_test.go

@@ -68,6 +68,14 @@ spec:
       tailLines: 1000
       limitBytes: 1024
     shares: 10
+  - watchList:
+      group: core
+      version: v1
+      resource: pods
+      namespace: default
+      seletor: app=x2
+      fieldSelector: spec.nodeName=x
+    shares: 250
 `
 
 	target := LoadProfile{}
@@ -77,7 +85,7 @@ spec:
 	assert.Equal(t, float64(100), target.Spec.Rate)
 	assert.Equal(t, 10000, target.Spec.Total)
 	assert.Equal(t, 2, target.Spec.Conns)
-	assert.Len(t, target.Spec.Requests, 6)
+	assert.Len(t, target.Spec.Requests, 7)
 
 	assert.Equal(t, 100, target.Spec.Requests[0].Shares)
 	assert.NotNil(t, target.Spec.Requests[0].StaleGet)
@@ -94,7 +102,7 @@ spec:
 	assert.NotNil(t, target.Spec.Requests[2].StaleList)
 	assert.Equal(t, "pods", target.Spec.Requests[2].StaleList.Resource)
 	assert.Equal(t, "v1", target.Spec.Requests[2].StaleList.Version)
-	assert.Equal(t, "core", target.Spec.Requests[0].StaleGet.Group)
+	assert.Equal(t, "core", target.Spec.Requests[2].StaleList.Group)
 	assert.Equal(t, "default", target.Spec.Requests[2].StaleList.Namespace)
 	assert.Equal(t, 0, target.Spec.Requests[2].StaleList.Limit)
 	assert.Equal(t, "app=x2", target.Spec.Requests[2].StaleList.Selector)
@@ -107,7 +115,7 @@ spec:
 	assert.NotNil(t, target.Spec.Requests[4].Put)
 	assert.Equal(t, "configmaps", target.Spec.Requests[4].Put.Resource)
 	assert.Equal(t, "v1", target.Spec.Requests[4].Put.Version)
-	assert.Equal(t, "core", target.Spec.Requests[0].StaleGet.Group)
+	assert.Equal(t, "core", target.Spec.Requests[4].Put.Group)
 	assert.Equal(t, "kperf", target.Spec.Requests[4].Put.Namespace)
 	assert.Equal(t, "kperf-", target.Spec.Requests[4].Put.Name)
 	assert.Equal(t, 1000, target.Spec.Requests[4].Put.KeySpaceSize)
@@ -121,6 +129,9 @@ spec:
 	assert.Equal(t, int64(1000), *target.Spec.Requests[5].GetPodLog.TailLines)
 	assert.Equal(t, int64(1024), *target.Spec.Requests[5].GetPodLog.LimitBytes)
 
+	assert.Equal(t, 250, target.Spec.Requests[6].Shares)
+	assert.NotNil(t, target.Spec.Requests[6].WatchList)
+
 	assert.NoError(t, target.Validate())
 }
 

request/random.go

@@ -48,6 +48,8 @@ func NewWeightedRandomRequests(spec *types.LoadProfileSpec) (*WeightedRandomRequ
 			builder = newRequestListBuilder(r.StaleList, "0", spec.MaxRetries)
 		case r.QuorumList != nil:
 			builder = newRequestListBuilder(r.QuorumList, "", spec.MaxRetries)
+		case r.WatchList != nil:
+			builder = newRequestWatchListBuilder(r.WatchList, spec.MaxRetries)
 		case r.StaleGet != nil:
 			builder = newRequestGetBuilder(r.StaleGet, "0", spec.MaxRetries)
 		case r.QuorumGet != nil:
@@ -127,7 +129,7 @@ func (r *WeightedRandomRequests) Stop() {
 
 // RESTRequestBuilder is used to build rest.Request.
 type RESTRequestBuilder interface {
-	Build(cli rest.Interface) (method string, _ *rest.Request)
+	Build(cli rest.Interface) Requester
 }
 
 type requestGetBuilder struct {
@@ -154,7 +156,7 @@ func newRequestGetBuilder(src *types.RequestGet, resourceVersion string, maxRetr
 }
 
 // Build implements RequestBuilder.Build.
-func (b *requestGetBuilder) Build(cli rest.Interface) (string, *rest.Request) {
+func (b *requestGetBuilder) Build(cli rest.Interface) Requester {
 	// https://kubernetes.io/docs/reference/using-api/#api-groups
 	comps := make([]string, 0, 5)
 	if b.version.Group == "" {
@@ -164,12 +166,17 @@ func (b *requestGetBuilder) Build(cli rest.Interface) (string, *rest.Request) {
 	}
 	comps = append(comps, b.resource, b.name)
 
-	return "GET", cli.Get().AbsPath(comps...).
-		SpecificallyVersionedParams(
-			&metav1.GetOptions{ResourceVersion: b.resourceVersion},
-			scheme.ParameterCodec,
-			schema.GroupVersion{Version: "v1"},
-		).MaxRetries(b.maxRetries)
+	return &DiscardRequester{
+		BaseRequester: BaseRequester{
+			method: "GET",
+			req: cli.Get().AbsPath(comps...).
+				SpecificallyVersionedParams(
+					&metav1.GetOptions{ResourceVersion: b.resourceVersion},
+					scheme.ParameterCodec,
+					schema.GroupVersion{Version: "v1"},
+				).MaxRetries(b.maxRetries),
+		},
+	}
 }
 
 type requestListBuilder struct {
@@ -200,7 +207,62 @@ func newRequestListBuilder(src *types.RequestList, resourceVersion string, maxRe
 }
 
 // Build implements RequestBuilder.Build.
-func (b *requestListBuilder) Build(cli rest.Interface) (string, *rest.Request) {
+func (b *requestListBuilder) Build(cli rest.Interface) Requester {
+	// https://kubernetes.io/docs/reference/using-api/#api-groups
+	comps := make([]string, 0, 5)
+	if b.version.Group == "" {
+		comps = append(comps, "api", b.version.Version)
+	} else {
+		comps = append(comps, "apis", b.version.Group, b.version.Version)
+	}
+	if b.namespace != "" {
+		comps = append(comps, "namespaces", b.namespace)
+	}
+	comps = append(comps, b.resource)
+
+	return &DiscardRequester{
+		BaseRequester: BaseRequester{
+			method: "LIST",
+			req: cli.Get().AbsPath(comps...).
+				SpecificallyVersionedParams(
+					&metav1.ListOptions{
+						LabelSelector:   b.labelSelector,
+						FieldSelector:   b.fieldSelector,
+						ResourceVersion: b.resourceVersion,
+						Limit:           b.limit,
+					},
+					scheme.ParameterCodec,
+					schema.GroupVersion{Version: "v1"},
+				).MaxRetries(b.maxRetries),
+		},
+	}
+}
+
+type requestWatchListBuilder struct {
+	version       schema.GroupVersion
+	resource      string
+	namespace     string
+	labelSelector string
+	fieldSelector string
+	maxRetries    int
+}
+
+func newRequestWatchListBuilder(src *types.RequestWatchList, maxRetries int) *requestWatchListBuilder {
+	return &requestWatchListBuilder{
+		version: schema.GroupVersion{
+			Group:   src.Group,
+			Version: src.Version,
+		},
+		resource:      src.Resource,
+		namespace:     src.Namespace,
+		labelSelector: src.Selector,
+		fieldSelector: src.FieldSelector,
+		maxRetries:    maxRetries,
+	}
+}
+
+// Build implements RequestBuilder.Build.
+func (b *requestWatchListBuilder) Build(cli rest.Interface) Requester {
 	// https://kubernetes.io/docs/reference/using-api/#api-groups
 	comps := make([]string, 0, 5)
 	if b.version.Group == "" {
@@ -213,17 +275,24 @@ func (b *requestListBuilder) Build(cli rest.Interface) (string, *rest.Request) {
 	}
 	comps = append(comps, b.resource)
 
-	return "LIST", cli.Get().AbsPath(comps...).
-		SpecificallyVersionedParams(
-			&metav1.ListOptions{
-				LabelSelector:   b.labelSelector,
-				FieldSelector:   b.fieldSelector,
-				ResourceVersion: b.resourceVersion,
-				Limit:           b.limit,
-			},
-			scheme.ParameterCodec,
-			schema.GroupVersion{Version: "v1"},
-		).MaxRetries(b.maxRetries)
+	return &WatchListRequester{
+		BaseRequester: BaseRequester{
+			method: "WATCHLIST",
+			req: cli.Get().AbsPath(comps...).
+				SpecificallyVersionedParams(
+					&metav1.ListOptions{
+						LabelSelector:       b.labelSelector,
+						FieldSelector:       b.fieldSelector,
+						ResourceVersion:     "",
+						Watch:               true,
+						SendInitialEvents:   toPtr(true),
+						AllowWatchBookmarks: true,
+					},
+					scheme.ParameterCodec,
+					schema.GroupVersion{Version: "v1"},
+				).MaxRetries(b.maxRetries),
+		},
+	}
 }
 
 type requestGetPodLogBuilder struct {
@@ -252,7 +321,7 @@ func newRequestGetPodLogBuilder(src *types.RequestGetPodLog, maxRetries int) *re
 }
 
 // Build implements RequestBuilder.Build.
-func (b *requestGetPodLogBuilder) Build(cli rest.Interface) (string, *rest.Request) {
+func (b *requestGetPodLogBuilder) Build(cli rest.Interface) Requester {
 	// https://kubernetes.io/docs/reference/using-api/#api-groups
 	apiPath, version := "api", "v1"
 
@@ -261,16 +330,21 @@ func (b *requestGetPodLogBuilder) Build(cli rest.Interface) (string, *rest.Reque
 	comps = append(comps, "namespaces", b.namespace)
 	comps = append(comps, "pods", b.name, "log")
 
-	return "POD_LOG", cli.Get().AbsPath(comps...).
-		SpecificallyVersionedParams(
-			&corev1.PodLogOptions{
-				Container:  b.container,
-				TailLines:  b.tailLines,
-				LimitBytes: b.limitBytes,
-			},
-			scheme.ParameterCodec,
-			schema.GroupVersion{Version: "v1"},
-		).MaxRetries(b.maxRetries)
+	return &DiscardRequester{
+		BaseRequester: BaseRequester{
+			method: "POD_LOG",
+			req: cli.Get().AbsPath(comps...).
+				SpecificallyVersionedParams(
+					&corev1.PodLogOptions{
+						Container:  b.container,
+						TailLines:  b.tailLines,
+						LimitBytes: b.limitBytes,
+					},
+					scheme.ParameterCodec,
+					schema.GroupVersion{Version: "v1"},
+				).MaxRetries(b.maxRetries),
+		},
+	}
 }
 
 func toPtr[T any](v T) *T {

request/requester.go

@@ -0,0 +1,59 @@
+package request
+
+import (
+	"context"
+	"io"
+	"net/url"
+	"time"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/rest"
+)
+
+type Requester interface {
+	Method() string
+	URL() *url.URL
+	Timeout(time.Duration)
+	Do(context.Context) (bytes int64, err error)
+}
+
+type BaseRequester struct {
+	method string
+	req    *rest.Request
+}
+
+func (reqr *BaseRequester) Method() string {
+	return reqr.method
+}
+
+func (reqr *BaseRequester) URL() *url.URL {
+	return reqr.req.URL()
+}
+
+func (reqr *BaseRequester) Timeout(timeout time.Duration) {
+	reqr.req.Timeout(timeout)
+}
+
+type DiscardRequester struct {
+	BaseRequester
+}
+
+func (reqr *DiscardRequester) Do(ctx context.Context) (bytes int64, err error) {
+	respBody, err := reqr.req.Stream(ctx)
+	if err != nil {
+		return 0, err
+	}
+	defer respBody.Close()
+
+	return io.Copy(io.Discard, respBody)
+}
+
+type WatchListRequester struct {
+	BaseRequester
+}
+
+func (reqr *WatchListRequester) Do(ctx context.Context) (bytes int64, err error) {
+	result := &unstructured.UnstructuredList{}
+	err = reqr.req.WatchList(ctx).Into(result)
+	return 0, err
+}