Introduce pagination and filtering to improve ClusterProfiler memory …

…usage

api/api-rule-violations-known.list

@@ -221,6 +221,7 @@ API rule violation: names_match,kubevirt.io/client-go/api/v1,CDRomTarget,ReadOnl
 API rule violation: names_match,kubevirt.io/client-go/api/v1,CPU,DedicatedCPUPlacement
 API rule violation: names_match,kubevirt.io/client-go/api/v1,CloudInitConfigDriveSource,UserDataSecretRef
 API rule violation: names_match,kubevirt.io/client-go/api/v1,CloudInitNoCloudSource,UserDataSecretRef
+API rule violation: names_match,kubevirt.io/client-go/api/v1,ClusterProfilerRequest,LabelSelector
 API rule violation: names_match,kubevirt.io/client-go/api/v1,DeveloperConfiguration,LessPVCSpaceToleration
 API rule violation: names_match,kubevirt.io/client-go/api/v1,Devices,GPUs
 API rule violation: names_match,kubevirt.io/client-go/api/v1,Devices,NetworkInterfaceMultiQueue

pkg/virt-api/rest/BUILD.bazel

@@ -31,6 +31,7 @@ go_library(
         "//vendor/github.com/emicklei/go-restful:go_default_library",
         "//vendor/github.com/golang/mock/gomock:go_default_library",
         "//vendor/github.com/gorilla/websocket:go_default_library",
+        "//vendor/github.com/json-iterator/go:go_default_library",
         "//vendor/k8s.io/api/authorization/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",

pkg/virt-api/rest/profiler.go

@@ -31,6 +31,8 @@ import (
 	"time"
 
 	restful "github.com/emicklei/go-restful"
+	jsoniter "github.com/json-iterator/go"
+	"k8s.io/apimachinery/pkg/labels"
 
 	v1 "kubevirt.io/client-go/api/v1"
 	"kubevirt.io/client-go/log"
@@ -41,6 +43,11 @@ import (
 	clientutil "kubevirt.io/client-go/util"
 )
 
+const (
+	maxClusterProfilerResultsPageSize     = 20
+	defaultClusterProfilerResultsPageSize = 10
+)
+
 func (app *SubresourceAPIApp) getAllComponentPods() ([]*k8sv1.Pod, error) {
 	namespace, err := clientutil.GetNamespace()
 	if err != nil {
@@ -63,6 +70,49 @@ func (app *SubresourceAPIApp) getAllComponentPods() ([]*k8sv1.Pod, error) {
 	return pods, nil
 }
 
+func (app *SubresourceAPIApp) unmarshalClusterProfilerRequest(request *restful.Request) (*v1.ClusterProfilerRequest, error) {
+	cpRequest := &v1.ClusterProfilerRequest{}
+	return cpRequest, jsoniter.NewDecoder(request.Request.Body).Decode(cpRequest)
+}
+
+func (app *SubresourceAPIApp) getPodsNextPage(cpRequest *v1.ClusterProfilerRequest) (pods []*k8sv1.Pod, cont string, err error) {
+	var (
+		listOptions = metav1.ListOptions{}
+		namespace   string
+		podList     *k8sv1.PodList
+	)
+
+	if selector, err := labels.Parse(cpRequest.LabelSelector); err != nil {
+		return nil, "", err
+	} else {
+		listOptions.LabelSelector = selector.String()
+	}
+
+	listOptions.Continue = cpRequest.Continue
+	listOptions.Limit = cpRequest.PageSize
+	if listOptions.Limit <= 0 {
+		listOptions.Limit = defaultClusterProfilerResultsPageSize
+	} else if listOptions.Limit > maxClusterProfilerResultsPageSize {
+		listOptions.Limit = maxClusterProfilerResultsPageSize
+	}
+
+	if namespace, err = clientutil.GetNamespace(); err != nil {
+		return nil, "", err
+	}
+
+	if podList, err = app.virtCli.CoreV1().Pods(namespace).List(context.Background(), listOptions); err != nil {
+		return nil, "", err
+	}
+
+	for _, pod := range podList.Items {
+		if podIsReadyComponent(&pod) {
+			pods = append(pods, pod.DeepCopy())
+		}
+	}
+
+	return pods, listOptions.Continue, nil
+}
+
 func podIsReadyComponent(pod *k8sv1.Pod) bool {
 	componentPrefixes := []string{"virt-controller", "virt-handler", "virt-api", "virt-operator"}
 
@@ -183,41 +233,48 @@ func (app *SubresourceAPIApp) DumpClusterProfilerHandler(request *restful.Reques
 		response.WriteErrorString(http.StatusForbidden, "Unable to dump profiler results. \"ClusterProfiler\" feature gate must be enabled")
 		return
 	}
-	pods, err := app.getAllComponentPods()
+
+	cpRequest, err := app.unmarshalClusterProfilerRequest(request)
 	if err != nil {
-		response.WriteErrorString(http.StatusInternalServerError, fmt.Sprintf("Internal error while looking up component pods for profiling: %v", err))
+		response.WriteErrorString(http.StatusBadRequest, fmt.Sprintf("failed to parse cluster profiler request: %v", err))
 		return
 	}
 
-	if len(pods) == 0 {
-		response.WriteErrorString(http.StatusInternalServerError, "Internal error, no component pods found")
+	pods, cont, err := app.getPodsNextPage(cpRequest)
+	if err != nil {
+		response.WriteErrorString(http.StatusBadRequest, fmt.Sprintf("Internal error while looking up component pods for profiling: %v", err))
 		return
 	}
 
-	tr := &http.Transport{
-		TLSClientConfig: &tls.Config{
-			InsecureSkipVerify: true,
-		},
+	if len(pods) == 0 {
+		response.WriteHeaderAndJson(http.StatusNoContent, v1.ClusterProfilerResults{}, restful.MIME_JSON)
+		return
 	}
 
-	client := http.Client{
-		Timeout:   time.Duration(5 * time.Second),
-		Transport: tr,
-	}
+	const command = "dump"
+	var (
+		tr = &http.Transport{
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: true,
+			},
+		}
+		client = http.Client{
+			Timeout:   5 * time.Second,
+			Transport: tr,
+		}
+		results = v1.ClusterProfilerResults{
+			ComponentResults: make(map[string]v1.ProfilerResult),
+			Continue:         cont,
+		}
 
-	command := "dump"
+		resultsLock = sync.Mutex{}
+		wg          = sync.WaitGroup{}
+		errorChan   = make(chan error, len(pods))
+	)
 
-	wg := sync.WaitGroup{}
 	wg.Add(len(pods))
-
-	errorChan := make(chan error, len(pods))
 	defer close(errorChan)
 
-	results := v1.ClusterProfilerResults{
-		ComponentResults: make(map[string]v1.ProfilerResult),
-	}
-	resultsLock := sync.Mutex{}
-
 	go func() {
 		for _, pod := range pods {
 			ip := pod.Status.PodIP
@@ -236,8 +293,7 @@ func (app *SubresourceAPIApp) DumpClusterProfilerHandler(request *restful.Reques
 				defer resp.Body.Close()
 
 				if resp.StatusCode != http.StatusOK {
-
-					errorChan <- fmt.Errorf("Encountered [%d] status code while contacting url [%s] for pod [%s]", resp.StatusCode, url, name)
+					errorChan <- fmt.Errorf("encountered [%d] status code while contacting url [%s] for pod [%s]", resp.StatusCode, url, name)
 					return
 
 				}

pkg/virt-api/rest/profiler_test.go

@@ -140,7 +140,7 @@ var _ = Describe("Cluster Profiler Subresources", func() {
 			table.Entry("stop function", app.StopClusterProfilerHandler),
 			table.Entry("dump function", app.DumpClusterProfilerHandler),
 		)
-		table.DescribeTable("should return successr when feature gate is enabled", func(fn func(*restful.Request, *restful.Response), cmd string) {
+		table.DescribeTable("should return success when feature gate is enabled", func(fn func(*restful.Request, *restful.Response), cmd string) {
 
 			results := v1.ClusterProfilerResults{
 				ComponentResults: make(map[string]v1.ProfilerResult),

staging/src/kubevirt.io/client-go/api/v1/types.go

@@ -2121,4 +2121,12 @@ type ProfilerResult struct {
 // +k8s:openapi-gen=true
 type ClusterProfilerResults struct {
 	ComponentResults map[string]ProfilerResult `json:"componentResults"`
+	Continue         string                    `json:"continue,omitempty"`
+}
+
+// +k8s:openapi-gen=true
+type ClusterProfilerRequest struct {
+	LabelSelector string `json:"labelSelectors"`
+	Continue      string `json:"continue,omitempty"`
+	PageSize      int64  `json:"pageSize"`
 }

staging/src/kubevirt.io/client-go/kubecli/profiler.go

@@ -89,7 +89,10 @@ func (v *ClusterProfiler) Stop() error {
 	return v.restClient.Get().RequestURI(uri).Do(context.Background()).Error()
 }
 
-func (v *ClusterProfiler) Dump() (*v1.ClusterProfilerResults, error) {
+// Dump returns at most cpRequest.PageSize profiler results. To fetch results from all kubevirt pods
+// Dump should be called with Continue fields set to Continue field value from the response to a previous request.
+// This should be repeated until Continue or ComponentsResult field in ClusterProfilerResponse is empty.
+func (v *ClusterProfiler) Dump(cpRequest *v1.ClusterProfilerRequest) (*v1.ClusterProfilerResults, error) {
 	preferredVersion, err := v.preferredVersion()
 	if err != nil {
 		return nil, err
@@ -100,7 +103,7 @@ func (v *ClusterProfiler) Dump() (*v1.ClusterProfilerResults, error) {
 
 	var profileResults v1.ClusterProfilerResults
 
-	result := v.restClient.Get().RequestURI(uri).Do(context.Background())
+	result := v.restClient.Get().RequestURI(uri).Body(cpRequest).Do(context.Background())
 	if data, err := result.Raw(); err != nil {
 		connErr, isConnectionErr := err.(*url.Error)
 

tests/infra_test.go

@@ -1561,7 +1561,7 @@ var _ = Describe("[Serial][sig-compute]Infrastructure", func() {
 				err = virtClient.ClusterProfiler().Stop()
 				Expect(err).ToNot(BeNil())
 
-				_, err = virtClient.ClusterProfiler().Dump()
+				_, err = virtClient.ClusterProfiler().Dump(&v1.ClusterProfilerRequest{})
 				Expect(err).ToNot(BeNil())
 			})
 			It("is enabled it should allow subresource access", func() {
@@ -1573,7 +1573,7 @@ var _ = Describe("[Serial][sig-compute]Infrastructure", func() {
 				err = virtClient.ClusterProfiler().Stop()
 				Expect(err).To(BeNil())
 
-				_, err = virtClient.ClusterProfiler().Dump()
+				_, err = virtClient.ClusterProfiler().Dump(&v1.ClusterProfilerRequest{})
 				Expect(err).To(BeNil())
 			})
 		})

tools/cluster-profiler/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "//staging/src/kubevirt.io/client-go/api/v1:go_default_library",
         "//staging/src/kubevirt.io/client-go/kubecli:go_default_library",
         "//vendor/github.com/spf13/pflag:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
     ],
 )