Update ray support functions to handle ray job api operation using tls check verification for insecure cluster

support/ray_cluster_client.go

@@ -18,6 +18,7 @@ package support
 
 import (
 	"bytes"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -45,20 +46,41 @@ type RayJobLogsResponse struct {
 	Logs string `json:"logs"`
 }
 
+type RayClusterClientConfig struct {
+	Address            string
+	Client             *http.Client
+	InsecureSkipVerify bool
+}
+
 var _ RayClusterClient = (*rayClusterClient)(nil)
 
 type rayClusterClient struct {
-	endpoint url.URL
+	endpoint    url.URL
+	httpClient  *http.Client
+	bearerToken string
 }
 
 type RayClusterClient interface {
 	CreateJob(job *RayJobSetup) (*RayJobResponse, error)
 	GetJobDetails(jobID string) (*RayJobDetailsResponse, error)
 	GetJobLogs(jobID string) (string, error)
+	GetJobs() (*[]RayJobDetailsResponse, error)
 }
 
-func NewRayClusterClient(dashboardEndpoint url.URL) RayClusterClient {
-	return &rayClusterClient{endpoint: dashboardEndpoint}
+func NewRayClusterClient(config RayClusterClientConfig, bearerToken string) (RayClusterClient, error) {
+	tr := &http.Transport{
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: config.InsecureSkipVerify},
+		Proxy:           http.ProxyFromEnvironment,
+	}
+	config.Client = &http.Client{Transport: tr}
+	endpoint, err := url.Parse(config.Address)
+	if err != nil {
+		return nil, fmt.Errorf("invalid dashboard endpoint address")
+	}
+	rayClusterApiClient := &rayClusterClient{
+		endpoint: *endpoint, httpClient: config.Client, bearerToken: bearerToken,
+	}
+	return rayClusterApiClient, nil
 }
 
 func (client *rayClusterClient) CreateJob(job *RayJobSetup) (response *RayJobResponse, err error) {
@@ -68,7 +90,8 @@ func (client *rayClusterClient) CreateJob(job *RayJobSetup) (response *RayJobRes
 	}
 
 	createJobURL := client.endpoint.String() + "/api/jobs/"
-	resp, err := http.Post(createJobURL, "application/json", bytes.NewReader(marshalled))
+
+	resp, err := client.httpClient.Post(createJobURL, "application/json", bytes.NewReader(marshalled))
 	if err != nil {
 		return
 	}
@@ -86,36 +109,88 @@ func (client *rayClusterClient) CreateJob(job *RayJobSetup) (response *RayJobRes
 	return
 }
 
+func (client *rayClusterClient) GetJobs() (response *[]RayJobDetailsResponse, err error) {
+	getAllJobsDetailsURL := client.endpoint.String() + "/api/jobs/"
+
+	req, err := http.NewRequest(http.MethodGet, getAllJobsDetailsURL, nil)
+	if err != nil {
+		return nil, err
+	}
+	if client.bearerToken != "" {
+		req.Header.Set("Authorization", "Bearer "+client.bearerToken)
+	}
+	resp, err := client.httpClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode == 503 {
+		return nil, fmt.Errorf("service unavailable")
+	}
+	respData, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != 200 {
+		return nil, fmt.Errorf("incorrect response code: %d for retrieving Ray Job details, response body: %s", resp.StatusCode, respData)
+	}
+	err = json.Unmarshal(respData, &response)
+	if err != nil {
+		return nil, err
+	}
+	return response, nil
+}
+
 func (client *rayClusterClient) GetJobDetails(jobID string) (response *RayJobDetailsResponse, err error) {
 	getJobDetailsURL := client.endpoint.String() + "/api/jobs/" + jobID
-	resp, err := http.Get(getJobDetailsURL)
+
+	req, err := http.NewRequest(http.MethodGet, getJobDetailsURL, nil)
 	if err != nil {
-		return
+		return nil, err
+	}
+	if client.bearerToken != "" {
+		req.Header.Set("Authorization", "Bearer "+client.bearerToken)
+	}
+
+	resp, err := client.httpClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode == 503 {
+		return nil, fmt.Errorf("service unavailable")
 	}
 
 	respData, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return
 	}
-
 	if resp.StatusCode != 200 {
 		return nil, fmt.Errorf("incorrect response code: %d for retrieving Ray Job details, response body: %s", resp.StatusCode, respData)
 	}
-
 	err = json.Unmarshal(respData, &response)
-	return
+	if err != nil {
+		return nil, err
+	}
+	return response, nil
 }
 
 func (client *rayClusterClient) GetJobLogs(jobID string) (logs string, err error) {
 	getJobLogsURL := client.endpoint.String() + "/api/jobs/" + jobID + "/logs"
-	resp, err := http.Get(getJobLogsURL)
+	req, err := http.NewRequest(http.MethodGet, getJobLogsURL, nil)
 	if err != nil {
-		return
+		return "", err
+	}
+	if client.bearerToken != "" {
+		req.Header.Set("Authorization", "Bearer "+client.bearerToken)
+	}
+	resp, err := client.httpClient.Do(req)
+	if err != nil {
+		return "", err
 	}
 
 	respData, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return
+		return "", err
 	}
 
 	if resp.StatusCode != 200 {

support/support.go

@@ -33,6 +33,7 @@ var (
 	TestTimeoutShort           = 1 * time.Minute
 	TestTimeoutMedium          = 2 * time.Minute
 	TestTimeoutLong            = 5 * time.Minute
+	TestTimeoutDouble          = 10 * time.Minute
 	TestTimeoutGpuProvisioning = 30 * time.Minute
 )