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,10 @@ 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"`
+	// StaleWatchList means this list request with zero resource version, and it uses streaming list.
+	StaleWatchList *RequestList `json:"staleWatchList,omitempty" yaml:"staleWatchList,omitempty"`
+	// QuorumWatchList means this list request without kube-apiserver cache and it uses streaming list.
+	QuorumWatchList *RequestList `json:"quorumWatchList,omitempty" yaml:"quorumWatchList,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.
@@ -109,6 +113,8 @@ type RequestList struct {
 	Selector string `json:"seletor" yaml:"seletor"`
 	// FieldSelector defines how to identify a set of objects with field selector.
 	FieldSelector string `json:"fieldSelector" yaml:"fieldSelector"`
+	// Set SendInitialEvents to true if you want to use streaming list.
+	SendInitialEvents bool `json:"sendInitialEvents" yaml:"sendInitialEvents"`
 }
 
 // RequestPut defines PUT request for target resource type.
@@ -198,6 +204,10 @@ func (r WeightedRequest) Validate() error {
 		return r.StaleList.Validate(true)
 	case r.QuorumList != nil:
 		return r.QuorumList.Validate(false)
+	case r.StaleWatchList != nil:
+		return r.StaleWatchList.Validate(true)
+	case r.QuorumWatchList != nil:
+		return r.QuorumWatchList.Validate(false)
 	case r.StaleGet != nil:
 		return r.StaleGet.Validate()
 	case r.QuorumGet != nil:

api/types/load_traffic_test.go

@@ -68,6 +68,22 @@ spec:
       tailLines: 1000
       limitBytes: 1024
     shares: 10
+  - staleWatchList:
+      group: core
+      version: v1
+      resource: pods
+      namespace: default
+      seletor: app=x2
+      fieldSelector: spec.nodeName=x
+    shares: 250
+  - quorumWatchList:
+      group: core
+      version: v1
+      resource: configmaps
+      namespace: default
+      limit: 10000
+      seletor: app=x3
+    shares: 450
 `
 
 	target := LoadProfile{}
@@ -77,7 +93,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, 8)
 
 	assert.Equal(t, 100, target.Spec.Requests[0].Shares)
 	assert.NotNil(t, target.Spec.Requests[0].StaleGet)
@@ -94,7 +110,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 +123,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 +137,12 @@ 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].StaleWatchList)
+
+	assert.Equal(t, 450, target.Spec.Requests[7].Shares)
+	assert.NotNil(t, target.Spec.Requests[7].QuorumWatchList)
+
 	assert.NoError(t, target.Validate())
 }
 

request/random.go

@@ -45,9 +45,13 @@ func NewWeightedRandomRequests(spec *types.LoadProfileSpec) (*WeightedRandomRequ
 		var builder RESTRequestBuilder
 		switch {
 		case r.StaleList != nil:
-			builder = newRequestListBuilder(r.StaleList, "0", spec.MaxRetries)
+			builder = newRequestListBuilder(r.StaleList, "0", spec.MaxRetries, false)
 		case r.QuorumList != nil:
-			builder = newRequestListBuilder(r.QuorumList, "", spec.MaxRetries)
+			builder = newRequestListBuilder(r.QuorumList, "", spec.MaxRetries, false)
+		case r.StaleWatchList != nil:
+			builder = newRequestListBuilder(r.StaleList, "0", spec.MaxRetries, true)
+		case r.QuorumWatchList != nil:
+			builder = newRequestListBuilder(r.QuorumList, "", spec.MaxRetries, true)
 		case r.StaleGet != nil:
 			builder = newRequestGetBuilder(r.StaleGet, "0", spec.MaxRetries)
 		case r.QuorumGet != nil:
@@ -127,7 +131,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 +158,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,43 +168,50 @@ 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 {
-	version         schema.GroupVersion
-	resource        string
-	namespace       string
-	limit           int64
-	labelSelector   string
-	fieldSelector   string
-	resourceVersion string
-	maxRetries      int
+	version           schema.GroupVersion
+	resource          string
+	namespace         string
+	limit             int64
+	labelSelector     string
+	fieldSelector     string
+	resourceVersion   string
+	maxRetries        int
+	sendInitialEvents bool
 }
 
-func newRequestListBuilder(src *types.RequestList, resourceVersion string, maxRetries int) *requestListBuilder {
+func newRequestListBuilder(src *types.RequestList, resourceVersion string, maxRetries int, sendInitialEvents bool) *requestListBuilder {
 	return &requestListBuilder{
 		version: schema.GroupVersion{
 			Group:   src.Group,
 			Version: src.Version,
 		},
-		resource:        src.Resource,
-		namespace:       src.Namespace,
-		limit:           int64(src.Limit),
-		labelSelector:   src.Selector,
-		fieldSelector:   src.FieldSelector,
-		resourceVersion: resourceVersion,
-		maxRetries:      maxRetries,
+		resource:          src.Resource,
+		namespace:         src.Namespace,
+		limit:             int64(src.Limit),
+		labelSelector:     src.Selector,
+		fieldSelector:     src.FieldSelector,
+		resourceVersion:   resourceVersion,
+		maxRetries:        maxRetries,
+		sendInitialEvents: sendInitialEvents,
 	}
 }
 
 // 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 == "" {
@@ -213,17 +224,44 @@ 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,
+	if b.sendInitialEvents {
+		return &WatchListRequester{
+			BaseRequester: BaseRequester{
+				method: "WATCHLIST",
+				req: cli.Get().AbsPath(comps...).
+					SpecificallyVersionedParams(
+						&metav1.ListOptions{
+							LabelSelector:       b.labelSelector,
+							FieldSelector:       b.fieldSelector,
+							ResourceVersion:     b.resourceVersion,
+							Limit:               b.limit,
+							Watch:               true,
+							SendInitialEvents:   &b.sendInitialEvents,
+							AllowWatchBookmarks: true,
+						},
+						scheme.ParameterCodec,
+						schema.GroupVersion{Version: "v1"},
+					).MaxRetries(b.maxRetries),
 			},
-			scheme.ParameterCodec,
-			schema.GroupVersion{Version: "v1"},
-		).MaxRetries(b.maxRetries)
+		}
+	}
+	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 requestGetPodLogBuilder struct {
@@ -252,7 +290,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 +299,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,78 @@
+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)
+	bytes = 0
+	if err == nil {
+		defer respBody.Close()
+		bytes, err = io.Copy(io.Discard, respBody)
+		// Based on HTTP2 Spec Section 8.1 [1],
+		//
+		// A server can send a complete response prior to the client
+		// sending an entire request if the response does not depend
+		// on any portion of the request that has not been sent and
+		// received. When this is true, a server MAY request that the
+		// client abort transmission of a request without error by
+		// sending a RST_STREAM with an error code of NO_ERROR after
+		// sending a complete response (i.e., a frame with the END_STREAM
+		// flag). Clients MUST NOT discard responses as a result of receiving
+		// such a RST_STREAM, though clients can always discard responses
+		// at their discretion for other reasons.
+		//
+		// We should mark NO_ERROR as nil here.
+		//
+		// [1]: https://httpwg.org/specs/rfc7540.html#HttpSequence
+		if err != nil && isHTTP2StreamNoError(err) {
+			err = nil
+		}
+	}
+	return bytes, err
+}
+
+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
+}