Use structured logging (kubernetes-sigs#330)

* Use structured logging

All logging calls are rewritten to use structured logging.

* test/integration: Use logutil.Fatal

Author: tchap

pkg/ext-proc/backend/datastore.go

@@ -98,7 +98,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/fake.go

@@ -5,6 +5,7 @@ import (
 
 	klog "k8s.io/klog/v2"
 	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha1"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 )
 
 type FakePodMetricsClient struct {
@@ -16,7 +17,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(logutil.VERBOSE).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

@@ -11,6 +11,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
 	}
 	if c.Datastore.inferencePool == nil || !reflect.DeepEqual(serverPool.Spec.Selector, c.Datastore.inferencePool.Spec.Selector) {
@@ -49,7 +51,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.V(logutil.DEFAULT).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.V(logutil.DEFAULT).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.V(logutil.DEFAULT).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.V(logutil.DEFAULT).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.V(logutil.DEFAULT).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.V(logutil.DEFAULT).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{

pkg/ext-proc/handlers/response.go

@@ -12,9 +12,9 @@ import (
 
 // HandleResponseHeaders processes response headers from the backend model server.
 func (s *Server) HandleResponseHeaders(reqCtx *RequestContext, req *extProcPb.ProcessingRequest) (*extProcPb.ProcessingResponse, error) {
-	klog.V(logutil.VERBOSE).Info("Processing ResponseHeaders")
+	klog.V(logutil.VERBOSE).InfoS("Processing ResponseHeaders")
 	h := req.Request.(*extProcPb.ProcessingRequest_ResponseHeaders)
-	klog.V(logutil.VERBOSE).Infof("Headers before: %+v\n", h)
+	klog.V(logutil.VERBOSE).InfoS("Headers before", "headers", h)
 
 	resp := &extProcPb.ProcessingResponse{
 		Response: &extProcPb.ProcessingResponse_ResponseHeaders{
@@ -66,7 +66,7 @@ func (s *Server) HandleResponseHeaders(reqCtx *RequestContext, req *extProcPb.Pr
     }
 }*/
 func (s *Server) HandleResponseBody(reqCtx *RequestContext, req *extProcPb.ProcessingRequest) (*extProcPb.ProcessingResponse, error) {
-	klog.V(logutil.VERBOSE).Info("Processing HandleResponseBody")
+	klog.V(logutil.VERBOSE).InfoS("Processing HandleResponseBody")
 	body := req.Request.(*extProcPb.ProcessingRequest_ResponseBody)
 
 	res := Response{}
@@ -81,7 +81,7 @@ func (s *Server) HandleResponseBody(reqCtx *RequestContext, req *extProcPb.Proce
 	// TODO(https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/178)
 	// will add the processing for streaming case.
 	reqCtx.ResponseComplete = true
-	klog.V(logutil.VERBOSE).Infof("Response: %+v", res)
+	klog.V(logutil.VERBOSE).InfoS("Response generated", "response", res)
 
 	resp := &extProcPb.ProcessingResponse{
 		Response: &extProcPb.ProcessingResponse_ResponseBody{