Merge pull request #3671 from telepresenceio/thallgren/intermittent-u…

…nit-failure

Fix intermittently failing agent_injector unit test.

cmd/traffic/cmd/manager/mutator/agent_injector_test.go

@@ -345,7 +345,7 @@ func TestTrafficAgentConfigGenerator(t *testing.T) {
 	unnamedNumericPortUID := makeUID()
 	multiPortUID := makeUID()
 
-	clientset := fake.NewSimpleClientset(
+	clientset := fake.NewClientset(
 		&core.Service{
 			TypeMeta: meta.TypeMeta{
 				Kind:       "Service",
@@ -383,7 +383,7 @@ func TestTrafficAgentConfigGenerator(t *testing.T) {
 					Name:       "http",
 					Protocol:   "TCP",
 					Port:       80,
-					TargetPort: intstr.FromInt(8899),
+					TargetPort: intstr.FromInt32(8899),
 				}},
 				Selector: map[string]string{
 					"app": "numeric-port",
@@ -404,7 +404,7 @@ func TestTrafficAgentConfigGenerator(t *testing.T) {
 				Ports: []core.ServicePort{{
 					Protocol:   "TCP",
 					Port:       80,
-					TargetPort: intstr.FromInt(8899),
+					TargetPort: intstr.FromInt32(8899),
 				}},
 				Selector: map[string]string{
 					"app": "unnamed-numeric-port",
@@ -790,13 +790,12 @@ func TestTrafficAgentConfigGenerator(t *testing.T) {
 	ctx, err := managerutil.WithAgentImageRetriever(ctx, func(context.Context, string) error { return nil })
 	require.NoError(t, err)
 	cw := NewWatcher("")
+	cw.DisableRollouts()
 	cw.Start(ctx)
 	require.NoError(t, cw.StartWatchers(ctx))
 
 	for _, test := range tests {
 		test := test // pin it
-		pod := test.request
-		cw.Blacklist(pod.Name, pod.Namespace) // prevent rollout
 		agentmap.GeneratorConfigFunc = env.GeneratorConfig
 		t.Run(test.name, func(t *testing.T) {
 			runFunc(t, ctx, &test)
@@ -889,11 +888,11 @@ func TestTrafficAgentInjector(t *testing.T) {
 	}
 
 	podObjectMetaInjected := func(name string) meta.ObjectMeta {
-		meta := podObjectMeta(name)
-		meta.Labels[agentconfig.WorkloadNameLabel] = name
-		meta.Labels[agentconfig.WorkloadKindLabel] = "Deployment"
-		meta.Labels[agentconfig.WorkloadEnabledLabel] = "true"
-		return meta
+		pm := podObjectMeta(name)
+		pm.Labels[agentconfig.WorkloadNameLabel] = name
+		pm.Labels[agentconfig.WorkloadKindLabel] = "Deployment"
+		pm.Labels[agentconfig.WorkloadEnabledLabel] = "true"
+		return pm
 	}
 
 	podNamedPort := core.Pod{
@@ -960,7 +959,7 @@ func TestTrafficAgentInjector(t *testing.T) {
 		}
 	}
 
-	clientset := fake.NewSimpleClientset(
+	clientset := fake.NewClientset(
 		&core.Service{
 			TypeMeta: meta.TypeMeta{
 				Kind:       "Service",
@@ -999,7 +998,7 @@ func TestTrafficAgentInjector(t *testing.T) {
 				Ports: []core.ServicePort{{
 					Protocol:   "TCP",
 					Port:       80,
-					TargetPort: intstr.FromInt(8888),
+					TargetPort: intstr.FromInt32(8888),
 				}},
 				Selector: map[string]string{
 					"service": "numeric-port",
@@ -1840,9 +1839,9 @@ func TestTrafficAgentInjector(t *testing.T) {
 				agentmap.GeneratorConfigFunc = newEnv.GeneratorConfig
 			}
 			cw := NewWatcher("")
+			cw.DisableRollouts()
 			cw.Start(ctx)
 			require.NoError(t, cw.StartWatchers(ctx))
-			cw.Blacklist(test.pod.Name, test.pod.Namespace)
 
 			var actualPatch PatchOps
 			var actualErr error

cmd/traffic/cmd/manager/mutator/watcher.go

@@ -46,6 +46,7 @@ type Map interface {
 	Blacklist(podName, namespace string)
 	Whitelist(podName, namespace string)
 	IsBlacklisted(podName, namespace string) bool
+	DisableRollouts()
 
 	store(ctx context.Context, acx agentconfig.SidecarExt) error
 	remove(ctx context.Context, name, namespace string) error
@@ -93,6 +94,9 @@ func (e *entry) workload(ctx context.Context) (agentconfig.SidecarExt, k8sapi.Wo
 // isRolloutNeeded checks if the agent's entry in telepresence-agents matches the actual state of the
 // pods. If it does, then there's no reason to trigger a rollout.
 func (c *configWatcher) isRolloutNeeded(ctx context.Context, wl k8sapi.Workload, ac *agentconfig.Sidecar) bool {
+	if c.rolloutDisabled {
+		return false
+	}
 	podMeta := wl.GetPodTemplate().GetObjectMeta()
 	if wl.GetDeletionTimestamp() != nil {
 		return false
@@ -428,6 +432,7 @@ type configWatcher struct {
 	nsLocks         *xsync.MapOf[string, *sync.RWMutex]
 	blacklistedPods *xsync.MapOf[string, time.Time]
 	startedAt       time.Time
+	rolloutDisabled bool
 
 	cms []cache.SharedIndexInformer
 	svs []cache.SharedIndexInformer
@@ -451,6 +456,10 @@ func (c *configWatcher) Whitelist(podName, namespace string) {
 	c.blacklistedPods.Delete(podName + "." + namespace)
 }
 
+func (c *configWatcher) DisableRollouts() {
+	c.rolloutDisabled = true
+}
+
 func (c *configWatcher) IsBlacklisted(podName, namespace string) bool {
 	_, ok := c.blacklistedPods.Load(podName + "." + namespace)
 	return ok

pkg/agentmap/discorvery.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"sort"
 
 	core "k8s.io/api/core/v1"
 	k8sErrors "k8s.io/apimachinery/pkg/api/errors"
@@ -217,6 +218,10 @@ func findServicesSelecting(ctx context.Context, namespace string, lbs labels.Lab
 			}
 		}
 	}
+	// Ensure predictable order of found services
+	sort.Slice(ms, func(i, j int) bool {
+		return ms[i].GetName() < ms[j].GetName()
+	})
 	dlog.Debugf(ctx, "Scanned %d services in namespace %s and found that %s selects labels %v", scanned, namespace, objectsStringer(ms), lbs)
 	return ms, nil
 }