Merge pull request #4035 from telepresenceio/thallgren/injector-race

Fix race condition between agent injector and service watcher

Author: thallgren

cmd/traffic/cmd/manager/mutator/agent_injector.go

@@ -146,12 +146,19 @@ func (a *agentInjector) Inject(ctx context.Context, req *admission.AdmissionRequ
 			// Not an error. It just means that the pod is not eligible for intercepts.
 			return nil, nil
 		}
-		sc = a.agentConfigs.Get(wl.GetName(), wl.GetNamespace())
-		switch {
-		case sc == nil:
-			clog.Tracef(ctx, "Skipping %s (no agent config)", wl)
-			return nil, nil
-		case sc.Manual:
+		if ia == "enabled" {
+			sc, err = a.agentConfigs.GetOrGenerate(ctx, wl)
+			if err != nil {
+				return nil, fmt.Errorf("unable to get or generate agent config for workload %s.%s: %w", wl.GetNamespace(), wl.GetName(), err)
+			}
+		} else {
+			sc = a.agentConfigs.Get(wl.GetName(), wl.GetNamespace())
+			if sc == nil {
+				clog.Tracef(ctx, "Skipping %s (no agent config)", wl)
+				return nil, nil
+			}
+		}
+		if sc.Manual {
 			clog.Tracef(ctx, "Skipping webhook where agent is manually injected %s", wl.GetNamespace())
 			return nil, nil
 		}

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

@@ -29,6 +29,7 @@ import (
 
 type Map interface {
 	Get(string, string) *agentconfig.Sidecar
+	GetOrGenerate(context.Context, k8sapi.Workload) (*agentconfig.Sidecar, error)
 	Store(*agentconfig.Sidecar)
 	Start(context.Context)
 	StartWatchers(context.Context) error
@@ -340,6 +341,27 @@ func (c *configWatcher) Get(key, ns string) (ac *agentconfig.Sidecar) {
 	return ac
 }
 
+// GetOrGenerate returns the Sidecar configuration for the given workload. If no configuration is found, it blocks the entry from access while
+// it generates a new one using the given generator.
+func (c *configWatcher) GetOrGenerate(ctx context.Context, wl k8sapi.Workload) (ac *agentconfig.Sidecar, err error) {
+	c.agentConfigs.Compute(wl.GetNamespace(), func(scMap map[string]*agentconfig.Sidecar, loaded bool) (map[string]*agentconfig.Sidecar, xsync.ComputeOp) {
+		if loaded {
+			ac = scMap[wl.GetName()]
+		} else {
+			var gc *agentmap.GeneratorConfig
+			gc, err = managerutil.GetEnv(ctx).GeneratorConfig(managerutil.GetAgentImage(ctx))
+			if err == nil {
+				ac, err = gc.Generate(ctx, wl, nil)
+				if err == nil {
+					scMap[wl.GetName()] = ac
+				}
+			}
+		}
+		return nil, xsync.CancelOp
+	})
+	return ac, err
+}
+
 func whereWeWatch(ns string) string {
 	if ns == "" {
 		return "cluster wide"

integration_test/cidr_conflict_test.go

@@ -69,7 +69,7 @@ func (s *cidrConflictSuite) SetupSuite() {
 	ctx := s.Context()
 	s.TelepresenceConnect(ctx)
 	st := itest.TelepresenceStatusOk(ctx)
-	itest.TelepresenceQuitOk(ctx)
+	itest.TelepresenceQuit(ctx)
 	s.subnets = st.RootDaemon.Subnets
 	if len(s.subnets) < 2 {
 		s.T().Skip("Test cannot run unless client maps at least two subnets")
@@ -94,7 +94,7 @@ func (s *cidrConflictSuite) Test_AutoConflictResolution() {
 	ctx := s.Context()
 	s.TelepresenceConnect(ctx)
 	st := itest.TelepresenceStatusOk(ctx)
-	defer itest.TelepresenceQuitOk(ctx)
+	defer itest.TelepresenceQuit(ctx)
 	sns := st.RootDaemon.Subnets
 	rq := s.Require()
 	rq.Less(len(sns), len(s.subnets), "pod and service subnets should be combined into one virtual subnet")

integration_test/itest/status.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/telepresenceio/clog"
+	"github.com/telepresenceio/telepresence/v2/pkg/client"
 	"github.com/telepresenceio/telepresence/v2/pkg/client/cli/cmd"
 )
 
@@ -41,6 +42,9 @@ func TelepresenceStatus(ctx context.Context, args ...string) (*StatusResponse, e
 		return nil, jErr
 	}
 	if cd := status.ContainerizedDaemon; cd != nil {
+		if cd.RoutingSnake == nil {
+			cd.RoutingSnake = &client.RoutingSnake{}
+		}
 		status.UserDaemon = cd.UserDaemonStatus
 		status.RootDaemon = &cmd.RootDaemonStatus{
 			Running:      cd.Running,
@@ -51,7 +55,7 @@ func TelepresenceStatus(ctx context.Context, args ...string) (*StatusResponse, e
 			PortMappings: cd.PortMappings,
 		}
 	} else if status.RootDaemon == nil {
-		status.RootDaemon = &cmd.RootDaemonStatus{}
+		status.RootDaemon = &cmd.RootDaemonStatus{RoutingSnake: &client.RoutingSnake{}}
 	}
 	return &status, nil
 }

pkg/client/cli/connect/connector.go

@@ -93,7 +93,7 @@ func quitHostConnector(ctx context.Context) {
 	var errs error
 	rootWillContinue := false
 	if err != nil {
-		if !errors.Is(err, fs.ErrNotExist) {
+		if !(errors.Is(err, fs.ErrNotExist) || errors.Is(err, daemon.ErrNoUserDaemon)) {
 			errs = errors.Join(errs, err)
 		}
 	} else {