feat: set appProtocol based on target service

ray-operator/controllers/ray/common/ingress.go

@@ -44,8 +44,8 @@ func BuildIngressForHeadService(ctx context.Context, cluster rayv1.RayCluster) (
 	pathType := networkingv1.PathTypeExact
 	servicePorts := getServicePorts(cluster)
 	dashboardPort := int32(utils.DefaultDashboardPort)
-	if port, ok := servicePorts["dashboard"]; ok {
-		dashboardPort = port
+	if portInfo, ok := servicePorts["dashboard"]; ok {
+		dashboardPort = portInfo.PortNumber
 	}
 
 	headSvcName, err := utils.GenerateHeadServiceName(utils.RayClusterCRD, cluster.Spec, cluster.Name)

ray-operator/controllers/ray/common/route.go

@@ -28,8 +28,8 @@ func BuildRouteForHeadService(cluster rayv1.RayCluster) (*routev1.Route, error)
 
 	servicePorts := getServicePorts(cluster)
 	dashboardPort := utils.DefaultDashboardPort
-	if port, ok := servicePorts["dashboard"]; ok {
-		dashboardPort = int(port)
+	if portInfo, ok := servicePorts[utils.DashboardPortName]; ok {
+		dashboardPort = int(portInfo.PortNumber)
 	}
 
 	weight := int32(100)

ray-operator/controllers/ray/common/service.go

@@ -62,12 +62,16 @@ func BuildServiceForHeadPod(ctx context.Context, cluster rayv1.RayCluster, label
 	defaultNamespace := cluster.Namespace
 	defaultType := cluster.Spec.HeadGroupSpec.ServiceType
 
-	defaultAppProtocol := utils.DefaultServiceAppProtocol
-	// `portsInt` is a map of port names to port numbers, while `ports` is a list of ServicePort objects
-	portsInt := getServicePorts(cluster)
+	// `servicePorts` is a map of port names to port infos, while `ports` is a list of ServicePort objects
+	servicePorts := getServicePorts(cluster)
 	ports := []corev1.ServicePort{}
-	for name, port := range portsInt {
-		svcPort := corev1.ServicePort{Name: name, Port: port, AppProtocol: &defaultAppProtocol}
+	for svc, portInfo := range servicePorts {
+		appProtocol := utils.DefaultServiceAppProtocol
+		if portInfo.AppProtocol != "" {
+			appProtocol = portInfo.AppProtocol
+		}
+
+		svcPort := corev1.ServicePort{Name: svc, Port: portInfo.PortNumber, AppProtocol: &appProtocol}
 		ports = append(ports, svcPort)
 	}
 	if cluster.Spec.HeadGroupSpec.HeadService != nil {
@@ -198,12 +202,17 @@ func BuildServeService(ctx context.Context, rayService rayv1.RayService, rayClus
 		defaultType = rayService.Spec.RayClusterSpec.HeadGroupSpec.ServiceType
 	}
 
-	// `portsInt` is a map of port names to port numbers, while `ports` is a list of ServicePort objects
-	portsInt := getServicePorts(rayCluster)
+	// `servicePorts` is a map of port names to port infos, while `ports` is a list of ServicePort objects
+	servicePorts := getServicePorts(rayCluster)
 	ports := []corev1.ServicePort{}
-	for name, port := range portsInt {
-		if name == utils.ServingPortName {
-			svcPort := corev1.ServicePort{Name: name, Port: port}
+	for svc, portInfo := range servicePorts {
+		if svc == utils.ServingPortName {
+			appProtocol := utils.DefaultServiceAppProtocol
+			if portInfo.AppProtocol != "" {
+				appProtocol = portInfo.AppProtocol
+			}
+
+			svcPort := corev1.ServicePort{Name: name, Port: portInfo.PortNumber, AppProtocol: &appProtocol}
 			ports = append(ports, svcPort)
 			break
 		}
@@ -360,7 +369,7 @@ func setLabelsforUserProvidedService(service *corev1.Service, labels map[string]
 }
 
 // getServicePorts will either user passing ports or default ports to create service.
-func getServicePorts(cluster rayv1.RayCluster) map[string]int32 {
+func getServicePorts(cluster rayv1.RayCluster) map[string]portInfo {
 	ports, err := getPortsFromCluster(cluster)
 	// Assign default ports
 	if err != nil || len(ports) == 0 {
@@ -369,35 +378,55 @@ func getServicePorts(cluster rayv1.RayCluster) map[string]int32 {
 
 	// check if metrics port is defined. If not, add default value for it.
 	if _, metricsDefined := ports[utils.MetricsPortName]; !metricsDefined {
-		ports[utils.MetricsPortName] = utils.DefaultMetricsPort
+		ports[utils.MetricsPortName] = portInfo{PortNumber: utils.DefaultMetricsPort, AppProtocol: utils.HTTPProtocol}
 	}
 
 	return ports
 }
 
+type portInfo struct {
+	PortNumber  int32
+	AppProtocol string
+}
+
 // getPortsFromCluster get the ports from head container and directly map them in service
 // It's user's responsibility to maintain rayStartParam ports and container ports mapping
 // TODO: Consider to infer ports from rayStartParams (source of truth) in the future.
-func getPortsFromCluster(cluster rayv1.RayCluster) (map[string]int32, error) {
-	svcPorts := map[string]int32{}
+func getPortsFromCluster(cluster rayv1.RayCluster) (map[string]portInfo, error) {
+	svcPorts := make(map[string]portInfo)
 
 	cPorts := cluster.Spec.HeadGroupSpec.Template.Spec.Containers[utils.RayContainerIndex].Ports
 	for _, port := range cPorts {
 		if port.Name == "" {
 			port.Name = fmt.Sprint(port.ContainerPort) + "-port"
 		}
-		svcPorts[port.Name] = port.ContainerPort
+		svcPorts[port.Name] = portInfo{PortNumber: port.ContainerPort}
 	}
 
 	return svcPorts, nil
 }
 
-func getDefaultPorts() map[string]int32 {
-	return map[string]int32{
-		utils.ClientPortName:    utils.DefaultClientPort,
-		utils.RedisPortName:     utils.DefaultRedisPort,
-		utils.DashboardPortName: utils.DefaultDashboardPort,
-		utils.MetricsPortName:   utils.DefaultMetricsPort,
-		utils.ServingPortName:   utils.DefaultServingPort,
+func getDefaultPorts() map[string]portInfo {
+	return map[string]portInfo{
+		utils.ClientPortName: {
+			PortNumber:  utils.DefaultClientPort,
+			AppProtocol: utils.GRPCProtocol,
+		},
+		utils.RedisPortName: {
+			PortNumber:  utils.DefaultRedisPort,
+			AppProtocol: utils.GRPCProtocol,
+		},
+		utils.DashboardPortName: {
+			PortNumber:  utils.DefaultDashboardPort,
+			AppProtocol: utils.HTTPProtocol,
+		},
+		utils.MetricsPortName: {
+			PortNumber:  utils.DefaultMetricsPort,
+			AppProtocol: utils.HTTPProtocol,
+		},
+		utils.ServingPortName: {
+			PortNumber:  utils.DefaultServingPort,
+			AppProtocol: utils.HTTPProtocol,
+		},
 	}
 }

ray-operator/controllers/ray/common/service_test.go

@@ -191,19 +191,19 @@ func TestBuildServiceForHeadPodWithAnnotations(t *testing.T) {
 }
 
 func TestGetPortsFromCluster(t *testing.T) {
-	svcPorts, err := getPortsFromCluster(*instanceWithWrongSvc)
+	servicePorts, err := getPortsFromCluster(*instanceWithWrongSvc)
 	assert.Nil(t, err)
 
 	// getPortsFromCluster creates service ports based on the container ports.
 	// It will assign a generated service port name if the container port name
 	// is not defined. To compare created service ports with container ports,
 	// all generated service port names need to be reverted to empty strings.
 	svcNames := map[int32]string{}
-	for name, port := range svcPorts {
-		if name == (fmt.Sprint(port) + "-port") {
-			name = ""
+	for svc, portInfo := range servicePorts {
+		if svc == (fmt.Sprint(portInfo.PortNumber) + "-port") {
+			svc = ""
 		}
-		svcNames[port] = name
+		svcNames[portInfo.PortNumber] = svc
 	}
 
 	cPorts := instanceWithWrongSvc.Spec.HeadGroupSpec.Template.Spec.Containers[utils.RayContainerIndex].Ports
@@ -222,10 +222,10 @@ func TestGetServicePortsWithMetricsPort(t *testing.T) {
 
 	// Test case 1: No ports are specified by the user.
 	cluster.Spec.HeadGroupSpec.Template.Spec.Containers[0].Ports = []corev1.ContainerPort{}
-	ports := getServicePorts(*cluster)
+	servicePorts := getServicePorts(*cluster)
 	// Verify that getServicePorts sets the default metrics port when the user doesn't specify any ports.
-	if ports[utils.MetricsPortName] != int32(utils.DefaultMetricsPort) {
-		t.Fatalf("Expected `%v` but got `%v`", int32(utils.DefaultMetricsPort), ports[utils.MetricsPortName])
+	if servicePorts[utils.MetricsPortName].PortNumber != int32(utils.DefaultMetricsPort) {
+		t.Fatalf("Expected `%v` but got `%v`", int32(utils.DefaultMetricsPort), servicePorts[utils.MetricsPortName].PortNumber)
 	}
 
 	// Test case 2: Only a random port is specified by the user.
@@ -235,10 +235,10 @@ func TestGetServicePortsWithMetricsPort(t *testing.T) {
 			ContainerPort: 1234,
 		},
 	}
-	ports = getServicePorts(*cluster)
+	servicePorts = getServicePorts(*cluster)
 	// Verify that getServicePorts sets the default metrics port when the user doesn't specify the metrics port but specifies other ports.
-	if ports[utils.MetricsPortName] != int32(utils.DefaultMetricsPort) {
-		t.Fatalf("Expected `%v` but got `%v`", int32(utils.DefaultMetricsPort), ports[utils.MetricsPortName])
+	if servicePorts[utils.MetricsPortName].PortNumber != int32(utils.DefaultMetricsPort) {
+		t.Fatalf("Expected `%v` but got `%v`", int32(utils.DefaultMetricsPort), servicePorts[utils.MetricsPortName].PortNumber)
 	}
 
 	// Test case 3: A custom metrics port is specified by the user.
@@ -248,10 +248,10 @@ func TestGetServicePortsWithMetricsPort(t *testing.T) {
 		ContainerPort: customMetricsPort,
 	}
 	cluster.Spec.HeadGroupSpec.Template.Spec.Containers[0].Ports = append(cluster.Spec.HeadGroupSpec.Template.Spec.Containers[0].Ports, metricsPort)
-	ports = getServicePorts(*cluster)
+	servicePorts = getServicePorts(*cluster)
 	// Verify that getServicePorts uses the user's custom metrics port when the user specifies the metrics port.
-	if ports[utils.MetricsPortName] != customMetricsPort {
-		t.Fatalf("Expected `%v` but got `%v`", customMetricsPort, ports[utils.MetricsPortName])
+	if servicePorts[utils.MetricsPortName].PortNumber != customMetricsPort {
+		t.Fatalf("Expected `%v` but got `%v`", customMetricsPort, servicePorts[utils.MetricsPortName].PortNumber)
 	}
 }
 

ray-operator/controllers/ray/utils/constant.go

@@ -78,6 +78,9 @@ const (
 	// The default AppProtocol for Kubernetes service
 	DefaultServiceAppProtocol = "tcp"
 
+	HTTPProtocol = "http"
+	GRPCProtocol = "grpc"
+
 	// The default application name
 	ApplicationName = "kuberay"