From 23e4c1770fb8b178279fb9305c065b781baa2e4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BD=97=E6=B3=BD=E8=BD=A9?= Date: Fri, 14 Feb 2025 10:46:38 +0000 Subject: [PATCH] resolve comments --- pkg/ext-proc/backend/pod_metrics_refresher.go | 9 +++++---- pkg/ext-proc/backend/provider.go | 10 ++++------ pkg/ext-proc/scheduling/scheduler.go | 1 - 3 files changed, 9 insertions(+), 11 deletions(-) diff --git a/pkg/ext-proc/backend/pod_metrics_refresher.go b/pkg/ext-proc/backend/pod_metrics_refresher.go index 9ffeb1d0..461cfa45 100644 --- a/pkg/ext-proc/backend/pod_metrics_refresher.go +++ b/pkg/ext-proc/backend/pod_metrics_refresher.go @@ -5,6 +5,7 @@ import ( "time" klog "k8s.io/klog/v2" + logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging" ) type PodMetricsRefresher struct { @@ -29,7 +30,7 @@ func NewPodMetricsRefresher(provider *Provider, pod Pod, interval, timeout time. func (r *PodMetricsRefresher) start() { go func() { - klog.V(2).Infof("Starting refresher for pod %v", r.pod) + klog.V(logutil.DEFAULT).InfoS("Starting refresher", "pod", r.pod) for { select { case <-r.done: @@ -39,7 +40,7 @@ func (r *PodMetricsRefresher) start() { err := r.refreshMetrics() if err != nil { - klog.Errorf("Failed to refresh metrics for pod %s: %v", r.pod, err) + klog.ErrorS(err, "Failed to refresh metrics", "pod", r.pod) } time.Sleep(r.interval) @@ -60,7 +61,7 @@ func (r *PodMetricsRefresher) refreshMetrics() error { return nil } - klog.V(4).Infof("Processing pod %v and metric %v", pod, existing.Metrics) + klog.V(logutil.DEBUG).InfoS("Refresh metrics", "pod", pod, "metrics", existing.Metrics) updated, err := r.provider.pmc.FetchMetrics(ctx, r.pod, existing) if err != nil { return err @@ -71,6 +72,6 @@ func (r *PodMetricsRefresher) refreshMetrics() error { } func (r *PodMetricsRefresher) stop() { - klog.V(2).Infof("Stopping refresher for pod %v", r.pod) + klog.V(logutil.DEFAULT).InfoS("Stopping refresher", "pod", r.pod) close(r.done) } diff --git a/pkg/ext-proc/backend/provider.go b/pkg/ext-proc/backend/provider.go index 674eae19..1827fceb 100644 --- a/pkg/ext-proc/backend/provider.go +++ b/pkg/ext-proc/backend/provider.go @@ -11,9 +11,7 @@ import ( ) const ( - // TODO: make it configurable. One idea is to provide a configuration singleton - // and put fields like refreshMetricsInterval in it. So far, we have to pass these - // fields across several layers. + // TODO: https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/336 metricsValidityPeriod = 5 * time.Second ) @@ -64,7 +62,7 @@ func (p *Provider) allPodMetrics(stale bool) []*PodMetrics { if !stale { if isPodMetricsStale(m) { // exclude stale metrics for scheduler - klog.V(4).Infof("Pod metrics for %s is stale, skipping", m.Pod) + klog.V(logutil.DEBUG).InfoS("Pod metrics is stale, skipping", "pod", m.Pod) } else { res = append(res, m) } @@ -84,7 +82,7 @@ func (p *Provider) allPodMetrics(stale bool) []*PodMetrics { func (p *Provider) UpdatePodMetrics(pod Pod, pm *PodMetrics) { pm.Metrics.UpdatedTime = time.Now() p.podMetrics.Store(pod, pm) - klog.V(4).Infof("Updated metrics for pod %s: %v", pod, pm.Metrics) + klog.V(logutil.DEBUG).InfoS("Updated metrics", "pod", pod, "metrics", pm.Metrics) } func (p *Provider) GetPodMetrics(pod Pod) (*PodMetrics, bool) { @@ -156,7 +154,7 @@ func (p *Provider) refreshPodsOnce(refreshMetricsInterval, refreshMetricsTimeout pod := k.(Pod) if _, ok := p.datastore.pods.Load(pod); !ok { p.podMetrics.Delete(pod) - if v, ok := p.podMetrics.LoadAndDelete(pod); ok { + if v, ok := p.podMetricsRefresher.LoadAndDelete(pod); ok { refresher := v.(*PodMetricsRefresher) refresher.stop() } diff --git a/pkg/ext-proc/scheduling/scheduler.go b/pkg/ext-proc/scheduling/scheduler.go index c3447c98..f20aecad 100644 --- a/pkg/ext-proc/scheduling/scheduler.go +++ b/pkg/ext-proc/scheduling/scheduler.go @@ -111,7 +111,6 @@ type PodMetricsProvider interface { // Schedule finds the target pod based on metrics and the requested lora adapter. func (s *Scheduler) Schedule(req *LLMRequest) (targetPod backend.Pod, err error) { - klog.V(logutil.VERBOSE).Infof("request: %v; metrics: %+v", req, s.podMetricsProvider.AllFreshPodMetrics()) pods, err := s.filter.Filter(req, s.podMetricsProvider.AllFreshPodMetrics()) if err != nil || len(pods) == 0 { return backend.Pod{}, fmt.Errorf(