Create lb service for inference Service

Signed-off-by: kerthcet <[email protected]>

Author: kerthcet

README.md

@@ -90,8 +90,10 @@ spec:
 
 #### Expose the service
 
+By default, llmaz will create a ClusterIP service named like `<service>-lb` for load balancing.
+
 ```cmd
-kubectl port-forward pod/opt-125m-0 8080:8080
+kubectl port-forward svc/opt-125m-lb 8080:8080
 ```
 
 #### Get registered models

config/rbac/role.yaml

@@ -13,6 +13,18 @@ rules:
   - list
   - update
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - services
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - admissionregistration.k8s.io
   resources:

pkg/controller/inference/service_controller.go

@@ -27,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	metaapplyv1 "k8s.io/client-go/applyconfigurations/meta/v1"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
@@ -66,6 +67,7 @@ func NewServiceReconciler(client client.Client, scheme *runtime.Scheme, record r
 //+kubebuilder:rbac:groups=inference.llmaz.io,resources=services,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=inference.llmaz.io,resources=services/status,verbs=get;update;patch
 //+kubebuilder:rbac:groups=inference.llmaz.io,resources=services/finalizers,verbs=update
+//+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.
@@ -97,6 +99,11 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, err
 	}
 
+	// Create a service for the leader pods of the lws for loadbalancing.
+	if err := CreateServiceIfNotExists(ctx, r.Client, r.Scheme, service); err != nil {
+		return ctrl.Result{}, err
+	}
+
 	// Handle status.
 
 	workload := &lws.LeaderWorkerSet{}
@@ -299,3 +306,48 @@ func setControllerReferenceForWorkload(owner metav1.Object, lws *applyconfigurat
 		WithController(true))
 	return nil
 }
+
+func CreateServiceIfNotExists(ctx context.Context, k8sClient client.Client, Scheme *runtime.Scheme, service *inferenceapi.Service) error {
+	log := ctrl.LoggerFrom(ctx)
+	// The load balancing service name.
+	svcName := service.Name + "-lb"
+
+	var svc corev1.Service
+	if err := k8sClient.Get(ctx, types.NamespacedName{Name: svcName, Namespace: service.Namespace}, &svc); err != nil {
+		if client.IgnoreNotFound(err) != nil {
+			return err
+		}
+		svc = corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      svcName,
+				Namespace: service.Namespace,
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{
+						Name:       "http",
+						Protocol:   corev1.ProtocolTCP,
+						Port:       modelSource.DEFAULT_BACKEND_PORT,
+						TargetPort: intstr.FromInt(modelSource.DEFAULT_BACKEND_PORT),
+					},
+				},
+				Selector: map[string]string{
+					lws.SetNameLabelKey: service.Name,
+					// the leader pod.
+					lws.WorkerIndexLabelKey: "0",
+				},
+			},
+		}
+
+		// Set the controller owner reference for garbage collection and reconciliation.
+		if err := ctrl.SetControllerReference(service, &svc, Scheme); err != nil {
+			return err
+		}
+		// create the service in the cluster
+		log.V(2).Info("Creating service.")
+		if err := k8sClient.Create(ctx, &svc); err != nil {
+			return err
+		}
+	}
+	return nil
+}

test/util/validation/validate_service.go

@@ -84,8 +84,12 @@ func ValidateService(ctx context.Context, k8sClient client.Client, service *infe
 				return err
 			}
 		}
-		return nil
 
+		if err := k8sClient.Get(ctx, types.NamespacedName{Name: service.Name + "-lb", Namespace: service.Namespace}, &corev1.Service{}); err != nil {
+			return err
+		}
+
+		return nil
 	}, util.IntegrationTimeout, util.Interval).Should(gomega.Succeed())
 }