Use structured logging

All logging calls are rewritten to use structured logging.

Author: tchap

pkg/ext-proc/backend/datastore.go

@@ -94,7 +94,7 @@ func RandomWeightedDraw(model *v1alpha1.InferenceModel, seed int64) string {
 	for _, model := range model.Spec.TargetModels {
 		weights += *model.Weight
 	}
-	klog.V(logutil.VERBOSE).Infof("Weights for Model(%v) total to: %v", model.Name, weights)
+	klog.V(logutil.VERBOSE).InfoS("Weights for model computed", "model", model.Name, "weights", weights)
 	randomVal := r.Int31n(weights)
 	for _, model := range model.Spec.TargetModels {
 		if randomVal < *model.Weight {

pkg/ext-proc/backend/endpointslice_reconciler.go

@@ -17,9 +17,7 @@ import (
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 )
 
-var (
-	serviceOwnerLabel = "kubernetes.io/service-name"
-)
+var serviceOwnerLabel = "kubernetes.io/service-name"
 
 type EndpointSliceReconciler struct {
 	client.Client
@@ -33,15 +31,15 @@ type EndpointSliceReconciler struct {
 func (c *EndpointSliceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	inferencePool, err := c.Datastore.getInferencePool()
 	if err != nil {
-		klog.V(logutil.DEFAULT).Infof("Skipping reconciling EndpointSlice because the InferencePool is not available yet: %v", err)
+		klog.V(logutil.DEFAULT).InfoS("Skipping reconciling EndpointSlice because the InferencePool is not available yet", "error", err)
 		return ctrl.Result{Requeue: true, RequeueAfter: time.Second}, nil
 	}
 
-	klog.V(logutil.DEFAULT).Info("Reconciling EndpointSlice ", req.NamespacedName)
+	klog.V(logutil.DEFAULT).InfoS("Reconciling EndpointSlice", "name", req.NamespacedName)
 
 	endpointSlice := &discoveryv1.EndpointSlice{}
 	if err := c.Get(ctx, req.NamespacedName, endpointSlice); err != nil {
-		klog.Errorf("Unable to get EndpointSlice: %v", err)
+		klog.ErrorS(err, "Unable to get EndpointSlice", "name", req.NamespacedName)
 		return ctrl.Result{}, err
 	}
 	c.updateDatastore(endpointSlice, inferencePool)
@@ -52,30 +50,31 @@ func (c *EndpointSliceReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 // TODO: Support multiple endpointslices for a single service
 func (c *EndpointSliceReconciler) updateDatastore(
 	slice *discoveryv1.EndpointSlice,
-	inferencePool *v1alpha1.InferencePool) {
+	inferencePool *v1alpha1.InferencePool,
+) {
 	podMap := make(map[Pod]bool)
 
 	for _, endpoint := range slice.Endpoints {
-		klog.V(logutil.DEFAULT).Infof("Zone: %v \n endpoint: %+v \n", c.Zone, endpoint)
+		klog.V(logutil.DEFAULT).InfoS("Processing endpoint", "zone", c.Zone, "endpoint", endpoint)
 		if c.validPod(endpoint) {
 			pod := Pod{
 				Name:    endpoint.TargetRef.Name,
 				Address: endpoint.Addresses[0] + ":" + strconv.Itoa(int(inferencePool.Spec.TargetPortNumber)),
 			}
 			podMap[pod] = true
-			klog.V(logutil.DEFAULT).Infof("Storing pod %v", pod)
+			klog.V(logutil.DEFAULT).InfoS("Storing pod", "pod", pod)
 			c.Datastore.pods.Store(pod, true)
 		}
 	}
 
 	removeOldPods := func(k, v any) bool {
 		pod, ok := k.(Pod)
 		if !ok {
-			klog.Errorf("Unable to cast key to Pod: %v", k)
+			klog.V(logutil.DEFAULT).ErrorS(nil, "Unable to cast key to Pod", "key", k)
 			return false
 		}
 		if _, ok := podMap[pod]; !ok {
-			klog.V(logutil.DEFAULT).Infof("Removing pod %v", pod)
+			klog.V(logutil.DEFAULT).InfoS("Removing pod", "pod", pod)
 			c.Datastore.pods.Delete(pod)
 		}
 		return true
@@ -105,5 +104,4 @@ func (c *EndpointSliceReconciler) SetupWithManager(mgr ctrl.Manager) error {
 func (c *EndpointSliceReconciler) validPod(endpoint discoveryv1.Endpoint) bool {
 	validZone := c.Zone == "" || c.Zone != "" && *endpoint.Zone == c.Zone
 	return validZone && *endpoint.Conditions.Ready
-
 }

pkg/ext-proc/backend/fake.go

@@ -16,7 +16,7 @@ func (f *FakePodMetricsClient) FetchMetrics(ctx context.Context, pod Pod, existi
 	if err, ok := f.Err[pod]; ok {
 		return nil, err
 	}
-	klog.V(1).Infof("pod: %+v\n existing: %+v \n new: %+v \n", pod, existing, f.Res[pod])
+	klog.V(1).InfoS("Fetching metrics for pod", "pod", pod, "existing", existing, "new", f.Res[pod])
 	return f.Res[pod], nil
 }
 

pkg/ext-proc/backend/inferencemodel_reconciler.go

@@ -26,19 +26,21 @@ func (c *InferenceModelReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 	if req.Namespace != c.PoolNamespacedName.Namespace {
 		return ctrl.Result{}, nil
 	}
-	klog.V(1).Infof("Reconciling InferenceModel %v", req.NamespacedName)
+
+	klogV := klog.V(logutil.DEFAULT)
+	klogV.InfoS("Reconciling InferenceModel", "name", req.NamespacedName)
 
 	infModel := &v1alpha1.InferenceModel{}
 	if err := c.Get(ctx, req.NamespacedName, infModel); err != nil {
 		if errors.IsNotFound(err) {
-			klog.V(1).Infof("InferenceModel %v not found. Removing from datastore since object must be deleted", req.NamespacedName)
+			klogV.InfoS("InferenceModel not found. Removing from datastore since object must be deleted", "name", req.NamespacedName)
 			c.Datastore.InferenceModels.Delete(infModel.Spec.ModelName)
 			return ctrl.Result{}, nil
 		}
-		klog.Error(err, "Unable to get InferenceModel")
+		klogV.ErrorS(err, "Unable to get InferenceModel", "name", req.NamespacedName)
 		return ctrl.Result{}, err
 	} else if !infModel.DeletionTimestamp.IsZero() {
-		klog.V(1).Infof("InferenceModel %v is marked for deletion. Removing from datastore", req.NamespacedName)
+		klogV.InfoS("InferenceModel is marked for deletion. Removing from datastore", "name", req.NamespacedName)
 		c.Datastore.InferenceModels.Delete(infModel.Spec.ModelName)
 		return ctrl.Result{}, nil
 	}
@@ -48,13 +50,15 @@ func (c *InferenceModelReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 }
 
 func (c *InferenceModelReconciler) updateDatastore(infModel *v1alpha1.InferenceModel) {
+	klogV := klog.V(logutil.DEFAULT)
+
 	if infModel.Spec.PoolRef.Name == c.PoolNamespacedName.Name {
-		klog.V(1).Infof("Incoming pool ref %v, server pool name: %v", infModel.Spec.PoolRef, c.PoolNamespacedName.Name)
-		klog.V(1).Infof("Adding/Updating InferenceModel: %v", infModel.Spec.ModelName)
+		klogV.InfoS("Updating datastore", "poolRef", infModel.Spec.PoolRef, "serverPoolName", c.PoolNamespacedName)
+		klogV.InfoS("Adding/Updating InferenceModel", "modelName", infModel.Spec.ModelName)
 		c.Datastore.InferenceModels.Store(infModel.Spec.ModelName, infModel)
 		return
 	}
-	klog.V(logutil.DEFAULT).Infof("Removing/Not adding InferenceModel: %v", infModel.Spec.ModelName)
+	klogV.InfoS("Removing/Not adding InferenceModel", "modelName", infModel.Spec.ModelName)
 	// If we get here. The model is not relevant to this pool, remove.
 	c.Datastore.InferenceModels.Delete(infModel.Spec.ModelName)
 }

pkg/ext-proc/backend/inferencepool_reconciler.go

@@ -10,6 +10,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha1"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 )
 
 // InferencePoolReconciler utilizes the controller runtime to reconcile Instance Gateway resources
@@ -28,11 +29,12 @@ func (c *InferencePoolReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 	if req.NamespacedName.Name != c.PoolNamespacedName.Name || req.NamespacedName.Namespace != c.PoolNamespacedName.Namespace {
 		return ctrl.Result{}, nil
 	}
-	klog.V(1).Info("reconciling InferencePool", req.NamespacedName)
+	klogV := klog.V(logutil.DEFAULT)
+	klogV.InfoS("Reconciling InferencePool", "name", req.NamespacedName)
 
 	serverPool := &v1alpha1.InferencePool{}
 	if err := c.Get(ctx, req.NamespacedName, serverPool); err != nil {
-		klog.Error(err, "unable to get InferencePool")
+		klogV.ErrorS(err, "Unable to get InferencePool", "name", req.NamespacedName)
 		return ctrl.Result{}, err
 	}
 
@@ -45,7 +47,7 @@ func (c *InferencePoolReconciler) updateDatastore(serverPool *v1alpha1.Inference
 	pool, _ := c.Datastore.getInferencePool()
 	if pool == nil ||
 		serverPool.ObjectMeta.ResourceVersion != pool.ObjectMeta.ResourceVersion {
-		klog.Infof("Updating inference pool to %v/%v", serverPool.ObjectMeta.Namespace, serverPool.ObjectMeta.Name)
+		klog.V(logutil.DEFAULT).InfoS("Updating inference pool", "target", klog.KMetadata(&serverPool.ObjectMeta))
 		c.Datastore.setInferencePool(serverPool)
 	}
 }

pkg/ext-proc/backend/provider.go

@@ -63,10 +63,10 @@ func (p *Provider) Init(refreshPodsInterval, refreshMetricsInterval, refreshProm
 	p.refreshPodsOnce()
 
 	if err := p.refreshMetricsOnce(); err != nil {
-		klog.Errorf("Failed to init metrics: %v", err)
+		klog.ErrorS(err, "Failed to init metrics")
 	}
 
-	klog.Infof("Initialized pods and metrics: %+v", p.AllPodMetrics())
+	klog.InfoS("Initialized pods and metrics", "metrics", p.AllPodMetrics())
 
 	// periodically refresh pods
 	go func() {
@@ -81,7 +81,7 @@ func (p *Provider) Init(refreshPodsInterval, refreshMetricsInterval, refreshProm
 		for {
 			time.Sleep(refreshMetricsInterval)
 			if err := p.refreshMetricsOnce(); err != nil {
-				klog.V(logutil.TRACE).Infof("Failed to refresh metrics: %v", err)
+				klog.V(logutil.TRACE).ErrorS(err, "Failed to refresh metrics")
 			}
 		}
 	}()
@@ -95,11 +95,11 @@ func (p *Provider) Init(refreshPodsInterval, refreshMetricsInterval, refreshProm
 	}()
 
 	// Periodically print out the pods and metrics for DEBUGGING.
-	if klog.V(logutil.DEBUG).Enabled() {
+	if klogV := klog.V(logutil.DEBUG); klogV.Enabled() {
 		go func() {
 			for {
 				time.Sleep(5 * time.Second)
-				klog.Infof("===DEBUG: Current Pods and metrics: %+v", p.AllPodMetrics())
+				klogV.InfoS("Current Pods and metrics gathered", "metrics", p.AllPodMetrics())
 			}
 		}()
 	}
@@ -138,18 +138,19 @@ func (p *Provider) refreshPodsOnce() {
 }
 
 func (p *Provider) refreshMetricsOnce() error {
+	klogV := klog.V(logutil.TRACE)
 	ctx, cancel := context.WithTimeout(context.Background(), fetchMetricsTimeout)
 	defer cancel()
 	start := time.Now()
 	defer func() {
 		d := time.Since(start)
 		// TODO: add a metric instead of logging
-		klog.V(logutil.TRACE).Infof("Refreshed metrics in %v", d)
+		klogV.InfoS("Metrics refreshed", "duration", d)
 	}()
 	var wg sync.WaitGroup
 	errCh := make(chan error)
 	processOnePod := func(key, value any) bool {
-		klog.V(logutil.TRACE).Infof("Processing pod %v and metric %v", key, value)
+		klogV.InfoS("Pod and metric being processed", "pod", key, "metric", value)
 		pod := key.(Pod)
 		existing := value.(*PodMetrics)
 		wg.Add(1)
@@ -161,7 +162,7 @@ func (p *Provider) refreshMetricsOnce() error {
 				return
 			}
 			p.UpdatePodMetrics(pod, updated)
-			klog.V(logutil.TRACE).Infof("Updated metrics for pod %s: %v", pod, updated.Metrics)
+			klogV.InfoS("Updated metrics for pod", "pod", pod, "metrics", updated.Metrics)
 		}()
 		return true
 	}
@@ -185,7 +186,7 @@ func (p *Provider) refreshMetricsOnce() error {
 }
 
 func (p *Provider) flushPrometheusMetricsOnce() {
-	klog.V(logutil.DEBUG).Infof("Flushing Prometheus Metrics")
+	klog.V(logutil.DEBUG).InfoS("Flushing Prometheus Metrics")
 
 	pool, _ := p.datastore.getInferencePool()
 	if pool == nil {

pkg/ext-proc/backend/vllm/metrics.go

@@ -32,8 +32,7 @@ const (
 	KvCacheMaxTokenCapacityMetricName = "vllm:gpu_cache_max_token_capacity"
 )
 
-type PodMetricsClientImpl struct {
-}
+type PodMetricsClientImpl struct{}
 
 // FetchMetrics fetches metrics from a given pod.
 func (p *PodMetricsClientImpl) FetchMetrics(
@@ -46,19 +45,20 @@ func (p *PodMetricsClientImpl) FetchMetrics(
 	url := fmt.Sprintf("http://%s/metrics", pod.Address)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
+		klog.ErrorS(err, "Failed create HTTP request", "method", http.MethodGet, "url", url)
 		return nil, fmt.Errorf("failed to create request: %v", err)
 	}
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		klog.Errorf("failed to fetch metrics from %s: %v", pod, err)
+		klog.ErrorS(err, "Failed to fetch metrics", "pod", pod)
 		return nil, fmt.Errorf("failed to fetch metrics from %s: %w", pod, err)
 	}
 	defer func() {
 		_ = resp.Body.Close()
 	}()
 
 	if resp.StatusCode != http.StatusOK {
-		klog.Errorf("unexpected status code from %s: %v", pod, resp.StatusCode)
+		klog.ErrorS(nil, "Unexpected status code returned", "pod", pod, "statusCode", resp.StatusCode)
 		return nil, fmt.Errorf("unexpected status code from %s: %v", pod, resp.StatusCode)
 	}
 
@@ -138,7 +138,7 @@ func promToPodMetrics(
 func getLatestLoraMetric(metricFamilies map[string]*dto.MetricFamily) (*dto.Metric, time.Time, error) {
 	loraRequests, ok := metricFamilies[LoraRequestInfoMetricName]
 	if !ok {
-		klog.Warningf("metric family %q not found", LoraRequestInfoMetricName)
+		klog.ErrorS(nil, "Metric family not found", "name", LoraRequestInfoMetricName)
 		return nil, time.Time{}, fmt.Errorf("metric family %q not found", LoraRequestInfoMetricName)
 	}
 	var latestTs float64
@@ -157,7 +157,7 @@ func getLatestLoraMetric(metricFamilies map[string]*dto.MetricFamily) (*dto.Metr
 func getLatestMetric(metricFamilies map[string]*dto.MetricFamily, metricName string) (*dto.Metric, error) {
 	mf, ok := metricFamilies[metricName]
 	if !ok {
-		klog.Warningf("metric family %q not found", metricName)
+		klog.ErrorS(nil, "Metric family not found", "name", metricName)
 		return nil, fmt.Errorf("metric family %q not found", metricName)
 	}
 	if len(mf.GetMetric()) == 0 {
@@ -171,6 +171,6 @@ func getLatestMetric(metricFamilies map[string]*dto.MetricFamily, metricName str
 			latest = m
 		}
 	}
-	klog.V(logutil.TRACE).Infof("Got metric value %+v for metric %v", latest, metricName)
+	klog.V(logutil.TRACE).InfoS("Metric value selected", "value", latest, "metric", metricName)
 	return latest, nil
 }

pkg/ext-proc/handlers/request.go

@@ -19,23 +19,24 @@ import (
 // parameter.
 // Envoy sends the request body to ext proc before sending the request to the backend server.
 func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.ProcessingRequest) (*extProcPb.ProcessingResponse, error) {
-	klog.V(logutil.VERBOSE).Infof("Handling request body")
+	klogV := klog.V(logutil.VERBOSE)
+	klogV.InfoS("Handling request body")
 
 	// Unmarshal request body (must be JSON).
 	v := req.Request.(*extProcPb.ProcessingRequest_RequestBody)
 	var rb map[string]interface{}
 	if err := json.Unmarshal(v.RequestBody.Body, &rb); err != nil {
-		klog.Errorf("Error unmarshaling request body: %v", err)
+		klog.ErrorS(err, "Error unmarshaling request body")
 		return nil, fmt.Errorf("error unmarshaling request body: %v", err)
 	}
-	klog.V(logutil.VERBOSE).Infof("Request body: %v", rb)
+	klogV.InfoS("Request body unmarshalled", "body", rb)
 
 	// Resolve target models.
 	model, ok := rb["model"].(string)
 	if !ok {
 		return nil, errors.New("model not found in request")
 	}
-	klog.V(logutil.VERBOSE).Infof("Model requested: %v", model)
+	klogV.InfoS("Model requested", "model", model)
 	modelName := model
 
 	// NOTE: The nil checking for the modelObject means that we DO allow passthrough currently.
@@ -56,7 +57,7 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 		ResolvedTargetModel: modelName,
 		Critical:            backend.IsCritical(modelObj),
 	}
-	klog.V(logutil.VERBOSE).Infof("LLM Request: %+v", llmReq)
+	klogV.InfoS("LLM request assembled", "request", llmReq)
 
 	requestBody := v.RequestBody.Body
 	var err error
@@ -65,17 +66,17 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 		rb["model"] = llmReq.ResolvedTargetModel
 		requestBody, err = json.Marshal(rb)
 		if err != nil {
-			klog.Errorf("Error marshaling request body: %v", err)
+			klog.V(logutil.DEFAULT).ErrorS(err, "Error marshaling request body")
 			return nil, fmt.Errorf("error marshaling request body: %v", err)
 		}
-		klog.V(logutil.VERBOSE).Infof("Updated body: %v", string(requestBody))
+		klogV.InfoS("Updated request body marshalled", "body", string(requestBody))
 	}
 
 	targetPod, err := s.scheduler.Schedule(llmReq)
 	if err != nil {
 		return nil, fmt.Errorf("failed to find target pod: %w", err)
 	}
-	klog.V(logutil.VERBOSE).Infof("Selected target model %v in target pod: %v\n", llmReq.ResolvedTargetModel, targetPod)
+	klogV.InfoS("Target model and pod selected", "model", llmReq.ResolvedTargetModel, "pod", targetPod)
 
 	reqCtx.Model = llmReq.Model
 	reqCtx.ResolvedTargetModel = llmReq.ResolvedTargetModel
@@ -101,7 +102,7 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 	}
 	// Print headers for debugging
 	for _, header := range headers {
-		klog.V(logutil.VERBOSE).Infof("[request_body] Header Key: %s, Header Value: %s\n", header.Header.Key, header.Header.RawValue)
+		klog.V(logutil.DEBUG).InfoS("Request body header", "key", header.Header.Key, "value", header.Header.RawValue)
 	}
 
 	resp := &extProcPb.ProcessingResponse{
@@ -136,10 +137,9 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 }
 
 func HandleRequestHeaders(reqCtx *RequestContext, req *extProcPb.ProcessingRequest) *extProcPb.ProcessingResponse {
-	klog.V(logutil.VERBOSE).Info("Handling request headers ...")
 	r := req.Request
 	h := r.(*extProcPb.ProcessingRequest_RequestHeaders)
-	klog.V(logutil.VERBOSE).Infof("Headers: %+v\n", h)
+	klog.V(logutil.VERBOSE).InfoS("Handling request headers", "headers", h)
 
 	resp := &extProcPb.ProcessingResponse{
 		Response: &extProcPb.ProcessingResponse_RequestHeaders{