feat: add health check for epp cluster (#966)

* feat: add health check for epp cluster

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

* remove tls

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

* don't use tls

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

* health checking flag

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

* fix import

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

* add tls options

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

---------

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

Author: zhengkezhou1

cmd/epp/runner/runner.go

@@ -93,7 +93,8 @@ var (
 	logVerbosity  = flag.Int("v", logging.DEFAULT, "number for the log level verbosity")
 	secureServing = flag.Bool(
 		"secureServing", runserver.DefaultSecureServing, "Enables secure serving. Defaults to true.")
-	certPath = flag.String(
+	healthChecking = flag.Bool("healthChecking", runserver.DefaultHealthChecking, "Enables health checking")
+	certPath       = flag.String(
 		"certPath", "", "The path to the certificate for secure serving. The certificate and private key files "+
 			"are assumed to be named tls.crt and tls.key, respectively. If not set, and secureServing is enabled, "+
 			"then a self-signed certificate is used.")
@@ -229,6 +230,7 @@ func (r *Runner) Run(ctx context.Context) error {
 		PoolNamespacedName:                       poolNamespacedName,
 		Datastore:                                datastore,
 		SecureServing:                            *secureServing,
+		HealthChecking:                           *healthChecking,
 		CertPath:                                 *certPath,
 		RefreshPrometheusMetricsInterval:         *refreshPrometheusMetricsInterval,
 		Director:                                 director,

pkg/epp/server/runserver.go

@@ -20,12 +20,15 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+
 	"time"
 
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/go-logr/logr"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/health"
+	healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -46,6 +49,7 @@ type ExtProcServerRunner struct {
 	PoolNamespacedName                       types.NamespacedName
 	Datastore                                datastore.Datastore
 	SecureServing                            bool
+	HealthChecking                           bool
 	CertPath                                 string
 	RefreshPrometheusMetricsInterval         time.Duration
 	Director                                 *requestcontrol.Director
@@ -66,6 +70,7 @@ const (
 	DefaultRefreshMetricsInterval                   = 50 * time.Millisecond            // default for --refreshMetricsInterval
 	DefaultRefreshPrometheusMetricsInterval         = 5 * time.Second                  // default for --refreshPrometheusMetricsInterval
 	DefaultSecureServing                            = true                             // default for --secureServing
+	DefaultHealthChecking                           = false                            // default for --healthChecking
 )
 
 // NewDefaultExtProcServerRunner creates a runner with default values.
@@ -77,6 +82,7 @@ func NewDefaultExtProcServerRunner() *ExtProcServerRunner {
 		DestinationEndpointHintMetadataNamespace: DefaultDestinationEndpointHintMetadataNamespace,
 		PoolNamespacedName:                       types.NamespacedName{Name: DefaultPoolName, Namespace: DefaultPoolNamespace},
 		SecureServing:                            DefaultSecureServing,
+		HealthChecking:                           DefaultHealthChecking,
 		RefreshPrometheusMetricsInterval:         DefaultRefreshPrometheusMetricsInterval,
 		// Dependencies can be assigned later.
 	}
@@ -152,6 +158,16 @@ func (r *ExtProcServerRunner) AsRunnable(logger logr.Logger) manager.Runnable {
 			extProcServer,
 		)
 
+		if r.HealthChecking {
+			healthcheck := health.NewServer()
+			healthgrpc.RegisterHealthServer(srv,
+				healthcheck,
+			)
+			svcName := extProcPb.ExternalProcessor_ServiceDesc.ServiceName
+			logger.Info("Setting ExternalProcessor service status to SERVING", "serviceName", svcName)
+			healthcheck.SetServingStatus(svcName, healthgrpc.HealthCheckResponse_SERVING)
+		}
+
 		// Forward to the gRPC runnable.
 		return runnable.GRPCServer("ext-proc", srv, r.GrpcPort).Start(ctx)
 	}))

test/testdata/envoy.yaml

@@ -170,6 +170,16 @@ data:
                 max_pending_requests: 40000
                 max_requests: 40000
                 max_retries: 1024
+          health_checks:
+            - timeout: 2s
+              interval: 10s
+              unhealthy_threshold: 3
+              healthy_threshold: 2
+              reuse_connection: true
+              grpc_health_check:
+                service_name: "envoy.service.ext_proc.v3.ExternalProcessor"
+              tls_options:
+                alpn_protocols: ["h2"]  
           # This ensures that envoy accepts untrusted certificates. We tried to explicitly
           # set TrustChainVerification to ACCEPT_UNSTRUSTED, but that actually didn't work
           # and what worked is setting the common_tls_context to empty.                
@@ -197,7 +207,6 @@ data:
                         socket_address:
                           address: vllm-llama3-8b-instruct-epp.$E2E_NS
                           port_value: 9002
-                    health_status: HEALTHY
                     load_balancing_weight: 1
 ---
 apiVersion: apps/v1

test/testdata/inferencepool-e2e.yaml

@@ -109,7 +109,7 @@ rules:
   - subjectaccessreviews
   verbs:
   - create
---- 
+---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata: