[apiserver] List services with pagination (#3309)

Author: dentiny

apiserver/pkg/http/client.go

@@ -463,6 +463,10 @@ func (krc *KuberayAPIServerClient) ListRayServices(request *api.ListRayServicesR
 		return nil, nil, fmt.Errorf("failed to create http request for url '%s': %w", getURL, err)
 	}
 
+	q := httpRequest.URL.Query()
+	q.Set("pageSize", strconv.FormatInt(int64(request.PageSize), 10))
+	q.Set("pageToken", request.PageToken)
+	httpRequest.URL.RawQuery = q.Encode()
 	httpRequest.Header.Add("Accept", "application/json")
 
 	bodyBytes, status, err := krc.executeHttpRequest(httpRequest, getURL)
@@ -473,6 +477,7 @@ func (krc *KuberayAPIServerClient) ListRayServices(request *api.ListRayServicesR
 	if err := krc.unmarshaler.Unmarshal(bodyBytes, response); err != nil {
 		return nil, status, fmt.Errorf("failed to unmarshal: %+w", err)
 	}
+
 	return response, nil, nil
 }
 

apiserver/pkg/manager/resource_manager.go

@@ -276,24 +276,26 @@ func (r *ResourceManager) GetService(ctx context.Context, serviceName, namespace
 	return getServiceByName(ctx, client, serviceName)
 }
 
-func (r *ResourceManager) ListServices(ctx context.Context, namespace string) ([]*rayv1api.RayService, error) {
+func (r *ResourceManager) ListServices(ctx context.Context, namespace string, pageToken string, pageSize int32) ([]*rayv1api.RayService, string /*nextPageToken*/, error) {
 	labelSelector := metav1.LabelSelector{
 		MatchLabels: map[string]string{
 			util.KubernetesManagedByLabelKey: util.ComponentName,
 		},
 	}
 	rayServiceList, err := r.getRayServiceClient(namespace).List(ctx, metav1.ListOptions{
 		LabelSelector: labels.Set(labelSelector.MatchLabels).String(),
+		Limit:         int64(pageSize),
+		Continue:      pageToken,
 	})
 	if err != nil {
-		return nil, util.Wrap(err, fmt.Sprintf("List RayService failed in %s", namespace))
+		return nil, "" /*nextPageToken*/, util.Wrap(err, fmt.Sprintf("List RayService failed in %s with next page token %s and page limit %d", namespace, pageToken, pageSize))
 	}
 	rayServices := make([]*rayv1api.RayService, 0)
 	for _, service := range rayServiceList.Items {
 		rayServices = append(rayServices, &service)
 	}
 
-	return rayServices, nil
+	return rayServices, rayServiceList.Continue, nil
 }
 
 func (r *ResourceManager) ListAllServices(ctx context.Context) ([]*rayv1api.RayService, error) {
@@ -305,7 +307,7 @@ func (r *ResourceManager) ListAllServices(ctx context.Context) ([]*rayv1api.RayS
 	}
 
 	for _, namespace := range namespaces.Items {
-		servicesByNamespace, err := r.ListServices(ctx, namespace.Name)
+		servicesByNamespace, _, err := r.ListServices(ctx, namespace.Name, "" /*pageToken*/, 0 /*pageSize*/)
 		if err != nil {
 			return nil, util.Wrap(err, "List All Rayservices failed")
 		}

apiserver/pkg/server/serve_server.go

@@ -87,10 +87,11 @@ func (s *RayServiceServer) ListRayServices(ctx context.Context, request *api.Lis
 	if request.Namespace == "" {
 		return nil, util.NewInvalidInputError("ray service namespace is empty. Please specify a valid value.")
 	}
-	services, err := s.resourceManager.ListServices(ctx, request.Namespace)
+	services, nextPageToken, err := s.resourceManager.ListServices(ctx, request.Namespace, request.PageToken, request.PageSize)
 	if err != nil {
 		return nil, util.Wrap(err, "failed to list rayservice.")
 	}
+
 	serviceEventMap := make(map[string][]corev1.Event)
 	for _, service := range services {
 		serviceEvents, err := s.resourceManager.GetServiceEvents(ctx, *service)
@@ -101,7 +102,8 @@ func (s *RayServiceServer) ListRayServices(ctx context.Context, request *api.Lis
 		serviceEventMap[service.Name] = serviceEvents
 	}
 	return &api.ListRayServicesResponse{
-		Services: model.FromCrdToAPIServices(services, serviceEventMap),
+		Services:      model.FromCrdToAPIServices(services, serviceEventMap),
+		NextPageToken: nextPageToken,
 	}, nil
 }
 

apiserver/test/e2e/service_server_e2e_test.go

@@ -253,6 +253,107 @@ func TestGetServicesInNamespace(t *testing.T) {
 	require.Equal(t, tCtx.GetNamespaceName(), response.Services[0].Namespace)
 }
 
+func TestGetServicesInNamespaceWithPagination(t *testing.T) {
+	const serviceCount = 2
+	expectedServiceNames := make([]string, 0, serviceCount)
+
+	tCtx, err := NewEnd2EndTestingContext(t)
+	require.NoError(t, err, "No error expected when creating testing context")
+
+	tCtx.CreateComputeTemplate(t)
+	t.Cleanup(func() {
+		tCtx.DeleteComputeTemplate(t)
+	})
+
+	for ii := 0; ii < serviceCount; ii++ {
+		testServiceRequest := createTestServiceV2(t, tCtx)
+		t.Cleanup(func() {
+			tCtx.DeleteRayService(t, testServiceRequest.Service.Name)
+		})
+		expectedServiceNames = append(expectedServiceNames, testServiceRequest.Service.Name)
+	}
+
+	// Test pagination with limit 1, which is less than the total number of services.
+	t.Run("Test pagination return part of the result services", func(t *testing.T) {
+		// Used to check all services have been returned.
+		gotServices := []bool{false, false}
+
+		pageToken := ""
+		for ii := 0; ii < serviceCount; ii++ {
+			response, actualRPCStatus, err := tCtx.GetRayAPIServerClient().ListRayServices(&api.ListRayServicesRequest{
+				Namespace: tCtx.GetNamespaceName(),
+				PageToken: pageToken,
+				PageSize:  int32(1),
+			})
+
+			require.NoError(t, err, "No error expected")
+			require.Nil(t, actualRPCStatus, "No RPC status expected")
+			require.NotNil(t, response, "A response is expected")
+			require.NotEmpty(t, response.Services, "A list of service is required")
+			require.Len(t, response.Services, 1)
+
+			for _, curService := range response.Services {
+				for jj := 0; jj < serviceCount; jj++ {
+					if expectedServiceNames[jj] == curService.Name {
+						gotServices[jj] = true
+						break
+					}
+				}
+			}
+
+			// Check next page token.
+			pageToken = response.NextPageToken
+			if ii == serviceCount-1 {
+				require.Empty(t, pageToken, "Last page token should be empty")
+			} else {
+				require.NotEmpty(t, pageToken, "Non-last page token should be non empty")
+			}
+		}
+
+		// Check all services created have been returned.
+		for idx := 0; idx < serviceCount; idx++ {
+			if !gotServices[idx] {
+				t.Errorf("ListServices did not return expected services %s", expectedServiceNames[idx])
+			}
+		}
+	})
+
+	// Test pagination with limit 3, which is larger than the total number of services.
+	t.Run("Test pagination return all result services", func(t *testing.T) {
+		// Used to check all services have been returned.
+		gotServices := []bool{false, false}
+
+		pageToken := ""
+		response, actualRPCStatus, err := tCtx.GetRayAPIServerClient().ListRayServices(&api.ListRayServicesRequest{
+			Namespace: tCtx.GetNamespaceName(),
+			PageToken: pageToken,
+			PageSize:  serviceCount + 1,
+		})
+
+		require.NoError(t, err, "No error expected")
+		require.Nil(t, actualRPCStatus, "No RPC status expected")
+		require.NotNil(t, response, "A response is expected")
+		require.NotEmpty(t, response.Services, "A list of services is required")
+		require.Len(t, response.Services, serviceCount)
+		require.Empty(t, pageToken, "Page token should be empty")
+		for _, curService := range response.Services {
+			for jj := 0; jj < serviceCount; jj++ {
+				if expectedServiceNames[jj] == curService.Name {
+					gotServices[jj] = true
+					break
+				}
+			}
+		}
+
+		// Check all services created have been returned.
+		for idx := 0; idx < serviceCount; idx++ {
+			if !gotServices[idx] {
+				t.Errorf("ListServices did not return expected services %s", expectedServiceNames[idx])
+			}
+		}
+	})
+}
+
 func TestGetService(t *testing.T) {
 	tCtx, err := NewEnd2EndTestingContext(t)
 	require.NoError(t, err, "No error expected when creating testing context")
@@ -366,9 +467,7 @@ func createTestServiceV2(t *testing.T, tCtx *End2EndTestingContext) *api.CreateR
 	require.NoError(t, err, "No error expected")
 	require.Nil(t, actualRPCStatus, "No RPC status expected")
 	require.NotNil(t, actualService, "A service is expected")
-
 	checkRayServiceCreatedSuccessfully(t, tCtx, actualService.Name)
-
 	return testServiceRequest
 }
 

proto/go_client/serve.pb.go

@@ -2053,7 +2053,7 @@
         },
         "nextPageToken": {
           "type": "string",
-          "description": "The token to list the next page of RayServices.",
+          "description": "The token to list the next page of RayServices.\n\nIf there are no more clusters, this field will be empty.",
           "readOnly": true
         }
       }

proto/kuberay_api.swagger.json

@@ -113,6 +113,8 @@ message ListRayServicesResponse {
   // The total number of RayServices for the given query.
   int32 total_size = 2 [(google.api.field_behavior) = OUTPUT_ONLY];
   // The token to list the next page of RayServices.
+  //
+  // If there are no more clusters, this field will be empty.
   string next_page_token = 3 [(google.api.field_behavior) = OUTPUT_ONLY];
 }
 

proto/serve.proto

@@ -580,7 +580,7 @@
         },
         "nextPageToken": {
           "type": "string",
-          "description": "The token to list the next page of RayServices.",
+          "description": "The token to list the next page of RayServices.\n\nIf there are no more clusters, this field will be empty.",
           "readOnly": true
         }
       }