Repository: kumahq/kuma
Pull request #1262 (2020-12-02 12:18:25)

fix(kuma-cp) upsert with retry on conflict (bp #1236)
This is an automated backport of pull request #1236 done by Mergify
---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.io/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.io/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.io/
</details>



Patch:
diff --git a/pkg/config/app/kuma-cp/kuma-cp.defaults.yaml b/pkg/config/app/kuma-cp/kuma-cp.defaults.yaml
index de04258ab..40c246c59 100644
--- a/pkg/config/app/kuma-cp/kuma-cp.defaults.yaml
+++ b/pkg/config/app/kuma-cp/kuma-cp.defaults.yaml
@@ -52,9 +52,16 @@ store:
   # Cache for read only operations. This cache is local to the instance of the control plane.
   cache:
     # If true then cache is enabled
-    enabled: true
+    enabled: true # ENV: KUMA_STORE_CACHE_ENABLED
     # Expiration time for elements in cache.
-    expirationTime: 1s
+    expirationTime: 1s # ENV: KUMA_STORE_CACHE_EXPIRATION_TIME
+
+  # Upsert configuration
+  upsert:
+    # Base time for exponential backoff on upsert (get and update) operations when retry is enabled
+    conflictRetryBaseBackoff: 100ms # ENV: KUMA_STORE_UPSERT_CONFLICT_RETRY_BASE_BACKOFF
+    # Max retries on upsert (get and update) operation when retry is enabled
+    conflictRetryMaxTimes: 5 # ENV: KUMA_STORE_UPSERT_CONFLICT_RETRY_MAX_TIMES
 
 # Configuration of Bootstrap Server, which provides bootstrap config to Dataplanes
 bootstrapServer:
diff --git a/pkg/config/core/resources/store/config.go b/pkg/config/core/resources/store/config.go
index 396f1284c..ce68df483 100644
--- a/pkg/config/core/resources/store/config.go
+++ b/pkg/config/core/resources/store/config.go
@@ -30,6 +30,8 @@ type StoreConfig struct {
 	Kubernetes *k8s.KubernetesStoreConfig `yaml:"kubernetes"`
 	// Cache configuration
 	Cache CacheStoreConfig `yaml:"cache"`
+	// Upsert configuration
+	Upsert UpsertConfig `yaml:"upsert"`
 }
 
 func DefaultStoreConfig() *StoreConfig {
@@ -38,6 +40,7 @@ func DefaultStoreConfig() *StoreConfig {
 		Postgres:   postgres.DefaultPostgresStoreConfig(),
 		Kubernetes: k8s.DefaultKubernetesStoreConfig(),
 		Cache:      DefaultCacheStoreConfig(),
+		Upsert:     DefaultUpsertConfig(),
 	}
 }
 
@@ -89,3 +92,29 @@ func DefaultCacheStoreConfig() CacheStoreConfig {
 		ExpirationTime: time.Second,
 	}
 }
+
+func DefaultUpsertConfig() UpsertConfig {
+	return UpsertConfig{
+		ConflictRetryBaseBackoff: 100 * time.Millisecond,
+		ConflictRetryMaxTimes:    5,
+	}
+}
+
+type UpsertConfig struct {
+	// Base time for exponential backoff on upsert (get and update) operations when retry is enabled
+	ConflictRetryBaseBackoff time.Duration `yaml:"conflictRetryBaseBackoff" envconfig:"kuma_store_upsert_conflict_retry_base_backoff"`
+	// Max retries on upsert (get and update) operation when retry is enabled
+	ConflictRetryMaxTimes uint `yaml:"conflictRetryMaxTimes" envconfig:"kuma_store_upsert_conflict_retry_max_times"`
+}
+
+func (u *UpsertConfig) Sanitize() {
+}
+
+func (u *UpsertConfig) Validate() error {
+	if u.ConflictRetryBaseBackoff < 0 {
+		return errors.New("RetryBaseBackoff cannot be lower than 0")
+	}
+	return nil
+}
+
+var _ config.Config = &UpsertConfig{}
diff --git a/pkg/core/resources/manager/manager.go b/pkg/core/resources/manager/manager.go
index a8474a3e5..c17b7b11b 100644
--- a/pkg/core/resources/manager/manager.go
+++ b/pkg/core/resources/manager/manager.go
@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/sethvargo/go-retry"
+
 	"github.com/kumahq/kuma/pkg/core"
 	core_mesh "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	"github.com/kumahq/kuma/pkg/core/resources/model"
@@ -95,22 +97,66 @@ func (r *resourcesManager) Update(ctx context.Context, resource model.Resource,
 	return r.Store.Update(ctx, resource, append(fs, store.ModifiedAt(time.Now()))...)
 }
 
-func Upsert(manager ResourceManager, key model.ResourceKey, resource model.Resource, fn func(resource model.Resource)) error {
-	create := false
-	err := manager.Get(context.Background(), resource, store.GetBy(key))
-	if err != nil {
-		if store.IsResourceNotFound(err) {
-			create = true
+type ConflictRetry struct {
+	BaseBackoff time.Duration
+	MaxTimes    uint
+}
+
+type UpsertOpts struct {
+	ConflictRetry ConflictRetry
+}
+
+type UpsertFunc func(opts *UpsertOpts)
+
+func WithConflictRetry(baseBackoff time.Duration, maxTimes uint) UpsertFunc {
+	return func(opts *UpsertOpts) {
+		opts.ConflictRetry.BaseBackoff = baseBackoff
+		opts.ConflictRetry.MaxTimes = maxTimes
+	}
+}
+
+func NewUpsertOpts(fs ...UpsertFunc) UpsertOpts {
+	opts := UpsertOpts{}
+	for _, f := range fs {
+		f(&opts)
+	}
+	return opts
+}
+
+func Upsert(manager ResourceManager, key model.ResourceKey, resource model.Resource, fn func(resource model.Resource), fs ...UpsertFunc) error {
+	upsert := func() error {
+		create := false
+		err := manager.Get(context.Background(), resource, store.GetBy(key))
+		if err != nil {
+			if store.IsResourceNotFound(err) {
+				create = true
+			} else {
+				return err
+			}
+		}
+		fn(resource)
+		if create {
+			return manager.Create(context.Background(), resource, store.CreateBy(key))
 		} else {
-			return err
+			return manager.Update(context.Background(), resource)
 		}
 	}
-	fn(resource)
-	if create {
-		return manager.Create(context.Background(), resource, store.CreateBy(key))
-	} else {
-		return manager.Update(context.Background(), resource)
+
+	opts := NewUpsertOpts(fs...)
+	if opts.ConflictRetry.BaseBackoff <= 0 || opts.ConflictRetry.MaxTimes == 0 {
+		return upsert()
 	}
+	backoff, _ := retry.NewExponential(opts.ConflictRetry.BaseBackoff) // we can ignore error because RetryBaseBackoff > 0
+	backoff = retry.WithMaxRetries(uint64(opts.ConflictRetry.MaxTimes), backoff)
+	return retry.Do(context.Background(), backoff, func(ctx context.Context) error {
+		resource.SetMeta(nil)
+		resource.GetSpec().Reset()
+		err := upsert()
+		if store.IsResourceConflict(err) {
+			return retry.RetryableError(err)
+		}
+		return err
+	})
 }
 
 type MeshNotFoundError struct {
diff --git a/pkg/core/resources/store/store.go b/pkg/core/resources/store/store.go
index f74e92b3a..b7ee2f3d0 100644
--- a/pkg/core/resources/store/store.go
+++ b/pkg/core/resources/store/store.go
@@ -124,6 +124,10 @@ func ErrorResourceConflict(rt model.ResourceType, name, mesh string) error {
 	return fmt.Errorf("Resource conflict: type=%q name=%q mesh=%q", rt, name, mesh)
 }
 
+func IsResourceConflict(err error) bool {
+	return err != nil && strings.HasPrefix(err.Error(), "Resource conflict")
+}
+
 func ErrorResourcePreconditionFailed(rt model.ResourceType, name, mesh string) error {
 	return fmt.Errorf("Resource precondition failed: type=%q name=%q mesh=%q", rt, name, mesh)
 }
diff --git a/pkg/kds/server/components.go b/pkg/kds/server/components.go
index dfb76c223..5c5aa0e1b 100644
--- a/pkg/kds/server/components.go
+++ b/pkg/kds/server/components.go
@@ -51,6 +51,7 @@ func DefaultStatusTracker(rt core_runtime.Runtime, log logr.Logger) StatusTracke
 			func() *time.Ticker {
 				return time.NewTicker(rt.Config().Multizone.Global.KDS.ZoneInsightFlushInterval)
 			},
+			rt.Config().Multizone.Global.KDS.ZoneInsightFlushInterval/10,
 			NewDataplaneInsightStore(rt.ResourceManager()),
 			l)
 	}, log)
diff --git a/pkg/kds/server/status_sink.go b/pkg/kds/server/status_sink.go
index 13c766e94..a77ed748a 100644
--- a/pkg/kds/server/status_sink.go
+++ b/pkg/kds/server/status_sink.go
@@ -4,6 +4,7 @@ import (
 	"time"
 
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
+	"github.com/kumahq/kuma/pkg/core/resources/store"
 
 	"github.com/go-logr/logr"
 
@@ -26,18 +27,26 @@ type ZoneInsightStore interface {
 func NewZoneInsightSink(
 	accessor StatusAccessor,
 	newTicker func() *time.Ticker,
+	flushBackoff time.Duration,
 	store ZoneInsightStore,
 	log logr.Logger) ZoneInsightSink {
-	return &zoneInsightSink{newTicker, accessor, store, log}
+	return &zoneInsightSink{
+		newTicker:    newTicker,
+		flushBackoff: flushBackoff,
+		accessor:     accessor,
+		store:        store,
+		log:          log,
+	}
 }
 
 var _ ZoneInsightSink = &zoneInsightSink{}
 
 type zoneInsightSink struct {
-	newTicker func() *time.Ticker
-	accessor  StatusAccessor
-	store     ZoneInsightStore
-	log       logr.Logger
+	newTicker    func() *time.Ticker
+	flushBackoff time.Duration
+	accessor     StatusAccessor
+	store        ZoneInsightStore
+	log          logr.Logger
 }
 
 func (s *zoneInsightSink) Start(stop <-chan struct{}) {
@@ -53,9 +62,13 @@ func (s *zoneInsightSink) Start(stop <-chan struct{}) {
 		}
 		copy := proto.Clone(currentState).(*system_proto.KDSSubscription)
 		if err := s.store.Upsert(zone, copy); err != nil {
-			s.log.Error(err, "failed to flush zone status", "zone", zone)
+			if store.IsResourceConflict(err) {
+				s.log.V(1).Info("failed to flush ZoneInsight because it was updated in other place. Will retry in the next tick", "zone", zone)
+			} else {
+				s.log.Error(err, "failed to flush zone status", "zone", zone)
+			}
 		} else {
-			s.log.V(1).Info("saved zone status", "zone", zone, "subscription", currentState)
+			s.log.V(1).Info("ZoneInsight saved", "zone", zone, "subscription", currentState)
 			lastStoredState = currentState
 		}
 	}
@@ -64,6 +77,7 @@ func (s *zoneInsightSink) Start(stop <-chan struct{}) {
 		select {
 		case <-ticker.C:
 			flush()
+			time.Sleep(s.flushBackoff)
 		case <-stop:
 			flush()
 			return
diff --git a/pkg/plugins/runtime/k8s/webhooks/injector/injector.go b/pkg/plugins/runtime/k8s/webhooks/injector/injector.go
index 0c0f74fa4..497fddba2 100644
--- a/pkg/plugins/runtime/k8s/webhooks/injector/injector.go
+++ b/pkg/plugins/runtime/k8s/webhooks/injector/injector.go
@@ -78,7 +78,7 @@ func (i *KumaInjector) InjectKuma(pod *kube_core.Pod) error {
 		log.V(1).Info("skip injecting Kuma", "name", pod.Name, "namespace", pod.Namespace)
 		return nil
 	}
-	log.Info("injecting Kuma", "name", pod.Name, "namespace", pod.Namespace)
+	log.Info("injecting Kuma", "name", pod.GenerateName, "namespace", pod.Namespace)
 	// sidecar container
 	if pod.Spec.Containers == nil {
 		pod.Spec.Containers = []kube_core.Container{}
diff --git a/pkg/sds/server/reconciler.go b/pkg/sds/server/reconciler.go
index 39672be2a..1cbb25f1e 100644
--- a/pkg/sds/server/reconciler.go
+++ b/pkg/sds/server/reconciler.go
@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/kumahq/kuma/pkg/config/core/resources/store"
 	"github.com/kumahq/kuma/pkg/xds/envoy/tls"
 
 	"github.com/pkg/errors"
@@ -43,6 +44,7 @@ type DataplaneReconciler struct {
 	meshCaProvider     sds_provider.SecretProvider
 	identityProvider   sds_provider.SecretProvider
 	cache              envoy_cache.SnapshotCache
+	upsertConfig       store.UpsertConfig
 
 	sync.RWMutex
 	certGenerated int
@@ -177,24 +179,10 @@ func (d *DataplaneReconciler) updateInsights(dataplaneId core_model.ResourceKey,
 		return err
 	}
 
-	create := false
-	dataplaneInsight := &mesh_core.DataplaneInsightResource{}
-	err = d.resManager.Get(context.Background(), dataplaneInsight, core_store.GetBy(dataplaneId))
-	if err != nil {
-		if core_store.IsResourceNotFound(err) {
-			create = true
-		} else {
-			return err
+	return core_manager.Upsert(d.resManager, dataplaneId, &mesh_core.DataplaneInsightResource{}, func(resource core_model.Resource) {
+		insight := resource.(*mesh_core.DataplaneInsightResource)
+		if err := insight.Spec.UpdateCert(core.Now(), cert.NotAfter); err != nil {
+			sdsServerLog.Error(err, "could not update the certificate", "dataplaneId", dataplaneId)
 		}
-	}
-
-	if err := dataplaneInsight.Spec.UpdateCert(core.Now(), cert.NotAfter); err != nil {
-		return err
-	}
-
-	if create {
-		return d.resManager.Create(context.Background(), dataplaneInsight, core_store.CreateBy(dataplaneId))
-	} else {
-		return d.resManager.Update(context.Background(), dataplaneInsight)
-	}
+	}, core_manager.WithConflictRetry(d.upsertConfig.ConflictRetryBaseBackoff, d.upsertConfig.ConflictRetryMaxTimes)) // retry because DataplaneInsight could be updated from other parts of the code
 }
diff --git a/pkg/sds/server/server.go b/pkg/sds/server/server.go
index 24fecb728..88b29e4ba 100644
--- a/pkg/sds/server/server.go
+++ b/pkg/sds/server/server.go
@@ -47,6 +47,7 @@ func RegisterSDS(rt core_runtime.Runtime, server *grpc.Server) error {
 		meshCaProvider:     caProvider,
 		identityProvider:   identityProvider,
 		cache:              cache,
+		upsertConfig:       rt.Config().Store.Upsert,
 	}
 	certGenerationsMetric := prometheus.NewGaugeFunc(prometheus.GaugeOpts{
 		Help: "Number of generated certificates",
diff --git a/pkg/xds/server/components.go b/pkg/xds/server/components.go
index 4f041221e..57465ef61 100644
--- a/pkg/xds/server/components.go
+++ b/pkg/xds/server/components.go
@@ -369,7 +369,9 @@ func DefaultDataplaneStatusTracker(rt core_runtime.Runtime) (DataplaneStatusTrac
 			func() *time.Ticker {
 				return time.NewTicker(rt.Config().XdsServer.DataplaneStatusFlushInterval)
 			},
-			NewDataplaneInsightStore(rt.ResourceManager()))
+			rt.Config().XdsServer.DataplaneStatusFlushInterval/10,
+			NewDataplaneInsightStore(rt.ResourceManager()),
+		)
 	})
 	return tracker, nil
 }
diff --git a/pkg/xds/server/dataplane_status_sink.go b/pkg/xds/server/dataplane_status_sink.go
index 9f737b938..c29564a35 100644
--- a/pkg/xds/server/dataplane_status_sink.go
+++ b/pkg/xds/server/dataplane_status_sink.go
@@ -9,6 +9,7 @@ import (
 	mesh_core "github.com/kumahq/kuma/pkg/core/resources/apis/mesh"
 	"github.com/kumahq/kuma/pkg/core/resources/manager"
 	core_model "github.com/kumahq/kuma/pkg/core/resources/model"
+	"github.com/kumahq/kuma/pkg/core/resources/store"
 )
 
 type DataplaneInsightSink interface {
@@ -22,16 +23,23 @@ type DataplaneInsightStore interface {
 func NewDataplaneInsightSink(
 	accessor SubscriptionStatusAccessor,
 	newTicker func() *time.Ticker,
+	flushBackoff time.Duration,
 	store DataplaneInsightStore) DataplaneInsightSink {
-	return &dataplaneInsightSink{newTicker, accessor, store}
+	return &dataplaneInsightSink{
+		newTicker:    newTicker,
+		accessor:     accessor,
+		flushBackoff: flushBackoff,
+		store:        store,
+	}
 }
 
 var _ DataplaneInsightSink = &dataplaneInsightSink{}
 
 type dataplaneInsightSink struct {
-	newTicker func() *time.Ticker
-	accessor  SubscriptionStatusAccessor
-	store     DataplaneInsightStore
+	newTicker    func() *time.Ticker
+	accessor     SubscriptionStatusAccessor
+	store        DataplaneInsightStore
+	flushBackoff time.Duration
 }
 
 func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
@@ -47,7 +55,8 @@ func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
 		}
 		copy := proto.Clone(currentState).(*mesh_proto.DiscoverySubscription)
 		if err := s.store.Upsert(dataplaneId, copy); err != nil {
-			if closing {
+			switch {
+			case closing:
 				// When XDS stream is closed, Dataplane Status Tracker executes OnStreamClose which closes stop channel
 				// The problem is that close() does not wait for this sink to do it's final work
 				// In the meantime Dataplane Lifecycle executes OnStreamClose which can remove Dataplane entity (and Insights due to ownership). Therefore both scenarios can happen:
@@ -56,11 +65,13 @@ func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
 				// We could build a synchronous mechanism that waits for Sink to be stopped before moving on to next Callbacks, but this is potentially dangerous
 				// that we could block waiting for storage instead of executing next callbacks.
 				xdsServerLog.V(1).Info("failed to flush Dataplane status on stream close. It can happen when Dataplane is deleted at the same time", "dataplaneid", dataplaneId, "err", err)
-			} else {
-				xdsServerLog.Error(err, "failed to flush Dataplane status", "dataplaneid", dataplaneId)
+			case store.IsResourceConflict(err):
+				xdsServerLog.V(1).Info("failed to flush DataplaneInsight because it was updated in other place. Will retry in the next tick", "dataplaneid", dataplaneId)
+			default:
+				xdsServerLog.Error(err, "failed to flush DataplaneInsight", "dataplaneid", dataplaneId)
 			}
 		} else {
-			xdsServerLog.V(1).Info("saved Dataplane status", "dataplaneid", dataplaneId, "subscription", currentState)
+			xdsServerLog.V(1).Info("DataplaneInsight saved", "dataplaneid", dataplaneId, "subscription", currentState)
 			lastStoredState = currentState
 		}
 	}
@@ -69,6 +80,9 @@ func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
 		select {
 		case <-ticker.C:
 			flush(false)
+			// On Kubernetes, because of the cache subsequent Get, Update requests can fail, because the cache is not strongly consistent.
+			// We handle the Resource Conflict logging on V1, but we can try to avoid the situation with backoff
+			time.Sleep(s.flushBackoff)
 		case <-stop:
 			flush(true)
 			return
@@ -77,7 +91,9 @@ func (s *dataplaneInsightSink) Start(stop <-chan struct{}) {
 }
 
 func NewDataplaneInsightStore(resManager manager.ResourceManager) DataplaneInsightStore {
-	return &dataplaneInsightStore{resManager}
+	return &dataplaneInsightStore{
+		resManager: resManager,
+	}
 }
 
 var _ DataplaneInsightStore = &dataplaneInsightStore{}
@@ -87,8 +103,8 @@ type dataplaneInsightStore struct {
 }
 
 func (s *dataplaneInsightStore) Upsert(dataplaneId core_model.ResourceKey, subscription *mesh_proto.DiscoverySubscription) error {
-	insight := &mesh_core.DataplaneInsightResource{}
-	return manager.Upsert(s.resManager, dataplaneId, insight, func(resource core_model.Resource) {
+	return manager.Upsert(s.resManager, dataplaneId, &mesh_core.DataplaneInsightResource{}, func(resource core_model.Resource) {
+		insight := resource.(*mesh_core.DataplaneInsightResource)
 		insight.Spec.UpdateSubscription(subscription)
 	})
 }


Test patch:
diff --git a/pkg/api-server/config_ws_test.go b/pkg/api-server/config_ws_test.go
index 6ce1751f7..a8817b99d 100644
--- a/pkg/api-server/config_ws_test.go
+++ b/pkg/api-server/config_ws_test.go
@@ -244,6 +244,10 @@ var _ = Describe("Config WS", func() {
               "enabled": true,
               "expirationTime": "1s"
             },
+            "upsert": {
+              "conflictRetryBaseBackoff": "100ms",
+              "conflictRetryMaxTimes": 5
+            },
             "type": "memory"
           },
           "xdsServer": {
diff --git a/pkg/config/loader_test.go b/pkg/config/loader_test.go
index 023152747..1026ec7db 100644
--- a/pkg/config/loader_test.go
+++ b/pkg/config/loader_test.go
@@ -107,6 +107,9 @@ var _ = Describe("Config loader", func() {
 			Expect(cfg.Store.Cache.Enabled).To(BeFalse())
 			Expect(cfg.Store.Cache.ExpirationTime).To(Equal(3 * time.Second))
 
+			Expect(cfg.Store.Upsert.ConflictRetryBaseBackoff).To(Equal(4 * time.Second))
+			Expect(cfg.Store.Upsert.ConflictRetryMaxTimes).To(Equal(uint(10)))
+
 			Expect(cfg.Store.Postgres.TLS.Mode).To(Equal(postgres.VerifyFull))
 			Expect(cfg.Store.Postgres.TLS.CertPath).To(Equal("/path/to/cert"))
 			Expect(cfg.Store.Postgres.TLS.KeyPath).To(Equal("/path/to/key"))
@@ -238,6 +241,9 @@ store:
   cache:
     enabled: false
     expirationTime: 3s
+  upsert:
+    conflictRetryBaseBackoff: 4s
+    conflictRetryMaxTimes: 10
 bootstrapServer:
   params:
     adminPort: 1234
@@ -405,6 +411,8 @@ sdsServer:
 				"KUMA_STORE_KUBERNETES_SYSTEM_NAMESPACE":                                                   "test-namespace",
 				"KUMA_STORE_CACHE_ENABLED":                                                                 "false",
 				"KUMA_STORE_CACHE_EXPIRATION_TIME":                                                         "3s",
+				"KUMA_STORE_UPSERT_CONFLICT_RETRY_BASE_BACKOFF":                                            "4s",
+				"KUMA_STORE_UPSERT_CONFLICT_RETRY_MAX_TIMES":                                               "10",
 				"KUMA_API_SERVER_READ_ONLY":                                                                "true",
 				"KUMA_API_SERVER_HTTP_PORT":                                                                "15681",
 				"KUMA_API_SERVER_HTTP_INTERFACE":                                                           "192.168.0.1",
diff --git a/pkg/xds/server/dataplane_status_sink_test.go b/pkg/xds/server/dataplane_status_sink_test.go
index 0f6365dc0..b1973030f 100644
--- a/pkg/xds/server/dataplane_status_sink_test.go
+++ b/pkg/xds/server/dataplane_status_sink_test.go
@@ -57,7 +57,7 @@ var _ = Describe("DataplaneInsightSink", func() {
 			var latestUpsert *DataplaneInsightUpsert
 
 			// given
-			sink := NewDataplaneInsightSink(accessor, func() *time.Ticker { return ticker }, recorder)
+			sink := NewDataplaneInsightSink(accessor, func() *time.Ticker { return ticker }, 1*time.Millisecond, recorder)
 			go sink.Start(stop)
 
 			// when