feat: Serve instance shell sessions from compute cells

.github/workflows/publish.yaml

@@ -56,8 +56,9 @@ jobs:
   # can only stamp one image across N paths, which can't express "three
   # different images, four different paths" at all.
   #
-  # compute-mcp is stamped with the manager's image and the manager's tag: one
-  # image carries both binaries, so the two paths cannot name different builds.
+  # compute-mcp and the shell agent are stamped with the manager's image and
+  # the manager's tag: one image carries all three binaries, so the paths
+  # cannot name different builds.
   publish-kustomize-bundles:
     needs:
       [
@@ -78,4 +79,5 @@ jobs:
         - { path: config/components/consumer-ui-plugin, name: ghcr.io/datum-cloud/compute-consumer-ui-plugin, tag: "${{ needs.publish-consumer-ui-plugin-image.outputs.tag }}" }
         - { path: config/components/provider-ui-plugin, name: ghcr.io/datum-cloud/compute-provider-ui-plugin, tag: "${{ needs.publish-provider-ui-plugin-image.outputs.tag }}" }
         - { path: config/components/compute-mcp, name: ghcr.io/datum-cloud/compute, tag: "${{ needs.publish-container-image.outputs.tag }}" }
+        - { path: config/components/shell-agent, name: ghcr.io/datum-cloud/compute, tag: "${{ needs.publish-container-image.outputs.tag }}" }
     secrets: inherit

Dockerfile

@@ -28,6 +28,7 @@ RUN go mod download
 # package whose embed.FS carries the knowledge and runbooks compute-mcp serves.
 COPY cmd/main.go cmd/main.go
 COPY cmd/compute-mcp/ cmd/compute-mcp/
+COPY cmd/shell-agent/ cmd/shell-agent/
 COPY api/ api/
 COPY internal/ internal/
 COPY pkg/ pkg/
@@ -61,6 +62,16 @@ RUN --mount=type=cache,target=/go/pkg/mod/ \
     -ldflags "-s -w" \
     -o compute-mcp ./cmd/compute-mcp
 
+# shell-agent runs in each cell beside its tunnel endpoint. It shares the
+# session API and the connection protocol package with the manager, so it
+# ships in the same image and cannot skew from the controllers that deliver
+# its sessions.
+RUN --mount=type=cache,target=/go/pkg/mod/ \
+  --mount=type=cache,target="/root/.cache/go-build" \
+  CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build \
+    -ldflags "-s -w" \
+    -o shell-agent ./cmd/shell-agent
+
 # Use distroless as minimal base image to package the binaries
 # Refer to https://github.com/GoogleContainerTools/distroless for more details
 #
@@ -74,6 +85,7 @@ FROM gcr.io/distroless/static:nonroot
 WORKDIR /
 COPY --from=builder /workspace/manager .
 COPY --from=builder /workspace/compute-mcp .
+COPY --from=builder /workspace/shell-agent .
 USER 65532:65532
 
 # The manager keeps the entrypoint it has always had, so nothing that runs this

cmd/shell-agent/main.go

@@ -0,0 +1,214 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+// Command shell-agent serves instance shell sessions in a compute cell. Each
+// replica pairs by ordinal with a tunnel endpoint, which is the only way a
+// client reaches it: the endpoint proxies one TCP target, this agent's exec
+// port. The agent claims sessions delivered to the cell, runs their commands
+// through the cell apiserver, and writes their status on the cell's copies,
+// which Karmada reflects back to the hub. It needs no hub credential.
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
+	computev1alpha "go.datum.net/compute/api/v1alpha"
+	"go.datum.net/compute/internal/features"
+	"go.datum.net/compute/internal/shellagent"
+)
+
+var setupLog = ctrl.Log.WithName("setup")
+
+type options struct {
+	listen            string
+	probeAddr         string
+	target            string
+	relayURLs         string
+	managedBy         string
+	endpointPodPrefix string
+	featureGates      string
+	drainTimeout      time.Duration
+	killGrace         time.Duration
+	pingInterval      time.Duration
+	pongTimeout       time.Duration
+}
+
+func main() {
+	defaults := shellagent.DefaultConfig()
+	var opts options
+	flag.StringVar(&opts.listen, "listen", ":7777", "Address to serve session connections on.")
+	flag.StringVar(&opts.probeAddr, "health-probe-bind-address", ":8081", "Address the probe endpoint binds to.")
+	flag.StringVar(&opts.target, "target", "",
+		"host:port the paired tunnel endpoint proxies to this agent, published as the session's target.")
+	flag.StringVar(&opts.relayURLs, "relay-urls", os.Getenv("DATUM_CONNECT_RELAY_URLS"),
+		"Comma-separated relays the paired tunnel endpoint uses, nearest first.")
+	flag.StringVar(&opts.managedBy, "managed-by", strings.Join(defaults.ManagedBy, ","),
+		"Comma-separated managed-by label values of instance pods that may take sessions.")
+	flag.StringVar(&opts.endpointPodPrefix, "endpoint-pod-prefix", "exec-endpoint",
+		"Name of the tunnel endpoint StatefulSet; the paired pod is <prefix>-<ordinal>.")
+	flag.StringVar(&opts.featureGates, "feature-gates", "",
+		"Feature gates, for example InstanceConsoleSessions=true.")
+	flag.DurationVar(&opts.drainTimeout, "drain-timeout", defaults.DrainTimeout,
+		"How long connected sessions may continue after the agent is asked to stop.")
+	flag.DurationVar(&opts.killGrace, "kill-grace", defaults.KillGrace,
+		"Time between SIGHUP and SIGKILL when a session's processes are stopped.")
+	flag.DurationVar(&opts.pingInterval, "ping-interval", defaults.PingInterval,
+		"How often the agent pings a connected client.")
+	flag.DurationVar(&opts.pongTimeout, "pong-timeout", defaults.PongTimeout,
+		"How long a connected client may send nothing, not even a pong, before its session ends as Disconnected.")
+	zapOpts := zap.Options{}
+	zapOpts.BindFlags(flag.CommandLine)
+	flag.Parse()
+	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&zapOpts)))
+
+	if err := run(opts, defaults); err != nil {
+		setupLog.Error(err, "shell agent failed")
+		os.Exit(1)
+	}
+}
+
+func run(opts options, cfg shellagent.Config) error {
+	if opts.featureGates != "" {
+		if err := features.MutableFeatureGate.Set(opts.featureGates); err != nil {
+			return fmt.Errorf("parse feature gates: %w", err)
+		}
+	}
+	signals := make(chan os.Signal, 2)
+	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
+
+	if !features.FeatureGate.Enabled(features.InstanceConsoleSessions) {
+		setupLog.Info("the InstanceConsoleSessions feature gate is off; not serving sessions")
+		<-signals
+		return nil
+	}
+
+	podName, namespace := os.Getenv("POD_NAME"), os.Getenv("POD_NAMESPACE")
+	ordinal, err := ordinalOf(podName)
+	if err != nil || namespace == "" {
+		return fmt.Errorf("POD_NAME must be a StatefulSet pod name and POD_NAMESPACE must be set: %w", err)
+	}
+	cfg.Namespace = namespace
+	cfg.Ordinal = ordinal
+	cfg.EndpointPodName = fmt.Sprintf("%s-%d", opts.endpointPodPrefix, ordinal)
+	cfg.Target = opts.target
+	cfg.RelayURLs = splitList(opts.relayURLs)
+	cfg.ManagedBy = splitList(opts.managedBy)
+	cfg.DrainTimeout = opts.drainTimeout
+	cfg.KillGrace = opts.killGrace
+	cfg.PingInterval = opts.pingInterval
+	cfg.PongTimeout = opts.pongTimeout
+
+	scheme := runtime.NewScheme()
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
+	utilruntime.Must(computev1alpha.AddToScheme(scheme))
+
+	cellConfig := ctrl.GetConfigOrDie()
+
+	mgr, err := ctrl.NewManager(cellConfig, ctrl.Options{
+		Scheme:                 scheme,
+		Metrics:                metricsserver.Options{BindAddress: "0"},
+		HealthProbeBindAddress: opts.probeAddr,
+	})
+	if err != nil {
+		return fmt.Errorf("create manager: %w", err)
+	}
+	cellClient, err := client.New(cellConfig, client.Options{Scheme: scheme})
+	if err != nil {
+		return err
+	}
+	executor, err := shellagent.NewAPIServer(cellConfig)
+	if err != nil {
+		return fmt.Errorf("configure pod exec: %w", err)
+	}
+
+	incarnation := podName + "/" + strconv.FormatInt(time.Now().UnixNano(), 36)
+	agent, err := shellagent.New(cfg, mgr.GetClient(), cellClient, executor, incarnation)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithCancel(ctrl.LoggerInto(context.Background(), ctrl.Log.WithName("shell-agent")))
+	defer cancel()
+	if err := agent.EnsureIdentity(ctx); err != nil {
+		return err
+	}
+	if err := agent.SetupWithManager(mgr); err != nil {
+		return fmt.Errorf("set up controller: %w", err)
+	}
+
+	server := &http.Server{Addr: opts.listen, Handler: agent.Handler(ctx), ReadHeaderTimeout: 10 * time.Second}
+	for _, r := range []manager.RunnableFunc{
+		func(ctx context.Context) error {
+			go func() {
+				<-ctx.Done()
+				_ = server.Close()
+			}()
+			if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
+				return err
+			}
+			return nil
+		},
+		agent.RunLiveness,
+		agent.RunSweeper,
+		agent.RunKeyRotation,
+	} {
+		if err := mgr.Add(r); err != nil {
+			return err
+		}
+	}
+	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
+		return err
+	}
+	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
+		return err
+	}
+
+	go func() {
+		<-signals
+		setupLog.Info("draining", "timeout", cfg.DrainTimeout)
+		agent.Drain(ctx)
+		setupLog.Info("drained")
+		cancel()
+		<-signals
+		os.Exit(1)
+	}()
+
+	setupLog.Info("starting shell agent", "endpointID", agent.EndpointID(), "target", cfg.Target)
+	return mgr.Start(ctx)
+}
+
+func ordinalOf(podName string) (int, error) {
+	i := strings.LastIndex(podName, "-")
+	if i < 0 {
+		return 0, fmt.Errorf("pod name %q has no ordinal", podName)
+	}
+	return strconv.Atoi(podName[i+1:])
+}
+
+func splitList(s string) []string {
+	var out []string
+	for _, v := range strings.Split(s, ",") {
+		if v = strings.TrimSpace(v); v != "" {
+			out = append(out, v)
+		}
+	}
+	return out
+}

config/components/shell-agent-cilium/kustomization.yaml

@@ -0,0 +1,10 @@
+# Cilium policies for the shell agent and its tunnel endpoints. Include beside
+# config/components/shell-agent on cells that run Cilium. Unlike the portable
+# policies they police traffic to the node's own apiserver, kubelet and etcd,
+# so the endpoint, which holds no credential, cannot reach them at all. The
+# agent never leaves the cell: it talks only to the cell apiserver.
+apiVersion: kustomize.config.k8s.io/v1alpha1
+kind: Component
+
+resources:
+  - policy.yaml

config/components/shell-agent-cilium/policy.yaml

@@ -0,0 +1,85 @@
+apiVersion: cilium.io/v2
+kind: CiliumNetworkPolicy
+metadata:
+  name: exec-agent
+  namespace: compute-shell-system
+spec:
+  endpointSelector:
+    matchLabels:
+      app.kubernetes.io/name: exec-agent
+  enableDefaultDeny:
+    ingress: true
+    egress: true
+  ingress:
+    - fromEndpoints:
+        - matchLabels:
+            app.kubernetes.io/name: exec-endpoint
+      toPorts:
+        - ports:
+            - port: "7777"
+              protocol: TCP
+    - fromEntities:
+        - host
+      toPorts:
+        - ports:
+            - port: "8081"
+              protocol: TCP
+  egress:
+    - toEntities:
+        - kube-apiserver
+      toPorts:
+        - ports:
+            - port: "443"
+              protocol: TCP
+            - port: "6443"
+              protocol: TCP
+    - toEndpoints:
+        - matchLabels:
+            k8s:io.kubernetes.pod.namespace: kube-system
+            k8s-app: kube-dns
+      toPorts:
+        - ports:
+            - port: "53"
+              protocol: UDP
+            - port: "53"
+              protocol: TCP
+---
+apiVersion: cilium.io/v2
+kind: CiliumNetworkPolicy
+metadata:
+  name: exec-endpoint
+  namespace: compute-shell-system
+spec:
+  endpointSelector:
+    matchLabels:
+      app.kubernetes.io/name: exec-endpoint
+  enableDefaultDeny:
+    ingress: true
+    egress: true
+  egress:
+    - toEndpoints:
+        - matchLabels:
+            app.kubernetes.io/name: exec-agent
+      toPorts:
+        - ports:
+            - port: "7777"
+              protocol: TCP
+    - toEndpoints:
+        - matchLabels:
+            k8s:io.kubernetes.pod.namespace: kube-system
+            k8s-app: kube-dns
+      toPorts:
+        - ports:
+            - port: "53"
+              protocol: UDP
+            - port: "53"
+              protocol: TCP
+    # Relays only: world excludes the cluster's own nodes and apiserver.
+    - toEntities:
+        - world
+      toPorts:
+        - ports:
+            - port: "443"
+              protocol: TCP
+            - port: "7842"
+              protocol: UDP