[apiserver]: merge http utils of apiserver v1/v2 (ray-project#3924)

* feat: merge http utils of apiserver v1/v2

Signed-off-by: Cheng-Yeh Chung <[email protected]>

* fix naming error in package alias

Co-authored-by: Rueian <[email protected]>
Signed-off-by: Cheng-Yeh Chung <[email protected]>

* fix naming error

Signed-off-by: Cheng-Yeh Chung <[email protected]>

* refactor function naming

Signed-off-by: Cheng-Yeh Chung <[email protected]>

* remove TODO comments solved by this PR

Signed-off-by: Cheng-Yeh Chung <[email protected]>

* merge retry config to RetryRoundTripper. fix v1 merge error

Signed-off-by: Cheng-Yeh Chung <[email protected]>

---------

Signed-off-by: Cheng-Yeh Chung <[email protected]>
Co-authored-by: Rueian <[email protected]>

Author: 2 people

apiserver/pkg/http/client.go

@@ -6,14 +6,14 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math"
 	"net/http"
 	"strconv"
 	"time"
 
 	rpcStatus "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/protobuf/encoding/protojson"
 
+	apiserverutil "github.com/ray-project/kuberay/apiserversdk/util"
 	api "github.com/ray-project/kuberay/proto/go_client"
 )
 
@@ -653,15 +653,6 @@ func (krc *KuberayAPIServerClient) doDelete(deleteURL string) (*rpcStatus.Status
 	return status, err
 }
 
-var retryableHTTPStatusCodes = map[int]struct{}{
-	http.StatusRequestTimeout:      {}, // 408
-	http.StatusTooManyRequests:     {}, // 429
-	http.StatusInternalServerError: {}, // 500
-	http.StatusBadGateway:          {}, // 502
-	http.StatusServiceUnavailable:  {}, // 503
-	http.StatusGatewayTimeout:      {}, // 504
-}
-
 func (krc *KuberayAPIServerClient) executeRequest(httpRequest *http.Request, URL string) ([]byte, *rpcStatus.Status, error) {
 	// Set the overall timeout
 	ctx, cancel := context.WithTimeout(context.Background(), krc.retryCfg.OverallTimeout)
@@ -713,7 +704,7 @@ func (krc *KuberayAPIServerClient) executeRequest(httpRequest *http.Request, URL
 			break
 		}
 
-		if statusCode == http.StatusOK {
+		if apiserverutil.IsSuccessfulStatusCode(statusCode) {
 			return bodyBytes, nil, nil
 		}
 
@@ -730,16 +721,15 @@ func (krc *KuberayAPIServerClient) executeRequest(httpRequest *http.Request, URL
 			HTTPStatusCode: statusCode,
 		}
 
-		// Retry only for HTTP status in the list
-		if _, retryable := retryableHTTPStatusCodes[statusCode]; !retryable {
+		if !apiserverutil.IsRetryableHTTPStatusCodes(statusCode) {
 			break
 		}
 
 		// Backoff before retry
-		sleep := krc.retryCfg.InitBackoff * time.Duration(math.Pow(krc.retryCfg.BackoffFactor, float64(attempt)))
-		if sleep > krc.retryCfg.MaxBackoff {
-			sleep = krc.retryCfg.MaxBackoff
-		}
+		sleep := apiserverutil.GetRetryBackoff(attempt,
+			krc.retryCfg.InitBackoff,
+			krc.retryCfg.BackoffFactor,
+			krc.retryCfg.MaxBackoff)
 
 		select {
 		case <-time.After(sleep):

apiserver/pkg/http/client_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 	rpcStatus "google.golang.org/genproto/googleapis/rpc/status"
 
-	"github.com/ray-project/kuberay/apiserver/pkg/util"
+	"github.com/ray-project/kuberay/apiserversdk/util"
 	api "github.com/ray-project/kuberay/proto/go_client"
 )
 

apiserver/pkg/util/config.go

@@ -25,17 +25,6 @@ const (
 	RayClusterImageAnnotationKey           = "ray.io/compute-image"
 
 	RayClusterDefaultImageRepository = "rayproject/ray"
-
-	// Max retry times for HTTP Client
-	HTTPClientDefaultMaxRetry = 3
-
-	// Retry backoff settings
-	HTTPClientDefaultBackoffBase = float64(2)
-	HTTPClientDefaultInitBackoff = 500 * time.Millisecond
-	HTTPClientDefaultMaxBackoff  = 10 * time.Second
-
-	// Overall timeout for retries
-	HTTPClientDefaultOverallTimeout = 30 * time.Second
 )
 
 const (

apiserver/test/e2e/types.go

@@ -23,7 +23,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 
 	kuberayHTTP "github.com/ray-project/kuberay/apiserver/pkg/http"
-	"github.com/ray-project/kuberay/apiserver/pkg/util"
+	"github.com/ray-project/kuberay/apiserversdk/util"
 	api "github.com/ray-project/kuberay/proto/go_client"
 	rayv1api "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/pkg/client/clientset/versioned/typed/ray/v1"

apiserversdk/proxy.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"io"
-	"math"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -16,7 +15,8 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
+	apiserverutil "github.com/ray-project/kuberay/apiserversdk/util"
+	rayutil "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 )
 
 type MuxConfig struct {
@@ -78,7 +78,7 @@ func requireKubeRayService(handler http.Handler, k8sClient *kubernetes.Clientset
 		}
 		services, err := k8sClient.CoreV1().Services(namespace).List(r.Context(), metav1.ListOptions{
 			FieldSelector: "metadata.name=" + serviceName,
-			LabelSelector: "app.kubernetes.io/name=" + utils.ApplicationName,
+			LabelSelector: "app.kubernetes.io/name=" + rayutil.ApplicationName,
 		})
 		if err != nil {
 			http.Error(w, "failed to list kuberay services", http.StatusInternalServerError)
@@ -99,10 +99,21 @@ type retryRoundTripper struct {
 
 	// Num of retries after the initial attempt
 	maxRetries int
+
+	// Retry backoff settings
+	initBackoff time.Duration
+	backoffBase float64
+	maxBackoff  time.Duration
 }
 
 func newRetryRoundTripper(base http.RoundTripper) http.RoundTripper {
-	return &retryRoundTripper{base: base, maxRetries: HTTPClientDefaultMaxRetry}
+	return &retryRoundTripper{
+		base:        base,
+		maxRetries:  apiserverutil.HTTPClientDefaultMaxRetry,
+		initBackoff: apiserverutil.HTTPClientDefaultInitBackoff,
+		backoffBase: apiserverutil.HTTPClientDefaultBackoffBase,
+		maxBackoff:  apiserverutil.HTTPClientDefaultMaxBackoff,
+	}
 }
 
 func (rrt *retryRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
@@ -136,11 +147,11 @@ func (rrt *retryRoundTripper) RoundTrip(req *http.Request) (*http.Response, erro
 			return resp, fmt.Errorf("request to %s %s failed with error: %w", req.Method, req.URL.String(), err)
 		}
 
-		if isSuccessfulStatusCode(resp.StatusCode) {
+		if apiserverutil.IsSuccessfulStatusCode(resp.StatusCode) {
 			return resp, nil
 		}
 
-		if !isRetryableHTTPStatusCodes(resp.StatusCode) {
+		if !apiserverutil.IsRetryableHTTPStatusCodes(resp.StatusCode) {
 			return resp, nil
 		}
 
@@ -158,11 +169,7 @@ func (rrt *retryRoundTripper) RoundTrip(req *http.Request) (*http.Response, erro
 			}
 		}
 
-		// TODO: move to HTTP util function in independent util file
-		sleepDuration := HTTPClientDefaultInitBackoff * time.Duration(math.Pow(HTTPClientDefaultBackoffBase, float64(attempt)))
-		if sleepDuration > HTTPClientDefaultMaxBackoff {
-			sleepDuration = HTTPClientDefaultMaxBackoff
-		}
+		sleepDuration := apiserverutil.GetRetryBackoff(attempt, rrt.initBackoff, rrt.backoffBase, rrt.maxBackoff)
 
 		// TODO: merge common utils for apiserver v1 and v2
 		if deadline, ok := ctx.Deadline(); ok {
@@ -180,23 +187,3 @@ func (rrt *retryRoundTripper) RoundTrip(req *http.Request) (*http.Response, erro
 	}
 	return resp, err
 }
-
-// TODO: move HTTP util function into independent util file / folder
-func isSuccessfulStatusCode(statusCode int) bool {
-	return 200 <= statusCode && statusCode < 300
-}
-
-// TODO: merge common utils for apiserver v1 and v2
-func isRetryableHTTPStatusCodes(statusCode int) bool {
-	switch statusCode {
-	case http.StatusRequestTimeout, // 408
-		http.StatusTooManyRequests,     // 429
-		http.StatusInternalServerError, // 500
-		http.StatusBadGateway,          // 502
-		http.StatusServiceUnavailable,  // 503
-		http.StatusGatewayTimeout:      // 504
-		return true
-	default:
-		return false
-	}
-}

apiserversdk/proxy_test.go

@@ -23,8 +23,9 @@ import (
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
+	apiserverutil "github.com/ray-project/kuberay/apiserversdk/util"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
-	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
+	rayutil "github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 	rayclient "github.com/ray-project/kuberay/ray-operator/pkg/client/clientset/versioned/typed/ray/v1"
 )
 
@@ -319,7 +320,7 @@ var _ = Describe("kuberay service", Ordered, func() {
 				ObjectMeta: metav1.ObjectMeta{
 					Name: svcName,
 					Labels: map[string]string{
-						utils.KubernetesApplicationNameLabelKey: utils.ApplicationName,
+						rayutil.KubernetesApplicationNameLabelKey: rayutil.ApplicationName,
 					},
 				},
 				Spec: corev1.ServiceSpec{
@@ -454,7 +455,7 @@ var _ = Describe("retryRoundTripper", func() {
 		resp, err := retrier.RoundTrip(req)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(resp.StatusCode).To(Equal(http.StatusInternalServerError))
-		Expect(attempts).To(Equal(int32(HTTPClientDefaultMaxRetry + 1)))
+		Expect(attempts).To(Equal(int32(apiserverutil.HTTPClientDefaultMaxRetry + 1)))
 	})
 
 	It("Retries on request with body", func() {

apiserversdk/config.go renamed to apiserversdk/util/config.go

@@ -1,8 +1,7 @@
-package apiserversdk
+package util
 
 import "time"
 
-// TODO: Make apiserver configs compatible with V1
 const (
 	// Max retry times for HTTP Client
 	HTTPClientDefaultMaxRetry = 3

apiserversdk/util/http.go

@@ -0,0 +1,33 @@
+package util
+
+import (
+	"math"
+	"net/http"
+	"time"
+)
+
+func GetRetryBackoff(attempt int, initBackoff time.Duration, backoffBase float64, maxBackoff time.Duration) time.Duration {
+	sleepDuration := initBackoff * time.Duration(math.Pow(backoffBase, float64(attempt)))
+	if sleepDuration > maxBackoff {
+		sleepDuration = maxBackoff
+	}
+	return sleepDuration
+}
+
+func IsSuccessfulStatusCode(statusCode int) bool {
+	return 200 <= statusCode && statusCode < 300
+}
+
+func IsRetryableHTTPStatusCodes(statusCode int) bool {
+	switch statusCode {
+	case http.StatusRequestTimeout, // 408
+		http.StatusTooManyRequests,     // 429
+		http.StatusInternalServerError, // 500
+		http.StatusBadGateway,          // 502
+		http.StatusServiceUnavailable,  // 503
+		http.StatusGatewayTimeout:      // 504
+		return true
+	default:
+		return false
+	}
+}