removed time.sleep and using ticker instead (#648)

* removed time.sleep and using ticker instead

Signed-off-by: Nir Rozenbaum <[email protected]>

* move ticker creation outside of go routine.
make sure refresh internal is valid at the ticker creation time

Signed-off-by: Nir Rozenbaum <[email protected]>

* add DefaultRefreshPrometheusMetricsInterval for test purposes.
once ticker was introduced instead of sleep, having 0 as the refresh internal is not valid.

Signed-off-by: Nir Rozenbaum <[email protected]>

* wait in test until metrics are available before running tests that rely on the values.

up until now, the metrics go routine ran in tests with time.Sleep(0), which means metrics were avaiable immediately.
while in tests in might be acceptable to wait few seconds using sleep, in the actual code (not tests) it's a bad practice to use
sleep which was replaced with a ticker (to perform periodic task in an endless loop).

Signed-off-by: Nir Rozenbaum <[email protected]>

---------

Signed-off-by: Nir Rozenbaum <[email protected]>

Author: nirrozenbaum

pkg/epp/backend/metrics/logger.go

@@ -32,6 +32,7 @@ const (
 	// Note currently the EPP treats stale metrics same as fresh.
 	// TODO: https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/336
 	metricsValidityPeriod = 5 * time.Second
+	debugPrintInterval    = 5 * time.Second
 )
 
 type Datastore interface {
@@ -46,16 +47,15 @@ type Datastore interface {
 // enabled; 2) flushes Prometheus metrics about the backend servers.
 func StartMetricsLogger(ctx context.Context, datastore Datastore, refreshPrometheusMetricsInterval time.Duration) {
 	logger := log.FromContext(ctx)
-
-	// Periodically flush prometheus metrics for inference pool
+	ticker := time.NewTicker(refreshPrometheusMetricsInterval)
 	go func() {
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ctx.Done():
 				logger.V(logutil.DEFAULT).Info("Shutting down prometheus metrics thread")
 				return
-			default:
-				time.Sleep(refreshPrometheusMetricsInterval)
+			case <-ticker.C: // Periodically flush prometheus metrics for inference pool
 				flushPrometheusMetricsOnce(logger, datastore)
 			}
 		}
@@ -64,13 +64,14 @@ func StartMetricsLogger(ctx context.Context, datastore Datastore, refreshPrometh
 	// Periodically print out the pods and metrics for DEBUGGING.
 	if logger := logger.V(logutil.DEBUG); logger.Enabled() {
 		go func() {
+			ticker := time.NewTicker(debugPrintInterval)
+			defer ticker.Stop()
 			for {
 				select {
 				case <-ctx.Done():
 					logger.V(logutil.DEFAULT).Info("Shutting down metrics logger thread")
 					return
-				default:
-					time.Sleep(5 * time.Second)
+				case <-ticker.C:
 					podsWithFreshMetrics := datastore.PodList(func(pm PodMetrics) bool {
 						return time.Since(pm.GetMetrics().UpdateTime) <= metricsValidityPeriod
 					})

pkg/epp/backend/metrics/pod_metrics.go

@@ -84,21 +84,19 @@ func (pm *podMetrics) startRefreshLoop() {
 	pm.once.Do(func() {
 		go func() {
 			pm.logger.V(logutil.DEFAULT).Info("Starting refresher", "pod", pm.GetPod())
+			ticker := time.NewTicker(pm.interval)
+			defer ticker.Stop()
 			for {
 				select {
 				case <-pm.done:
 					return
 				case <-pm.parentCtx.Done():
 					return
-				default:
+				case <-ticker.C: // refresh metrics periodically
+					if err := pm.refreshMetrics(); err != nil {
+						pm.logger.V(logutil.TRACE).Error(err, "Failed to refresh metrics", "pod", pm.GetPod())
+					}
 				}
-
-				err := pm.refreshMetrics()
-				if err != nil {
-					pm.logger.V(logutil.TRACE).Error(err, "Failed to refresh metrics", "pod", pm.GetPod())
-				}
-
-				time.Sleep(pm.interval)
 			}
 		}()
 	})

pkg/epp/server/runserver.go

@@ -76,6 +76,7 @@ func NewDefaultExtProcServerRunner() *ExtProcServerRunner {
 		PoolName:                                 DefaultPoolName,
 		PoolNamespace:                            DefaultPoolNamespace,
 		SecureServing:                            DefaultSecureServing,
+		RefreshPrometheusMetricsInterval:         DefaultRefreshPrometheusMetricsInterval,
 		// Datastore can be assigned later.
 	}
 }

test/integration/epp/hermetic_test.go

@@ -1548,6 +1548,8 @@ func setUpHermeticServer(t *testing.T, podAndMetrics map[backendmetrics.Pod]*bac
 		}
 	}()
 
+	time.Sleep(serverRunner.RefreshPrometheusMetricsInterval) // wait for metrics to get available before running tests that rely on these metrics
+
 	// check if all pods are synced to datastore
 	assert.EventuallyWithT(t, func(t *assert.CollectT) {
 		assert.Len(t, serverRunner.Datastore.PodGetAll(), len(podAndMetrics), "Datastore not synced")