atunnel: ingress CONNECT support

cmd/ateapi/internal/controlapi/syncer_test.go

@@ -968,11 +968,12 @@ func TestReleaseActorOnDeadWorker_StatusTransitions(t *testing.T) {
 type conflictStore struct {
 	store.Interface
 	conflictTriggered atomic.Bool
+	shouldInject      func(worker *ateapipb.Worker) bool
 	onUpdate          func(ctx context.Context, worker *ateapipb.Worker)
 }
 
 func (c *conflictStore) UpdateWorker(ctx context.Context, worker *ateapipb.Worker, expectedVersion int64) error {
-	if c.onUpdate != nil && c.conflictTriggered.CompareAndSwap(false, true) {
+	if c.shouldInject != nil && c.shouldInject(worker) && c.conflictTriggered.CompareAndSwap(false, true) {
 		c.onUpdate(ctx, worker)
 	}
 	return c.Interface.UpdateWorker(ctx, worker, expectedVersion)
@@ -998,7 +999,20 @@ func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) {
 
 	var cs *conflictStore
 	persistence, fakeK8s, fakeAte, syncer, cleanup := setupSyncerTestWithStore(t, ctx, func(s store.Interface) store.Interface {
-		cs = &conflictStore{Interface: s}
+		// Configure the injector before the syncer starts. It only fires for the
+		// update under test, so the initial worker creation cannot consume it.
+		cs = &conflictStore{
+			Interface: s,
+			shouldInject: func(w *ateapipb.Worker) bool {
+				return w.GetSandboxClass() == "microvm"
+			},
+			onUpdate: func(c context.Context, w *ateapipb.Worker) {
+				if cw, err := s.GetWorker(c, ns, poolName, podName); err == nil {
+					cw.NodeName = "node2"
+					_ = s.UpdateWorker(c, cw, cw.Version)
+				}
+			},
+		}
 		return cs
 	}, pool)
 	defer func() {
@@ -1068,14 +1082,6 @@ func TestSyncer_UpdateWorker_RetryOnVersionConflict(t *testing.T) {
 		t.Fatalf("pool informer cache failed to update: %v", err)
 	}
 
-	// Configure conflictStore to inject a concurrent version bump in Redis when the syncer calls UpdateWorker.
-	cs.onUpdate = func(c context.Context, w *ateapipb.Worker) {
-		if cw, err := cs.Interface.GetWorker(c, ns, poolName, podName); err == nil {
-			cw.NodeName = "node2"
-			_ = cs.Interface.UpdateWorker(c, cw, cw.Version)
-		}
-	}
-
 	// Touch the pod ONCE in K8s so the syncer reconciles it. The first reconcile's
 	// UpdateWorker hits ErrVersionConflict (injected by conflictStore), which requeues
 	// the key with backoff; the retry re-fetches the latest version from Redis.

cmd/atecontroller/internal/controllers/workerpool_apply.go

@@ -78,6 +78,7 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin
 		WithArgs(
 			"--pod-uid=$(POD_UID)",
 			"--atunnel-listen-address=0.0.0.0:443",
+			"--atunnel-connect-listen-address=0.0.0.0:444",
 			"--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem",
 			"--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem",
 			"--atunnel-egress-listen-address=0.0.0.0:15001",
@@ -86,7 +87,11 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin
 		WithPorts(corev1ac.ContainerPort().
 			WithName("https").
 			WithContainerPort(443).
-			WithProtocol(corev1.ProtocolTCP)).
+			WithProtocol(corev1.ProtocolTCP),
+			corev1ac.ContainerPort().
+				WithName("connect").
+				WithContainerPort(444).
+				WithProtocol(corev1.ProtocolTCP)).
 		WithSecurityContext(ateomSecurityContext(wp.Spec.SandboxClass)).
 		WithEnv(ateomContainerEnv(otel)...).
 		WithVolumeMounts(

cmd/atecontroller/internal/controllers/workerpool_apply_test.go

@@ -735,6 +735,7 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf
 			WithArgs(
 				"--pod-uid=$(POD_UID)",
 				"--atunnel-listen-address=0.0.0.0:443",
+				"--atunnel-connect-listen-address=0.0.0.0:444",
 				"--atunnel-credential-bundle="+atunnelIdentityMountPath+"/credential-bundle.pem",
 				"--atunnel-trust-bundle="+atunnelIdentityMountPath+"/trust-bundle.pem",
 				"--atunnel-egress-listen-address=0.0.0.0:15001",
@@ -743,7 +744,11 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf
 			WithPorts(corev1ac.ContainerPort().
 				WithName("https").
 				WithContainerPort(443).
-				WithProtocol(corev1.ProtocolTCP)).
+				WithProtocol(corev1.ProtocolTCP),
+				corev1ac.ContainerPort().
+					WithName("connect").
+					WithContainerPort(444).
+					WithProtocol(corev1.ProtocolTCP)).
 			WithSecurityContext(corev1ac.SecurityContext().
 				WithRunAsUser(0).
 				WithRunAsGroup(0).

cmd/ateom-gvisor/main.go

@@ -65,12 +65,13 @@ var (
 	podUID = pflag.String("pod-uid", "", "The UID of the current pod")
 
 	// TODO(liorlieberman) have a sub package for all atunnel releated things like that
-	atunnelListenAddress       = pflag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS")
-	workerCredentialBundle     = pflag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS")
-	podIdentityTrustBundle     = pflag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet")
-	atunnelClientIdentity      = pflag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS")
-	atunnelEgressListenAddress = pflag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP")
-	egressGatewayTrustBundle   = pflag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway")
+	atunnelListenAddress        = pflag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS")
+	atunnelConnectListenAddress = pflag.String("atunnel-connect-listen-address", "0.0.0.0:444", "Address for actor ingress mTLS CONNECT")
+	workerCredentialBundle      = pflag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS")
+	podIdentityTrustBundle      = pflag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet")
+	atunnelClientIdentity       = pflag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS")
+	atunnelEgressListenAddress  = pflag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP")
+	egressGatewayTrustBundle    = pflag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway")
 
 	showVersion  = pflag.Bool("version", false, "Print version and exit.")
 	logLevelFlag = pflag.String("log-level", "info", "Minimum log level: debug, info, warn, or error.")
@@ -261,6 +262,16 @@ func runAtunnel(ctx context.Context, upstream *url.URL) (*atunnel.Server, *atunn
 		}
 	}()
 	slog.InfoContext(ctx, "atunnel serving", slog.String("address", *atunnelListenAddress))
+	atunnelConnectListener, err := net.Listen("tcp", *atunnelConnectListenAddress)
+	if err != nil {
+		return nil, nil, 0, fmt.Errorf("while opening atunnel CONNECT listener: %w", err)
+	}
+	go func() {
+		if err := atunnelIngress.ServeConnect(ctx, atunnelConnectListener); err != nil {
+			serverboot.Fatal(ctx, "Failed to serve actor CONNECT ingress", err)
+		}
+	}()
+	slog.InfoContext(ctx, "atunnel CONNECT serving", slog.String("address", *atunnelConnectListenAddress))
 
 	atunnelEgress, err := atunnel.NewEgress(atunnel.TCPOriginalDestination)
 	if err != nil {

cmd/ateom-microvm/main.go

@@ -71,12 +71,13 @@ var (
 	otlpRelaySocket = flag.String("otlp-relay-socket", ateompath.AteletOTLPSocketPath(),
 		"Unix socket of atelet's OTLP relay to export telemetry through, keeping it off the pod network. Empty, or absent at startup, exports directly to OTEL_EXPORTER_OTLP_ENDPOINT instead.")
 
-	atunnelListenAddress       = flag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS")
-	workerCredentialBundle     = flag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS")
-	podIdentityTrustBundle     = flag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet")
-	atunnelClientIdentity      = flag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS")
-	atunnelEgressListenAddress = flag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP")
-	egressGatewayTrustBundle   = flag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway")
+	atunnelListenAddress        = flag.String("atunnel-listen-address", "0.0.0.0:443", "Address for actor ingress HTTPS")
+	atunnelConnectListenAddress = flag.String("atunnel-connect-listen-address", "0.0.0.0:444", "Address for actor ingress mTLS CONNECT")
+	workerCredentialBundle      = flag.String("atunnel-credential-bundle", "/run/podidentity.podcert.ate.dev/credential-bundle.pem", "Worker Pod credential bundle used by atunnel for inbound serving and outbound mTLS")
+	podIdentityTrustBundle      = flag.String("atunnel-trust-bundle", "/run/podidentity.podcert.ate.dev/trust-bundle.pem", "Pod identity trust bundle used for router clients and the node-local atelet")
+	atunnelClientIdentity       = flag.String("atunnel-client-identity", "spiffe://cluster.local/ns/ate-system/sa/atenet-router", "SPIFFE identity allowed to call actor ingress HTTPS")
+	atunnelEgressListenAddress  = flag.String("atunnel-egress-listen-address", "0.0.0.0:15001", "Address for transparently intercepted actor egress TCP")
+	egressGatewayTrustBundle    = flag.String("atunnel-egress-trust-bundle", "/run/servicedns.podcert.ate.dev/trust-bundle.pem", "Service DNS trust bundle for the remote egress gateway")
 )
 
 const (
@@ -221,6 +222,16 @@ func do(ctx context.Context) error {
 		}
 	}()
 	slog.InfoContext(ctx, "atunnel serving", slog.String("address", *atunnelListenAddress))
+	atunnelConnectListener, err := net.Listen("tcp", *atunnelConnectListenAddress)
+	if err != nil {
+		return fmt.Errorf("while opening atunnel CONNECT listener: %w", err)
+	}
+	go func() {
+		if err := atunnelIngress.ServeConnect(ctx, atunnelConnectListener); err != nil {
+			serverboot.Fatal(ctx, "Failed to serve actor CONNECT ingress", err)
+		}
+	}()
+	slog.InfoContext(ctx, "atunnel CONNECT serving", slog.String("address", *atunnelConnectListenAddress))
 	atunnelEgress, err := atunnel.NewEgress(atunnel.TCPOriginalDestination)
 	if err != nil {
 		return fmt.Errorf("while configuring atunnel egress: %w", err)

demos/counter/counter.go

@@ -69,6 +69,7 @@ func main() {
 	secondFileCounterDirectory := pflag.String("second-file-counter-directory", "", "Directory for a second file counter; empty disables it. Used to exercise an Actor with more than one durable volume")
 	validateExistingFilePath := pflag.String("validate-existing-file-path", "", "Path to existing file to validate reading")
 	extraPort := pflag.Int("extra-port", 0, "Additional port to listen on, for exercising atenet-router's arbitrary-port ingress support; 0 disables it")
+	tcpPort := pflag.Int("tcp-port", 0, "Plain TCP echo port for exercising atunnel CONNECT ingress; 0 disables it")
 	pflag.Parse()
 	ctx := context.Background()
 
@@ -177,6 +178,28 @@ func main() {
 		}()
 	}
 
+	if *tcpPort > 0 {
+		go func() {
+			listener, err := net.Listen("tcp", fmt.Sprintf(":%d", *tcpPort))
+			if err != nil {
+				slog.ErrorContext(ctx, "Error starting counter TCP echo server", slog.Any("err", err))
+				os.Exit(1)
+			}
+			slog.InfoContext(ctx, "Starting counter TCP echo server", slog.Int("port", *tcpPort))
+			for {
+				conn, err := listener.Accept()
+				if err != nil {
+					slog.ErrorContext(ctx, "Counter TCP echo accept failed", slog.Any("err", err))
+					return
+				}
+				go func() {
+					defer conn.Close()
+					_, _ = io.Copy(conn, conn)
+				}()
+			}
+		}()
+	}
+
 	// Write some random data to a file in the root filesystem, to test
 	// filesystem checkpoint/restore.
 	if err := writeRandomFile(); err != nil {

demos/counter/counter.yaml.tmpl

@@ -69,6 +69,7 @@ spec:
     # listener a test can address by CONNECTing to <actor-dns>:9090, distinct
     # from the primary port 80 every other assertion in this demo uses.
     - --extra-port=9090
+    - --tcp-port=9091
 ${VALIDATE_EXISTING_FILE_PATH_ARG}
     readyz:
       httpGet: