Introduce pagination and filtering to improve ClusterProfiler memory …

…usage

Signed-off-by: Tomasz Knopik <tomasz.knopik@gmail.com>

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

@@ -30,17 +30,20 @@ import (
 	"sync"
 	"time"
 
-	restful "github.com/emicklei/go-restful"
-
-	v1 "kubevirt.io/client-go/api/v1"
-	"kubevirt.io/client-go/log"
-
+	"github.com/emicklei/go-restful"
 	k8sv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
+	"k8s.io/apimachinery/pkg/labels"
+	v1 "kubevirt.io/client-go/api/v1"
+	"kubevirt.io/client-go/log"
 	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 +66,52 @@ func (app *SubresourceAPIApp) getAllComponentPods() ([]*k8sv1.Pod, error) {
 	return pods, nil
 }
 
+func (app *SubresourceAPIApp) unmarshalClusterProfilerRequest(request *restful.Request) (*v1.ClusterProfilerRequest, error) {
+	cpRequest := &v1.ClusterProfilerRequest{}
+	if request.Request.Body == nil {
+		return nil, fmt.Errorf("empty request body")
+	}
+	return cpRequest, json.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, podList.Continue, nil
+}
+
 func podIsReadyComponent(pod *k8sv1.Pod) bool {
 	componentPrefixes := []string{"virt-controller", "virt-handler", "virt-api", "virt-operator"}
 
@@ -183,41 +232,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 +292,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

@@ -20,9 +20,12 @@
 package rest
 
 import (
+	"bytes"
 	"crypto/tls"
+	"encoding/json"
 	"flag"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
 	"strconv"
@@ -140,7 +143,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("start/stop 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),
@@ -160,6 +163,30 @@ var _ = Describe("Cluster Profiler Subresources", func() {
 		},
 			table.Entry("start function", app.StartClusterProfilerHandler, "start"),
 			table.Entry("stop function", app.StopClusterProfilerHandler, "stop"),
+		)
+
+		table.DescribeTable("dump 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),
+			}
+
+			b, err := json.Marshal(&v1.ClusterProfilerRequest{})
+			Expect(err).To(BeNil())
+			request.Request.Body = ioutil.NopCloser(bytes.NewBuffer(b))
+
+			backend.AppendHandlers(
+				ghttp.CombineHandlers(
+					ghttp.VerifyRequest("GET", fmt.Sprintf("/%s-profiler", cmd)),
+					ghttp.RespondWithJSONEncoded(http.StatusOK, results),
+				),
+			)
+
+			enableFeatureGate(virtconfig.ClusterProfiler)
+			expectPodList()
+			fn(request, response)
+			Expect(recorder.Code).To(Equal(http.StatusOK))
+		},
 			table.Entry("dump function", app.DumpClusterProfilerHandler, "dump"),
 		)
 	})

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"`
 }