diff --git a/.github/workflows/build_images.yaml b/.github/workflows/build_images.yaml index 7e164188..fd32015d 100644 --- a/.github/workflows/build_images.yaml +++ b/.github/workflows/build_images.yaml @@ -71,7 +71,17 @@ jobs: ghcr.io/${{ steps.get_repo_owner.outputs.repo_owner }}/interlink/interlink:latest file: ./docker/Dockerfile.interlink platforms: linux/amd64, linux/arm64, linux/aarch64 - + - name: Build container base image ssh-tunnel + uses: docker/build-push-action@v6 + with: + context: ./ + outputs: "type=registry,push=true" + tags: | + ghcr.io/${{ steps.get_repo_owner.outputs.repo_owner }}/interlink/ssh-tunnel:${{ env.RELEASE_VERSION }} + ghcr.io/${{ steps.get_repo_owner.outputs.repo_owner }}/interlink/ssh-tunnel:latest + file: ./docker/Dockerfile.ssh-tunnel + platforms: linux/amd64, linux/arm64, linux/aarch64 + virtual-kubelet-refresh-token: runs-on: ubuntu-latest #env: diff --git a/docker/Dockerfile.ssh-tunnel b/docker/Dockerfile.ssh-tunnel new file mode 100644 index 00000000..6b40f904 --- /dev/null +++ b/docker/Dockerfile.ssh-tunnel @@ -0,0 +1,15 @@ +# Image for the SSH port-forward shadow pod (Network.ShadowMode: ssh). +# +# It runs `ssh -N -L ...` against an HPC login node, so it needs an ssh client and, +# for GSSAPI sites, the Kerberos client tools to obtain and renew a ticket from a +# keytab. socat provides the local listeners used by Network.SSH.ForwardMode "exec", +# where the login node grants no forwarding privilege. Nothing from this repository +# is installed: the shadow runs no interLink binary, only stock ssh. +FROM debian:bookworm-slim + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + openssh-client krb5-user socat ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +CMD ["/bin/sh"] diff --git a/pkg/interlink/types.go b/pkg/interlink/types.go index 5356d7cf..1e63d4a9 100644 --- a/pkg/interlink/types.go +++ b/pkg/interlink/types.go @@ -42,6 +42,12 @@ type PodStatus struct { PodNamespace string `json:"namespace"` // JobID is the remote system's job identifier (e.g., SLURM job ID, container ID) JobID string `json:"JID"` + // NodeName is the remote compute node the pod was allocated on, as the plugin's + // site resolves it (typically a fully qualified hostname reachable from the login + // node). Plugins report it once the workload is actually running; it is empty + // while the job is still queued, and empty for plugins that do not track it. + // interLink uses it to point per-pod shadow tunnels at the right host. + NodeName string `json:"nodeName,omitempty"` // Containers holds the status of all regular containers in the pod Containers []v1.ContainerStatus `json:"containers"` // InitContainers holds the status of all init containers in the pod diff --git a/pkg/virtualkubelet/config.go b/pkg/virtualkubelet/config.go index db867798..e0426d33 100644 --- a/pkg/virtualkubelet/config.go +++ b/pkg/virtualkubelet/config.go @@ -173,4 +173,104 @@ type Network struct { Slirp4netnsURL string `yaml:"Slirp4netnsURL,omitempty"` // UnsharedMode is the flag for unshared network mode in slirp4netns UnshareMode string `yaml:"UnshareMode,omitempty"` + // ShadowMode selects which shadow implementation is rendered for offloaded pods + // with exposed ports: "wstunnel" (default) or "ssh". + ShadowMode string `yaml:"ShadowMode,omitempty"` + // SSH configures the SSH port-forward shadow, used when ShadowMode is "ssh" + SSH SSHTunnel `yaml:"SSH,omitempty"` +} + +// Shadow implementations selectable through Network.ShadowMode. +const ( + // ShadowModeWstunnel exposes the offloaded pod's ports by having the workload + // dial out to a public ingress and run a wstunnel client. This is the default. + ShadowModeWstunnel = "wstunnel" + // ShadowModeSSH exposes them the other way round: the shadow dials in to an SSH + // login node and forwards each port to the compute node the job landed on. The + // workload runs nothing, and no compute node needs outbound internet access. + ShadowModeSSH = "ssh" +) + +// Traffic-forwarding strategies selectable through SSHTunnel.ForwardMode. +const ( + // SSHForwardModePortForward uses `ssh -L`, and needs AllowTcpForwarding on the + // login node. This is the default. + SSHForwardModePortForward = "portforward" + // SSHForwardModeExec pipes each connection through a command run on the login + // node, for sites that do not grant TCP forwarding. + SSHForwardModeExec = "exec" +) + +// DefaultSSHExecConnectCommand relays a connection on stdin/stdout in "exec" mode. +const DefaultSSHExecConnectCommand = "nc" + +// SSH authentication methods selectable through SSHTunnel.Auth. +const ( + // SSHAuthPublicKey authenticates with a private key from KeySecret. + SSHAuthPublicKey = "publickey" + // SSHAuthKerberos authenticates with GSSAPI, using a keytab from KeytabSecret. + SSHAuthKerberos = "kerberos" +) + +// SSHTunnel configures the SSH port-forward shadow. +// +// The shadow runs `ssh -N -L ::` against the site's login +// node, one -L per exposed port, so cluster traffic reaches services inside an +// offloaded pod without the compute node needing any outbound connectivity. It +// covers the same ground as the wstunnel shadow, in the opposite direction; it does +// not give the offloaded pod access back into the cluster (see Network.FullMesh). +type SSHTunnel struct { + // LoginHost is the SSH login node to forward through (required) + LoginHost string `yaml:"LoginHost,omitempty"` + // Port is the login node's SSH port (default 22) + Port int `yaml:"Port,omitempty"` + // User is the login name on the login node (required) + User string `yaml:"User,omitempty"` + // Image is the container image running in the shadow. It must provide an ssh + // client, and kinit/klist when Auth is "kerberos". + Image string `yaml:"Image,omitempty"` + // Auth selects the authentication method: "publickey" (default) or "kerberos" + Auth string `yaml:"Auth,omitempty"` + // ForwardMode selects how traffic reaches the compute node: + // + // "portforward" (default) — one `ssh -L` per exposed port. Cheapest and most + // direct, but the login node must set AllowTcpForwarding yes for this + // account. Sites that disable it refuse every channel with + // "administratively prohibited". + // "exec" — a local listener per exposed port, each connection piped through a + // command run on the login node (see ExecConnectCommand). Needs no + // forwarding privilege at all, at the cost of one ssh process per connection. + ForwardMode string `yaml:"ForwardMode,omitempty"` + // ExecConnectCommand is the command run on the login node in "exec" mode. It is + // invoked as ` ` and must relay the connection on + // its stdin and stdout. Defaults to "nc". + ExecConnectCommand string `yaml:"ExecConnectCommand,omitempty"` + // KeySecret is the Secret holding the SSH private key ("publickey" auth) + KeySecret string `yaml:"KeySecret,omitempty"` + // KeySecretKey is the key inside KeySecret holding the private key (default "id_ed25519") + KeySecretKey string `yaml:"KeySecretKey,omitempty"` + // KeytabSecret is the Secret holding the Kerberos keytab ("kerberos" auth) + KeytabSecret string `yaml:"KeytabSecret,omitempty"` + // KeytabSecretKey is the key inside KeytabSecret holding the keytab (default "user.keytab") + KeytabSecretKey string `yaml:"KeytabSecretKey,omitempty"` + // Principal is the Kerberos principal to obtain a ticket for ("kerberos" auth) + Principal string `yaml:"Principal,omitempty"` + // Krb5ConfigMap is an optional ConfigMap holding a krb5.conf to mount at /etc/krb5.conf + Krb5ConfigMap string `yaml:"Krb5ConfigMap,omitempty"` + // KnownHostsConfigMap is an optional ConfigMap holding a known_hosts file. When + // unset the shadow falls back to StrictHostKeyChecking=accept-new, which trusts + // whatever key the login node presents on first contact. + KnownHostsConfigMap string `yaml:"KnownHostsConfigMap,omitempty"` + // ReplicateCredentials copies the referenced Secret and ConfigMaps from the + // virtual kubelet's own namespace into the shadow's namespace, so offloaded pods + // in arbitrary (e.g. per-user) namespaces work without pre-seeding credentials + // everywhere. Defaults to true. Note this makes the credential readable by anyone + // who can read Secrets in those namespaces. + ReplicateCredentials *bool `yaml:"ReplicateCredentials,omitempty"` + // NodeWaitTimeout bounds how long the shadow waits for the plugin to report the + // compute node before failing (default "2h"). Queue waits are normal, so this is + // generous by design. + NodeWaitTimeout string `yaml:"NodeWaitTimeout,omitempty"` + // ExtraOptions are additional ssh client options, each passed verbatim as -o + ExtraOptions []string `yaml:"ExtraOptions,omitempty"` } diff --git a/pkg/virtualkubelet/execute.go b/pkg/virtualkubelet/execute.go index de49f901..1f6f5db1 100644 --- a/pkg/virtualkubelet/execute.go +++ b/pkg/virtualkubelet/execute.go @@ -1531,6 +1531,10 @@ func checkPodsStatus(ctx context.Context, p *Provider, pod *v1.Pod, token string // if the PodUID match with the one in etcd we are talking of the same thing. GOOD if podRemoteStatus.PodUID == string(podRefInCluster.UID) { + // The plugin reports the remote compute node only once the job is actually + // running, so this stays a no-op for as long as the job sits in the queue. + p.publishShadowNodeName(ctx, podRefInCluster, podRemoteStatus.NodeName) + // check if the pod is already in a terminal state (Failed or Succeeded) if currentPhase, terminal := p.podTerminalPhase(podRemoteStatus.PodUID); terminal { if podRefInCluster.Status.Phase == currentPhase { diff --git a/pkg/virtualkubelet/mesh.go b/pkg/virtualkubelet/mesh.go index e4c05b25..4e38c520 100644 --- a/pkg/virtualkubelet/mesh.go +++ b/pkg/virtualkubelet/mesh.go @@ -4,9 +4,7 @@ import ( "bytes" "context" "crypto/rand" - "crypto/sha256" "encoding/base64" - "encoding/hex" "encoding/json" "fmt" "os" @@ -20,217 +18,6 @@ import ( k8stypes "k8s.io/apimachinery/pkg/types" ) -func sanitizeDNSName(name string) string { - // Convert to lowercase - name = strings.ToLower(name) - - // Replace any invalid characters with hyphens - var builder strings.Builder - for _, r := range name { - if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') { - builder.WriteRune(r) - } else { - builder.WriteRune('-') - } - } - name = builder.String() - - // Remove leading and trailing hyphens - name = strings.Trim(name, "-") - - // Collapse consecutive hyphens into a single hyphen - for strings.Contains(name, "--") { - name = strings.ReplaceAll(name, "--", "-") - } - - // Truncate to 63 characters (max label length) - if len(name) > 63 { - name = name[:63] - // Ensure we don't end with a hyphen after truncation - name = strings.TrimRight(name, "-") - } - - // If the result is empty, provide a default - if name == "" { - name = "default" - } - - return name -} - -// sanitizeFullDNSName sanitizes a full DNS name (with dots) to ensure it meets RFC 1123 requirements -func sanitizeFullDNSName(fullName string) string { - // Split by dots to handle each label separately - labels := strings.Split(fullName, ".") - - // Sanitize each label - sanitizedLabels := make([]string, 0, len(labels)) - for _, label := range labels { - if label == "" { - continue - } - sanitized := sanitizeDNSName(label) - if sanitized != "" { - sanitizedLabels = append(sanitizedLabels, sanitized) - } - } - - // Rejoin with dots - result := strings.Join(sanitizedLabels, ".") - - // Ensure total length doesn't exceed 253 characters - if len(result) > 253 { - // Truncate from the beginning (keeping the domain suffix) - excess := len(result) - 253 - result = result[excess:] - // Make sure we don't start with a dot after truncation - result = strings.TrimLeft(result, ".") - } - - return result -} - -// uniqueTruncate shortens a name and adds an 8-character hash to prevent naming collisions. -func uniqueTruncate(s string, maxLen int, full string) string { - if len(s) <= maxLen { - return s - } - h := sha256.Sum256([]byte(full)) - suffix := hex.EncodeToString(h[:4]) - keep := maxLen - len(suffix) - 1 - return strings.TrimRight(s[:keep], "-") + "-" + suffix -} - -type wstunnelResourceIdentity struct { - Name string - Namespace string -} - -func isShadowSameNamespace(pod *v1.Pod) bool { - if pod == nil || pod.Annotations == nil { - return false - } - return pod.Annotations["interlink.eu/shadow-same-ns"] == "true" -} - -func computeWstunnelResourceIdentity(pod *v1.Pod) (wstunnelResourceIdentity, error) { - if pod == nil { - return wstunnelResourceIdentity{}, fmt.Errorf("pod is nil") - } - if pod.Namespace == "" { - return wstunnelResourceIdentity{}, fmt.Errorf("pod namespace is empty") - } - - var name, namespace string - if isShadowSameNamespace(pod) { - name, namespace = computeWstunnelResourceNamesForSameNamespace(pod.Name, pod.Namespace) - } else { - name, namespace = computeWstunnelResourceNames(pod.Name, pod.Namespace) - } - - identity := wstunnelResourceIdentity{Name: name, Namespace: namespace} - if len(fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) > 63 { - return wstunnelResourceIdentity{}, fmt.Errorf("wstunnel ingress hostname label %q exceeds 63 characters; shorten pod/namespace or disable interlink.eu/shadow-same-ns", fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) - } - - return identity, nil -} - -func computeWstunnelResourceNamesForSameNamespace(podName, podNamespace string) (resourceBaseName, namespace string) { - // Sanitize namespace and pod name for DNS compliance - sanitizedNamespace := sanitizeDNSName(podNamespace) - sanitizedPodName := sanitizeDNSName(podName) - - // Use the original namespace. Do not truncate it: in same-namespace mode - // resources must be created in the pod's real namespace. - namespace = podNamespace - - // Create a unique resource name to avoid conflicts in the same namespace - // Add "wstunnel-" prefix to distinguish shadow pod resources - resourceBaseName = "wstunnel-" + sanitizedPodName + "-" + sanitizedNamespace - // Hash on the unsanitized names: sanitizeDNSName truncates to 63 chars, long pods could still collide - fullBaseName := "wstunnel-" + podName + "-" + podNamespace - - // Ensure resourceBaseName doesn't exceed 63 characters - if len(resourceBaseName) > 63 { - // Truncate while keeping some of both names - maxPodNameLen := 28 - maxNsLen := 28 - if len(sanitizedPodName) > maxPodNameLen { - sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) - } - if len(sanitizedNamespace) > maxNsLen { - sanitizedNamespace = sanitizedNamespace[:maxNsLen] - } - resourceBaseName = "wstunnel-" + sanitizedPodName + "-" + sanitizedNamespace - resourceBaseName = strings.TrimRight(resourceBaseName, "-") - } - - // Additional check for total length after combining with namespace - ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, namespace) - if len(ingressFirstLabel) > 63 { - maxNameLen := 63 - len(namespace) - 1 - if maxNameLen > 9 && len(resourceBaseName) > maxNameLen { // >9: room for the 8-char hash suffix - resourceBaseName = uniqueTruncate(resourceBaseName, maxNameLen, fullBaseName) - } - } - - return resourceBaseName, namespace -} - -func computeWstunnelResourceNames(podName, podNamespace string) (resourceBaseName, wstunnelNamespace string) { - // Sanitize namespace and pod name for DNS compliance - sanitizedNamespace := sanitizeDNSName(podNamespace) - sanitizedPodName := sanitizeDNSName(podName) - - wstunnelNamespace = sanitizedNamespace + "-wstunnel" - // Ensure wstunnelNamespace is valid (max 63 chars for namespace) - if len(wstunnelNamespace) > 63 { - wstunnelNamespace = sanitizedNamespace[:min(54, len(sanitizedNamespace))] + "-wstunnel" - } - - resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace - fullBaseName := podName + "-" + podNamespace - // Ensure resourceBaseName doesn't exceed 63 characters - if len(resourceBaseName) > 63 { - // Truncate while keeping some of both names - maxPodNameLen := 31 - maxNsLen := 31 - if len(sanitizedPodName) > maxPodNameLen { - sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) - } - if len(sanitizedNamespace) > maxNsLen { - sanitizedNamespace = sanitizedNamespace[:maxNsLen] - } - resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace - resourceBaseName = strings.TrimRight(resourceBaseName, "-") - } - - ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, wstunnelNamespace) - if len(ingressFirstLabel) > 63 { - // If combined length exceeds 63, we need to truncate - // Strategy: keep both parts but truncate proportionally - maxNameLen := 31 - maxNsLen := 31 - - truncatedName := resourceBaseName - if len(truncatedName) > maxNameLen { - truncatedName = uniqueTruncate(truncatedName, maxNameLen, fullBaseName) - } - - truncatedNs := wstunnelNamespace - if len(truncatedNs) > maxNsLen { - truncatedNs = truncatedNs[:maxNsLen] - truncatedNs = strings.TrimRight(truncatedNs, "-") - } - - resourceBaseName = truncatedName - wstunnelNamespace = truncatedNs - } - - return resourceBaseName, wstunnelNamespace -} - func generateWGKeypair() (string, string, error) { // 32 random bytes -> clamp per X25519 rules -> public = X25519(priv, basepoint) privRaw := make([]byte, 32) @@ -273,7 +60,7 @@ func deriveWGPublicKey(privB64 string) (string, error) { } // addWstunnelClientAnnotation adds the wstunnel client command annotation to the original pod -func (p *Provider) addWstunnelClientAnnotation(ctx context.Context, pod *v1.Pod, td *WstunnelTemplateData) error { +func (p *Provider) addWstunnelClientAnnotation(ctx context.Context, pod *v1.Pod, td *ShadowTemplateData) error { if pod.Annotations == nil { pod.Annotations = make(map[string]string) } @@ -399,7 +186,7 @@ func clearConflictingNetworkAnnotations(pod *v1.Pod, fullMeshEnabledForPod bool) delete(pod.Annotations, annWGClientSnippet) } -func (p *Provider) generateFullMeshScript(ctx context.Context, td *WstunnelTemplateData, ingressEndpoint string, podUID string) (string, error) { +func (p *Provider) generateFullMeshScript(ctx context.Context, td *ShadowTemplateData, ingressEndpoint string, podUID string) (string, error) { serverPub, err := deriveWGPublicKey(td.WGPrivateKey) if err != nil { diff --git a/pkg/virtualkubelet/shadow.go b/pkg/virtualkubelet/shadow.go new file mode 100644 index 00000000..02427048 --- /dev/null +++ b/pkg/virtualkubelet/shadow.go @@ -0,0 +1,353 @@ +package virtualkubelet + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "regexp" + "strings" + + "github.com/containerd/containerd/log" + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8stypes "k8s.io/apimachinery/pkg/types" +) + +// shadowResourcePrefix is prepended to shadow resource names created in the +// offloaded pod's own namespace, so they cannot collide with the pod itself. +const shadowResourcePrefix = "shadow-" + +// shadowNamespaceSuffix is appended to the offloaded pod's namespace to build +// the dedicated namespace shadow resources live in by default. +const shadowNamespaceSuffix = "-shadow" + +// shadowNodeConfigMapSuffix names the per-shadow ConfigMap carrying the remote +// compute node the workload was allocated on. +const shadowNodeConfigMapSuffix = "-node" + +// shadowNodeNameKey is the key inside that ConfigMap holding the node name. +const shadowNodeNameKey = "compute-node" + +// maxComputeNodeNameLen is the longest DNS name, and so the longest thing a plugin +// can legitimately report as a compute node. +const maxComputeNodeNameLen = 253 + +// computeNodeNamePattern matches a hostname or an IP literal and nothing else. The +// shadow interpolates the reported node into a shell command, both locally and on +// the login node, so anything outside this set has to be refused rather than +// escaped: a name containing a space injects an extra ssh argument +// (`-oProxyCommand=...` runs a command in the shadow), and one containing a quote +// or `$(` breaks out of the relay command in exec mode. +var computeNodeNamePattern = regexp.MustCompile(`^[A-Za-z0-9]([A-Za-z0-9._:-]*[A-Za-z0-9])?$`) + +// isValidComputeNodeName reports whether a plugin-supplied node name is safe to put +// in front of ssh. +func isValidComputeNodeName(nodeName string) bool { + return len(nodeName) <= maxComputeNodeNameLen && computeNodeNamePattern.MatchString(nodeName) +} + +func sanitizeDNSName(name string) string { + // Convert to lowercase + name = strings.ToLower(name) + + // Replace any invalid characters with hyphens + var builder strings.Builder + for _, r := range name { + if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') { + builder.WriteRune(r) + } else { + builder.WriteRune('-') + } + } + name = builder.String() + + // Remove leading and trailing hyphens + name = strings.Trim(name, "-") + + // Collapse consecutive hyphens into a single hyphen + for strings.Contains(name, "--") { + name = strings.ReplaceAll(name, "--", "-") + } + + // Truncate to 63 characters (max label length) + if len(name) > 63 { + name = name[:63] + // Ensure we don't end with a hyphen after truncation + name = strings.TrimRight(name, "-") + } + + // If the result is empty, provide a default + if name == "" { + name = "default" + } + + return name +} + +// sanitizeFullDNSName sanitizes a full DNS name (with dots) to ensure it meets RFC 1123 requirements +func sanitizeFullDNSName(fullName string) string { + // Split by dots to handle each label separately + labels := strings.Split(fullName, ".") + + // Sanitize each label + sanitizedLabels := make([]string, 0, len(labels)) + for _, label := range labels { + if label == "" { + continue + } + sanitized := sanitizeDNSName(label) + if sanitized != "" { + sanitizedLabels = append(sanitizedLabels, sanitized) + } + } + + // Rejoin with dots + result := strings.Join(sanitizedLabels, ".") + + // Ensure total length doesn't exceed 253 characters + if len(result) > 253 { + // Truncate from the beginning (keeping the domain suffix) + excess := len(result) - 253 + result = result[excess:] + // Make sure we don't start with a dot after truncation + result = strings.TrimLeft(result, ".") + } + + return result +} + +// uniqueTruncate shortens a name and adds an 8-character hash to prevent naming collisions. +func uniqueTruncate(s string, maxLen int, full string) string { + if len(s) <= maxLen { + return s + } + h := sha256.Sum256([]byte(full)) + suffix := hex.EncodeToString(h[:4]) + keep := maxLen - len(suffix) - 1 + return strings.TrimRight(s[:keep], "-") + "-" + suffix +} + +type shadowResourceIdentity struct { + Name string + Namespace string +} + +func isShadowSameNamespace(pod *v1.Pod) bool { + if pod == nil || pod.Annotations == nil { + return false + } + return pod.Annotations["interlink.eu/shadow-same-ns"] == "true" +} + +func computeShadowResourceIdentity(pod *v1.Pod) (shadowResourceIdentity, error) { + if pod == nil { + return shadowResourceIdentity{}, fmt.Errorf("pod is nil") + } + if pod.Namespace == "" { + return shadowResourceIdentity{}, fmt.Errorf("pod namespace is empty") + } + + var name, namespace string + if isShadowSameNamespace(pod) { + name, namespace = computeShadowResourceNamesForSameNamespace(pod.Name, pod.Namespace) + } else { + name, namespace = computeShadowResourceNames(pod.Name, pod.Namespace) + } + + identity := shadowResourceIdentity{Name: name, Namespace: namespace} + if len(fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) > 63 { + return shadowResourceIdentity{}, fmt.Errorf("shadow ingress hostname label %q exceeds 63 characters; shorten pod/namespace or disable interlink.eu/shadow-same-ns", fmt.Sprintf("%s-%s", identity.Name, identity.Namespace)) + } + + return identity, nil +} + +func computeShadowResourceNamesForSameNamespace(podName, podNamespace string) (resourceBaseName, namespace string) { + // Sanitize namespace and pod name for DNS compliance + sanitizedNamespace := sanitizeDNSName(podNamespace) + sanitizedPodName := sanitizeDNSName(podName) + + // Use the original namespace. Do not truncate it: in same-namespace mode + // resources must be created in the pod's real namespace. + namespace = podNamespace + + // Create a unique resource name to avoid conflicts in the same namespace + resourceBaseName = shadowResourcePrefix + sanitizedPodName + "-" + sanitizedNamespace + // Hash on the unsanitized names: sanitizeDNSName truncates to 63 chars, long pods could still collide + fullBaseName := shadowResourcePrefix + podName + "-" + podNamespace + + // Ensure resourceBaseName doesn't exceed 63 characters + if len(resourceBaseName) > 63 { + // Truncate while keeping some of both names + maxPodNameLen := 28 + maxNsLen := 28 + if len(sanitizedPodName) > maxPodNameLen { + sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) + } + if len(sanitizedNamespace) > maxNsLen { + sanitizedNamespace = sanitizedNamespace[:maxNsLen] + } + resourceBaseName = shadowResourcePrefix + sanitizedPodName + "-" + sanitizedNamespace + resourceBaseName = strings.TrimRight(resourceBaseName, "-") + } + + // Additional check for total length after combining with namespace + ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, namespace) + if len(ingressFirstLabel) > 63 { + maxNameLen := 63 - len(namespace) - 1 + if maxNameLen > 9 && len(resourceBaseName) > maxNameLen { // >9: room for the 8-char hash suffix + resourceBaseName = uniqueTruncate(resourceBaseName, maxNameLen, fullBaseName) + } + } + + return resourceBaseName, namespace +} + +func computeShadowResourceNames(podName, podNamespace string) (resourceBaseName, shadowNamespace string) { + // Sanitize namespace and pod name for DNS compliance + sanitizedNamespace := sanitizeDNSName(podNamespace) + sanitizedPodName := sanitizeDNSName(podName) + + shadowNamespace = sanitizedNamespace + shadowNamespaceSuffix + // Ensure shadowNamespace is valid (max 63 chars for namespace) + if len(shadowNamespace) > 63 { + shadowNamespace = sanitizedNamespace[:min(63-len(shadowNamespaceSuffix), len(sanitizedNamespace))] + shadowNamespaceSuffix + } + + resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace + fullBaseName := podName + "-" + podNamespace + // Ensure resourceBaseName doesn't exceed 63 characters + if len(resourceBaseName) > 63 { + // Truncate while keeping some of both names + maxPodNameLen := 31 + maxNsLen := 31 + if len(sanitizedPodName) > maxPodNameLen { + sanitizedPodName = uniqueTruncate(sanitizedPodName, maxPodNameLen, fullBaseName) + } + if len(sanitizedNamespace) > maxNsLen { + sanitizedNamespace = sanitizedNamespace[:maxNsLen] + } + resourceBaseName = sanitizedPodName + "-" + sanitizedNamespace + resourceBaseName = strings.TrimRight(resourceBaseName, "-") + } + + ingressFirstLabel := fmt.Sprintf("%s-%s", resourceBaseName, shadowNamespace) + if len(ingressFirstLabel) > 63 { + // If combined length exceeds 63, we need to truncate + // Strategy: keep both parts but truncate proportionally + maxNameLen := 31 + maxNsLen := 31 + + truncatedName := resourceBaseName + if len(truncatedName) > maxNameLen { + truncatedName = uniqueTruncate(truncatedName, maxNameLen, fullBaseName) + } + + truncatedNs := shadowNamespace + if len(truncatedNs) > maxNsLen { + truncatedNs = truncatedNs[:maxNsLen] + truncatedNs = strings.TrimRight(truncatedNs, "-") + } + + resourceBaseName = truncatedName + shadowNamespace = truncatedNs + } + + return resourceBaseName, shadowNamespace +} + +// hasShadow reports whether a shadow Deployment is rendered for this pod, either +// because the pod exposes ports over a tunnel or because mesh networking wraps +// every offloaded pod. +func (p *Provider) hasShadow(pod *v1.Pod) bool { + return p.shouldCreateShadow(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) +} + +// shadowNodeConfigMapName returns the ConfigMap carrying the compute node name +// for the given shadow. +func shadowNodeConfigMapName(shadowName string) string { + return shadowName + shadowNodeConfigMapSuffix +} + +// resetShadowNodeConfigMap creates, or blanks, the per-shadow node ConfigMap. +// The shadow is rendered before the remote batch system has allocated anything, +// so the ConfigMap has to exist - and be mountable - while still empty. Blanking +// an existing one stops a previous run's node from being tunnelled to. +func (p *Provider) resetShadowNodeConfigMap(ctx context.Context, identity shadowResourceIdentity) error { + cm := &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: shadowNodeConfigMapName(identity.Name), + Namespace: identity.Namespace, + }, + Data: map[string]string{shadowNodeNameKey: ""}, + } + + _, err := p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Create(ctx, cm, metav1.CreateOptions{}) + if err == nil { + return nil + } + if !apierrors.IsAlreadyExists(err) { + return fmt.Errorf("failed to create shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err) + } + if _, err := p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Update(ctx, cm, metav1.UpdateOptions{}); err != nil { + return fmt.Errorf("failed to reset shadow node configmap %s/%s: %w", identity.Namespace, cm.Name, err) + } + return nil +} + +// publishShadowNodeName records the remote compute node a pod was allocated on +// into its shadow's node ConfigMap, so the shadow can direct its tunnel at the +// right host. The ConfigMap is mounted rather than passed as an env var on +// purpose: kubelet refreshes it in place, whereas changing the pod spec would +// restart the shadow and change the pod IP already reported to Kubernetes as the +// offloaded pod's IP. +// +// Repeated calls with an unchanged value are dropped, so the status loop does not +// write on every poll. +func (p *Provider) publishShadowNodeName(ctx context.Context, pod *v1.Pod, nodeName string) { + if nodeName == "" || !p.hasShadow(pod) { + return + } + if !isValidComputeNodeName(nodeName) { + log.G(ctx).Errorf( + "Refusing to publish compute node %q for %s/%s: not a hostname or IP address. The shadow passes this to ssh, so it is rejected rather than escaped.", + nodeName, pod.Namespace, pod.Name) + return + } + if last, ok := p.shadowNodeNames.Load(string(pod.UID)); ok && last == nodeName { + return + } + + identity, err := computeShadowResourceIdentity(pod) + if err != nil { + log.G(ctx).Warningf("Failed to compute shadow resource identity for %s/%s: %v", pod.Namespace, pod.Name, err) + return + } + + patch, err := json.Marshal(map[string]map[string]string{"data": {shadowNodeNameKey: nodeName}}) + if err != nil { + log.G(ctx).Warningf("Failed to marshal shadow node patch for %s/%s: %v", pod.Namespace, pod.Name, err) + return + } + + name := shadowNodeConfigMapName(identity.Name) + _, err = p.clientSet.CoreV1().ConfigMaps(identity.Namespace).Patch( + ctx, name, k8stypes.StrategicMergePatchType, patch, metav1.PatchOptions{}, + ) + if err != nil { + log.G(ctx).Warningf("Failed to publish compute node %q to shadow configmap %s/%s: %v", nodeName, identity.Namespace, name, err) + return + } + + p.shadowNodeNames.Store(string(pod.UID), nodeName) + log.G(ctx).Infof("Published compute node %q for shadow %s/%s", nodeName, identity.Namespace, identity.Name) +} + +// forgetShadowNodeName drops the cached node name for a pod that is going away, +// so a pod recreated under a new UID starts from a clean slate. +func (p *Provider) forgetShadowNodeName(pod *v1.Pod) { + p.shadowNodeNames.Delete(string(pod.UID)) +} diff --git a/pkg/virtualkubelet/shadow_ssh.go b/pkg/virtualkubelet/shadow_ssh.go new file mode 100644 index 00000000..c76c33bf --- /dev/null +++ b/pkg/virtualkubelet/shadow_ssh.go @@ -0,0 +1,301 @@ +package virtualkubelet + +import ( + "context" + "fmt" + "reflect" + "strings" + "time" + + "github.com/containerd/containerd/log" + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +const ( + // DefaultSSHPort is the login node port used when none is configured. + DefaultSSHPort = 22 + // DefaultSSHKeySecretKey is the key holding the private key inside KeySecret. + DefaultSSHKeySecretKey = "id_ed25519" + // DefaultSSHKeytabSecretKey is the key holding the keytab inside KeytabSecret. + DefaultSSHKeytabSecretKey = "user.keytab" + // DefaultSSHNodeWaitTimeout bounds the wait for the plugin to report a compute + // node. Batch queues routinely make a pod wait hours before it starts. + DefaultSSHNodeWaitTimeout = "2h" +) + +// defaultSSHNodeWait is DefaultSSHNodeWaitTimeout as a duration, so falling back to +// it needs no parsing. TestDefaultSSHNodeWaitMatchesTimeout keeps the two in step. +const defaultSSHNodeWait = 2 * time.Hour + +// defaultSSHTunnelImage returns the image the SSH shadow runs, pinned to this +// virtual kubelet's own version so the two are released together. +func defaultSSHTunnelImage() string { + return "ghcr.io/interlink-hq/interlink/ssh-tunnel:" + KubeletVersion +} + +// isSSHShadow reports whether shadows are rendered as SSH port-forwards. +func (p *Provider) isSSHShadow() bool { + return p.config.Network.ShadowMode == ShadowModeSSH +} + +// NormalizeShadowConfig fills in shadow defaults and rejects combinations that +// cannot work, so a misconfigured deployment fails at startup rather than when the +// first pod with an exposed port shows up. +func NormalizeShadowConfig(config *Config) error { + n := &config.Network + + n.ShadowMode = strings.ToLower(strings.TrimSpace(n.ShadowMode)) + switch n.ShadowMode { + case "": + n.ShadowMode = ShadowModeWstunnel + case ShadowModeWstunnel, ShadowModeSSH: + default: + return fmt.Errorf("unknown Network.ShadowMode %q: expected %q or %q", n.ShadowMode, ShadowModeWstunnel, ShadowModeSSH) + } + + if n.ShadowMode != ShadowModeSSH { + return nil + } + + // Mesh needs the compute node to dial out to the cluster, which is exactly what + // an SSH-only site cannot do. Combining the two is tracked separately. + if n.FullMesh { + return fmt.Errorf("Network.FullMesh is not supported with ShadowMode %q: the SSH shadow only exposes the offloaded pod's ports to the cluster, it does not give the pod access back into it (see interlink-hq/interLink#548)", ShadowModeSSH) + } + + s := &n.SSH + if strings.TrimSpace(s.LoginHost) == "" { + return fmt.Errorf("Network.SSH.LoginHost is required with ShadowMode %q", ShadowModeSSH) + } + if strings.TrimSpace(s.User) == "" { + return fmt.Errorf("Network.SSH.User is required with ShadowMode %q", ShadowModeSSH) + } + + if s.Port == 0 { + s.Port = DefaultSSHPort + } + if s.Image == "" { + s.Image = defaultSSHTunnelImage() + } + if s.NodeWaitTimeout == "" { + s.NodeWaitTimeout = DefaultSSHNodeWaitTimeout + } + if _, err := time.ParseDuration(s.NodeWaitTimeout); err != nil { + return fmt.Errorf("invalid Network.SSH.NodeWaitTimeout %q: %w", s.NodeWaitTimeout, err) + } + if s.ReplicateCredentials == nil { + replicate := true + s.ReplicateCredentials = &replicate + } + + s.ForwardMode = strings.ToLower(strings.TrimSpace(s.ForwardMode)) + switch s.ForwardMode { + case "": + s.ForwardMode = SSHForwardModePortForward + case SSHForwardModePortForward: + case SSHForwardModeExec: + if strings.TrimSpace(s.ExecConnectCommand) == "" { + s.ExecConnectCommand = DefaultSSHExecConnectCommand + } + default: + return fmt.Errorf("unknown Network.SSH.ForwardMode %q: expected %q or %q", s.ForwardMode, SSHForwardModePortForward, SSHForwardModeExec) + } + + s.Auth = strings.ToLower(strings.TrimSpace(s.Auth)) + switch s.Auth { + case "": + s.Auth = SSHAuthPublicKey + fallthrough + case SSHAuthPublicKey: + if strings.TrimSpace(s.KeySecret) == "" { + return fmt.Errorf("Network.SSH.KeySecret is required with Auth %q", SSHAuthPublicKey) + } + if s.KeySecretKey == "" { + s.KeySecretKey = DefaultSSHKeySecretKey + } + case SSHAuthKerberos: + if strings.TrimSpace(s.KeytabSecret) == "" { + return fmt.Errorf("Network.SSH.KeytabSecret is required with Auth %q", SSHAuthKerberos) + } + if strings.TrimSpace(s.Principal) == "" { + return fmt.Errorf("Network.SSH.Principal is required with Auth %q", SSHAuthKerberos) + } + if s.KeytabSecretKey == "" { + s.KeytabSecretKey = DefaultSSHKeytabSecretKey + } + default: + return fmt.Errorf("unknown Network.SSH.Auth %q: expected %q or %q", s.Auth, SSHAuthPublicKey, SSHAuthKerberos) + } + + return nil +} + +// sshNodeWaitSeconds converts the configured wait into whole seconds for the shell +// loop in the template. NormalizeShadowConfig has already validated the duration, +// so the fallback only guards against a Provider built without it. +func (p *Provider) sshNodeWaitSeconds() int { + d, err := time.ParseDuration(p.config.Network.SSH.NodeWaitTimeout) + if err != nil { + d = defaultSSHNodeWait + } + return int(d.Seconds()) +} + +// sshCredentialSecret returns the Secret name the configured auth method needs. +func sshCredentialSecret(s SSHTunnel) string { + if s.Auth == SSHAuthKerberos { + return s.KeytabSecret + } + return s.KeySecret +} + +// replicateShadowCredentials copies the SSH credential Secret and any supporting +// ConfigMaps from the virtual kubelet's own namespace into the shadow's namespace. +// +// Shadows follow the offloaded pod, which for multi-tenant setups means arbitrary +// per-user namespaces; without this every one of them would have to be seeded with +// the credential by hand before offloading could work. +func (p *Provider) replicateShadowCredentials(ctx context.Context, identity shadowResourceIdentity) error { + s := p.config.Network.SSH + if s.ReplicateCredentials == nil || !*s.ReplicateCredentials { + return nil + } + + source := p.config.Namespace + if source == "" || source == identity.Namespace { + return nil + } + + if name := sshCredentialSecret(s); name != "" { + if err := p.replicateSecret(ctx, source, identity.Namespace, name); err != nil { + return err + } + } + for _, name := range []string{s.Krb5ConfigMap, s.KnownHostsConfigMap} { + if name == "" { + continue + } + if err := p.replicateConfigMap(ctx, source, identity.Namespace, name); err != nil { + return err + } + } + return nil +} + +// shadowReplicatedFromAnnotation marks an object as a copy this provider made into +// a shadow namespace. Replication only overwrites objects carrying it, so a Secret +// the namespace's owner already had under the same name is never destroyed. +const shadowReplicatedFromAnnotation = "interlink.eu/replicated-from" + +// replicationTarget describes an object already present in the target namespace. +type replicationTarget struct { + exists bool + annotations map[string]string + sameContent bool +} + +// checkReplicationTarget refuses to overwrite an object that interLink did not put +// there. With interlink.eu/shadow-same-ns the target is the offloaded pod's own +// namespace, which for a multi-tenant cluster is somebody's personal namespace, so +// a name collision would otherwise silently replace their data. +// +// An unmarked object whose content already matches is adopted rather than refused: +// that is what a copy made by a version predating the marker looks like, and +// rewriting it with identical content loses nothing. +func checkReplicationTarget(kind, target, name string, found replicationTarget) error { + if !found.exists || found.annotations[shadowReplicatedFromAnnotation] != "" || found.sameContent { + return nil + } + return fmt.Errorf( + "refusing to overwrite %s %s/%s: it already exists, holds different content and was not created by interLink. "+ + "Rename the credential, or set Network.SSH.ReplicateCredentials to false and provision it yourself", + kind, target, name) +} + +func replicatedMeta(name, target, source string) metav1.ObjectMeta { + return metav1.ObjectMeta{ + Name: name, + Namespace: target, + Annotations: map[string]string{shadowReplicatedFromAnnotation: source}, + } +} + +func (p *Provider) replicateSecret(ctx context.Context, source, target, name string) error { + src, err := p.clientSet.CoreV1().Secrets(source).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("failed to read SSH credential secret %s/%s: %w", source, name, err) + } + + existing, err := p.clientSet.CoreV1().Secrets(target).Get(ctx, name, metav1.GetOptions{}) + if err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("failed to inspect secret %s/%s: %w", target, name, err) + } + found := replicationTarget{exists: err == nil} + if found.exists { + found.annotations = existing.Annotations + found.sameContent = existing.Type == src.Type && reflect.DeepEqual(existing.Data, src.Data) + } + if err := checkReplicationTarget("secret", target, name, found); err != nil { + return err + } + + copied := &v1.Secret{ + ObjectMeta: replicatedMeta(name, target, source), + Type: src.Type, + Data: src.Data, + } + if err := p.applyOrUpdateSecret(ctx, copied); err != nil { + return fmt.Errorf("failed to replicate secret %s into %s: %w", name, target, err) + } + log.G(ctx).Infof("Replicated SSH credential secret %s from %s to %s", name, source, target) + return nil +} + +func (p *Provider) replicateConfigMap(ctx context.Context, source, target, name string) error { + src, err := p.clientSet.CoreV1().ConfigMaps(source).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return fmt.Errorf("failed to read configmap %s/%s: %w", source, name, err) + } + + existing, err := p.clientSet.CoreV1().ConfigMaps(target).Get(ctx, name, metav1.GetOptions{}) + if err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("failed to inspect configmap %s/%s: %w", target, name, err) + } + found := replicationTarget{exists: err == nil} + if found.exists { + found.annotations = existing.Annotations + found.sameContent = reflect.DeepEqual(existing.Data, src.Data) && + reflect.DeepEqual(existing.BinaryData, src.BinaryData) + } + if err := checkReplicationTarget("configmap", target, name, found); err != nil { + return err + } + + copied := &v1.ConfigMap{ + ObjectMeta: replicatedMeta(name, target, source), + Data: src.Data, + BinaryData: src.BinaryData, + } + if err := p.applyOrUpdateConfigMap(ctx, copied); err != nil { + return fmt.Errorf("failed to replicate configmap %s into %s: %w", name, target, err) + } + log.G(ctx).Infof("Replicated configmap %s from %s to %s", name, source, target) + return nil +} + +// warnOnUDPPorts reports ports the SSH shadow cannot carry. ssh -L forwards TCP +// only, so a UDP port is silently unreachable; say so rather than let it look like +// a broken tunnel. +func warnOnUDPPorts(ctx context.Context, ports []PortMapping) { + var udp []string + for _, port := range ports { + if strings.EqualFold(port.Protocol, "UDP") { + udp = append(udp, fmt.Sprintf("%d", port.Port)) + } + } + if len(udp) > 0 { + log.G(ctx).Warningf("SSH shadow cannot forward UDP ports %s: ssh -L is TCP only", strings.Join(udp, ", ")) + } +} diff --git a/pkg/virtualkubelet/shadow_ssh_test.go b/pkg/virtualkubelet/shadow_ssh_test.go new file mode 100644 index 00000000..86c50de4 --- /dev/null +++ b/pkg/virtualkubelet/shadow_ssh_test.go @@ -0,0 +1,438 @@ +package virtualkubelet + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + appsv1 "k8s.io/api/apps/v1" + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/kubernetes/scheme" +) + +const ( + testKeytabSecret = "hpc-keytab" + testPrincipal = "alice@EXAMPLE.ORG" +) + +func sshConfig() Config { + return Config{ + Namespace: "interlink", + Network: Network{ + EnableTunnel: true, + ShadowMode: ShadowModeSSH, + SSH: SSHTunnel{ + LoginHost: "login.hpc.example.org", + User: "alice", + KeySecret: "hpc-ssh-key", + }, + }, + } +} + +func normalized(t *testing.T, config Config) Config { + t.Helper() + assert.NoError(t, NormalizeShadowConfig(&config)) + return config +} + +func TestNormalizeShadowConfigDefaults(t *testing.T) { + t.Run("defaults to the wstunnel shadow", func(t *testing.T) { + config := Config{} + assert.NoError(t, NormalizeShadowConfig(&config)) + assert.Equal(t, ShadowModeWstunnel, config.Network.ShadowMode) + }) + + t.Run("fills in ssh defaults", func(t *testing.T) { + config := normalized(t, sshConfig()) + s := config.Network.SSH + + assert.Equal(t, DefaultSSHPort, s.Port) + assert.Equal(t, SSHAuthPublicKey, s.Auth) + assert.Equal(t, DefaultSSHKeySecretKey, s.KeySecretKey) + assert.Equal(t, DefaultSSHNodeWaitTimeout, s.NodeWaitTimeout) + assert.Equal(t, SSHForwardModePortForward, s.ForwardMode) + assert.Contains(t, s.Image, "ssh-tunnel") + assert.NotNil(t, s.ReplicateCredentials) + assert.True(t, *s.ReplicateCredentials, "credentials replicate by default so per-user namespaces work") + }) + + t.Run("exec forward mode gets a relay command", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.ForwardMode = "EXEC" + + config = normalized(t, config) + assert.Equal(t, SSHForwardModeExec, config.Network.SSH.ForwardMode) + assert.Equal(t, DefaultSSHExecConnectCommand, config.Network.SSH.ExecConnectCommand) + }) + + t.Run("kerberos defaults", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KeySecret = "" + config.Network.SSH.Auth = "KERBEROS" + config.Network.SSH.KeytabSecret = testKeytabSecret + config.Network.SSH.Principal = testPrincipal + + config = normalized(t, config) + assert.Equal(t, SSHAuthKerberos, config.Network.SSH.Auth) + assert.Equal(t, DefaultSSHKeytabSecretKey, config.Network.SSH.KeytabSecretKey) + }) +} + +func TestNormalizeShadowConfigRejects(t *testing.T) { + tests := []struct { + name string + mutate func(*Config) + wantErr string + }{ + { + name: "unknown shadow mode", + mutate: func(c *Config) { c.Network.ShadowMode = "carrier-pigeon" }, + wantErr: "unknown Network.ShadowMode", + }, + { + // The mesh needs the compute node to dial the cluster, which is precisely + // what an ssh-only site cannot do. + name: "ssh together with full mesh", + mutate: func(c *Config) { c.Network.FullMesh = true }, + wantErr: "FullMesh is not supported", + }, + { + name: "missing login host", + mutate: func(c *Config) { c.Network.SSH.LoginHost = "" }, + wantErr: "LoginHost is required", + }, + { + name: "missing user", + mutate: func(c *Config) { c.Network.SSH.User = "" }, + wantErr: "User is required", + }, + { + name: "public key auth without a key secret", + mutate: func(c *Config) { c.Network.SSH.KeySecret = "" }, + wantErr: "KeySecret is required", + }, + { + name: "kerberos without a keytab", + mutate: func(c *Config) { + c.Network.SSH.Auth = SSHAuthKerberos + c.Network.SSH.Principal = testPrincipal + }, + wantErr: "KeytabSecret is required", + }, + { + name: "kerberos without a principal", + mutate: func(c *Config) { + c.Network.SSH.Auth = SSHAuthKerberos + c.Network.SSH.KeytabSecret = testKeytabSecret + }, + wantErr: "Principal is required", + }, + { + name: "unknown auth method", + mutate: func(c *Config) { c.Network.SSH.Auth = "password" }, + wantErr: "unknown Network.SSH.Auth", + }, + { + name: "unparseable node wait timeout", + mutate: func(c *Config) { c.Network.SSH.NodeWaitTimeout = "forever" }, + wantErr: "invalid Network.SSH.NodeWaitTimeout", + }, + { + name: "unknown forward mode", + mutate: func(c *Config) { c.Network.SSH.ForwardMode = "smoke-signal" }, + wantErr: "unknown Network.SSH.ForwardMode", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config := sshConfig() + tt.mutate(&config) + + err := NormalizeShadowConfig(&config) + + assert.Error(t, err, "a misconfiguration must fail at startup, not at first offload") + assert.Contains(t, err.Error(), tt.wantErr) + }) + } +} + +// renderSSHShadow renders the ssh template and returns the manifest plus the +// Deployment decoded from it, using the same decoder applyShadowManifests uses. +func renderSSHShadow(t *testing.T, config Config, ports []PortMapping) (string, *appsv1.Deployment) { + t.Helper() + config = normalized(t, config) + p := &Provider{config: config} + + manifest, err := p.executeShadowTemplate(t.Context(), ShadowTemplateData{ + Name: "shadow-nb-user", + Namespace: "user", + ExposedPorts: ports, + NodeConfigMap: "shadow-nb-user-node", + NodeNameKey: shadowNodeNameKey, + SSH: config.Network.SSH, + SSHNodeWaitSeconds: p.sshNodeWaitSeconds(), + }) + assert.NoError(t, err) + + decoder := serializer.NewCodecFactory(scheme.Scheme).UniversalDeserializer() + var deployment *appsv1.Deployment + for _, doc := range strings.Split(manifest, "---") { + if strings.TrimSpace(doc) == "" { + continue + } + obj, _, err := decoder.Decode([]byte(doc), nil, nil) + assert.NoError(t, err, "every rendered document must decode:\n%s", doc) + if d, ok := obj.(*appsv1.Deployment); ok { + deployment = d + } + } + assert.NotNil(t, deployment, "the ssh template must render a Deployment") + return manifest, deployment +} + +func TestSSHShadowTemplate(t *testing.T) { + tcp := []PortMapping{{Port: 8888, Name: "notebook", Protocol: "TCP"}} + + t.Run("forwards each exposed port to the node the plugin reported", func(t *testing.T) { + manifest, deployment := renderSSHShadow(t, sshConfig(), tcp) + + assert.Contains(t, manifest, `-L "0.0.0.0:8888:$node:8888"`) + assert.Contains(t, manifest, "alice@login.hpc.example.org") + assert.Contains(t, manifest, "-i /interlink/ssh/"+DefaultSSHKeySecretKey) + // The node arrives through a mounted ConfigMap, not the pod spec: restarting + // the shadow would change the pod IP already reported for the offloaded pod. + assert.Contains(t, manifest, "shadow-nb-user-node") + + spec := deployment.Spec.Template.Spec + assert.Equal(t, "linux", spec.NodeSelector["kubernetes.io/os"], + "the shadow must not be scheduled onto the virtual node it shadows") + assert.Empty(t, spec.Tolerations) + }) + + t.Run("waits for the compute node in an init container", func(t *testing.T) { + _, deployment := renderSSHShadow(t, sshConfig(), tcp) + + names := make([]string, 0, len(deployment.Spec.Template.Spec.InitContainers)) + for _, c := range deployment.Spec.Template.Spec.InitContainers { + names = append(names, c.Name) + } + assert.Contains(t, names, "wait-for-node", + "a queued job is the normal case; the shadow shows Init rather than crash-looping") + }) + + t.Run("kerberos renders kinit and no ssh key", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KeySecret = "" + config.Network.SSH.Auth = SSHAuthKerberos + config.Network.SSH.KeytabSecret = testKeytabSecret + config.Network.SSH.Principal = testPrincipal + + manifest, deployment := renderSSHShadow(t, config, tcp) + + assert.Contains(t, manifest, "kinit -k -t /interlink/keytab/"+DefaultSSHKeytabSecretKey+" alice@EXAMPLE.ORG") + assert.Contains(t, manifest, "GSSAPIAuthentication=yes") + assert.NotContains(t, manifest, "/interlink/ssh/") + + var initNames, names []string + for _, c := range deployment.Spec.Template.Spec.InitContainers { + initNames = append(initNames, c.Name) + } + for _, c := range deployment.Spec.Template.Spec.Containers { + names = append(names, c.Name) + } + // kinit runs after the wait so the ticket is fresh however long the queue was, + // and the renewer keeps it alive for the life of the tunnel. + assert.Equal(t, []string{"wait-for-node", "kinit"}, initNames) + assert.Contains(t, names, "kinit-renew") + }) + + t.Run("exec mode relays through the login node instead of forwarding", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.ForwardMode = SSHForwardModeExec + + manifest, deployment := renderSSHShadow(t, config, tcp) + + // No -L anywhere: sites running the exec mode are exactly the ones whose sshd + // refuses to open a forwarded channel at all. + assert.NotContains(t, manifest, "-L \"0.0.0.0:") + assert.Contains(t, manifest, "socat TCP-LISTEN:8888,fork,reuseaddr,bind=0.0.0.0") + // the node is single-quoted for the login node's shell, which re-parses it + assert.Contains(t, manifest, `remote_cmd="$connect_cmd '$node'"`) + // One multiplexed connection, or every request would pay an SSH handshake and + // the login node would see a session per connection. + assert.Contains(t, manifest, "ssh -M -N -o ControlMaster=yes") + assert.Contains(t, manifest, "ControlPath=%s") + + // The Service still fronts the same containerPort, so nothing above the socket + // can tell the two modes apart. + container := deployment.Spec.Template.Spec.Containers[0] + assert.Equal(t, int32(8888), container.Ports[0].ContainerPort) + }) + + t.Run("skips UDP ports, which ssh -L cannot carry", func(t *testing.T) { + manifest, _ := renderSSHShadow(t, sshConfig(), []PortMapping{ + {Port: 8888, Name: "notebook", Protocol: "TCP"}, + {Port: 9999, Name: "telemetry", Protocol: "UDP"}, + }) + + assert.Contains(t, manifest, `-L "0.0.0.0:8888:$node:8888"`) + assert.NotContains(t, manifest, "9999") + }) + + t.Run("pins host keys when a known_hosts ConfigMap is configured", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KnownHostsConfigMap = "hpc-known-hosts" + + manifest, _ := renderSSHShadow(t, config, tcp) + + assert.Contains(t, manifest, "StrictHostKeyChecking=yes") + assert.Contains(t, manifest, "UserKnownHostsFile=/interlink/known-hosts/known_hosts") + assert.NotContains(t, manifest, "accept-new") + }) + + t.Run("falls back to accept-new without one", func(t *testing.T) { + manifest, _ := renderSSHShadow(t, sshConfig(), tcp) + assert.Contains(t, manifest, "StrictHostKeyChecking=accept-new") + }) +} + +func TestReplicateShadowCredentials(t *testing.T) { + const source = "interlink" + target := shadowResourceIdentity{Name: "shadow-nb-user", Namespace: "user"} + + newClient := func() *fake.Clientset { + return fake.NewSimpleClientset( + &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: source}, + Data: map[string][]byte{DefaultSSHKeySecretKey: []byte("PRIVATE KEY")}, + }, + &v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-known-hosts", Namespace: source}, + Data: map[string]string{"known_hosts": "login.hpc.example.org ssh-ed25519 AAAA"}, + }, + ) + } + + t.Run("copies the credential into the shadow namespace", func(t *testing.T) { + config := sshConfig() + config.Network.SSH.KnownHostsConfigMap = "hpc-known-hosts" + client := newClient() + p := &Provider{clientSet: client, config: normalized(t, config)} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), target)) + + secret, err := client.CoreV1().Secrets(target.Namespace).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, []byte("PRIVATE KEY"), secret.Data[DefaultSSHKeySecretKey]) + + cm, err := client.CoreV1().ConfigMaps(target.Namespace).Get(t.Context(), "hpc-known-hosts", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Contains(t, cm.Data["known_hosts"], "ssh-ed25519") + }) + + t.Run("does nothing when replication is switched off", func(t *testing.T) { + config := normalized(t, sshConfig()) + off := false + config.Network.SSH.ReplicateCredentials = &off + client := newClient() + p := &Provider{clientSet: client, config: config} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), target)) + + _, err := client.CoreV1().Secrets(target.Namespace).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.Error(t, err, "the operator opted out; nothing should be copied") + }) + + t.Run("reports a missing source credential instead of rendering a broken shadow", func(t *testing.T) { + p := &Provider{clientSet: fake.NewSimpleClientset(), config: normalized(t, sshConfig())} + + err := p.replicateShadowCredentials(t.Context(), target) + + assert.Error(t, err) + assert.Contains(t, err.Error(), "hpc-ssh-key") + }) + + t.Run("is a no-op when the shadow already lives in the source namespace", func(t *testing.T) { + client := newClient() + p := &Provider{clientSet: client, config: normalized(t, sshConfig())} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), shadowResourceIdentity{Name: "s", Namespace: source})) + }) +} + +// TestDefaultSSHNodeWaitMatchesTimeout keeps the duration fallback in step with the +// string default that ends up in the rendered manifest. +func TestDefaultSSHNodeWaitMatchesTimeout(t *testing.T) { + parsed, err := time.ParseDuration(DefaultSSHNodeWaitTimeout) + assert.NoError(t, err) + assert.Equal(t, parsed, defaultSSHNodeWait) +} + +// TestReplicateCredentialsRefusesToClobber covers the multi-tenant case: with +// interlink.eu/shadow-same-ns the shadow lands in the offloaded pod's own +// namespace, so a Secret the owner already has under the credential's name must +// not be silently replaced with the HPC key. +func TestReplicateCredentialsRefusesToClobber(t *testing.T) { + const source = "interlink" + const target = "alice" + + config := normalized(t, sshConfig()) + credential := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: source}, + Data: map[string][]byte{DefaultSSHKeySecretKey: []byte("hpc-private-key")}, + } + + t.Run("a secret the user already owns is left alone", func(t *testing.T) { + theirs := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: target}, + Data: map[string][]byte{"theirs": []byte("do not lose me")}, + } + client := fake.NewSimpleClientset(credential, theirs) + p := &Provider{clientSet: client, config: config} + + err := p.replicateShadowCredentials(t.Context(), shadowResourceIdentity{Name: "shadow-nb", Namespace: target}) + + assert.ErrorContains(t, err, "refusing to overwrite") + kept, getErr := client.CoreV1().Secrets(target).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, getErr) + assert.Equal(t, []byte("do not lose me"), kept.Data["theirs"]) + }) + + // A copy made by a version that predates the marker has no annotation, but its + // content matches, so upgrading must not start failing every offloaded pod. + t.Run("an unmarked copy with identical content is adopted", func(t *testing.T) { + unmarked := &v1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "hpc-ssh-key", Namespace: target}, + Data: map[string][]byte{DefaultSSHKeySecretKey: []byte("hpc-private-key")}, + } + client := fake.NewSimpleClientset(credential, unmarked) + p := &Provider{clientSet: client, config: config} + + err := p.replicateShadowCredentials(t.Context(), shadowResourceIdentity{Name: "shadow-nb", Namespace: target}) + + assert.NoError(t, err, "an identical copy loses nothing by being rewritten") + adopted, getErr := client.CoreV1().Secrets(target).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, getErr) + assert.Equal(t, source, adopted.Annotations[shadowReplicatedFromAnnotation]) + }) + + t.Run("a copy interLink made earlier is refreshed", func(t *testing.T) { + client := fake.NewSimpleClientset(credential) + p := &Provider{clientSet: client, config: config} + identity := shadowResourceIdentity{Name: "shadow-nb", Namespace: target} + + assert.NoError(t, p.replicateShadowCredentials(t.Context(), identity)) + first, err := client.CoreV1().Secrets(target).Get(t.Context(), "hpc-ssh-key", metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, source, first.Annotations[shadowReplicatedFromAnnotation]) + + // a second offloaded pod in the same namespace must not trip the guard + assert.NoError(t, p.replicateShadowCredentials(t.Context(), identity)) + }) +} diff --git a/pkg/virtualkubelet/shadow_test.go b/pkg/virtualkubelet/shadow_test.go new file mode 100644 index 00000000..bd573db1 --- /dev/null +++ b/pkg/virtualkubelet/shadow_test.go @@ -0,0 +1,219 @@ +package virtualkubelet + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/assert" + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8stypes "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" +) + +// podWithPort is an offloaded pod that exposes a port, which is what makes the VK +// render a shadow for it. +func podWithPort(name, namespace string, uid string) *v1.Pod { + return &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, UID: k8stypes.UID(uid)}, + Spec: v1.PodSpec{ + Containers: []v1.Container{{ + Name: "app", + Ports: []v1.ContainerPort{{ContainerPort: 8888}}, + }}, + }, + } +} + +func tunnelProvider(client *fake.Clientset) *Provider { + return &Provider{clientSet: client, config: Config{Network: Network{EnableTunnel: true}}} +} + +// configMapPatches returns the ConfigMap patch calls recorded by the fake client, +// which is how these tests tell "wrote to the API" from "decided not to". +func configMapPatches(client *fake.Clientset) []k8stesting.Action { + var patches []k8stesting.Action + for _, a := range client.Actions() { + if a.Matches("patch", "configmaps") { + patches = append(patches, a) + } + } + return patches +} + +func TestResetShadowNodeConfigMap(t *testing.T) { + identity := shadowResourceIdentity{Name: "pod-default", Namespace: testNamespaceDefault} + + t.Run("creates the configmap with an empty node so the shadow can mount it before the job is scheduled", func(t *testing.T) { + client := fake.NewSimpleClientset() + p := tunnelProvider(client) + + assert.NoError(t, p.resetShadowNodeConfigMap(t.Context(), identity)) + + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), shadowNodeConfigMapName(identity.Name), metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "", cm.Data[shadowNodeNameKey]) + }) + + t.Run("blanks a leftover node from a previous run", func(t *testing.T) { + client := fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: shadowNodeConfigMapName(identity.Name), Namespace: identity.Namespace}, + Data: map[string]string{shadowNodeNameKey: "stale-node.hpc.example.org"}, + }) + p := tunnelProvider(client) + + assert.NoError(t, p.resetShadowNodeConfigMap(t.Context(), identity)) + + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), shadowNodeConfigMapName(identity.Name), metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "", cm.Data[shadowNodeNameKey], + "a stale node would make the shadow tunnel to a host whose job is gone") + }) +} + +func TestPublishShadowNodeName(t *testing.T) { + pod := podWithPort("nb", testNamespaceDefault, "uid-1") + identity, err := computeShadowResourceIdentity(pod) + assert.NoError(t, err) + cmName := shadowNodeConfigMapName(identity.Name) + + newClient := func() *fake.Clientset { + return fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: cmName, Namespace: identity.Namespace}, + Data: map[string]string{shadowNodeNameKey: ""}, + }) + } + + t.Run("writes the node reported by the plugin", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), cmName, metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "node042.hpc.example.org", cm.Data[shadowNodeNameKey]) + }) + + t.Run("an empty node name leaves the configmap alone while the job is queued", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "") + + assert.Empty(t, configMapPatches(client), "waiting for an allocation must not write to the API") + }) + + t.Run("repeated reports of the same node do not write again", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + before := len(configMapPatches(client)) + for range 5 { + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + } + + assert.Len(t, configMapPatches(client), before, + "the status loop polls continuously; only changes should reach the API") + }) + + t.Run("a pod with no shadow is skipped", func(t *testing.T) { + client := newClient() + p := &Provider{clientSet: client, config: Config{Network: Network{EnableTunnel: false}}} + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + + assert.Empty(t, configMapPatches(client)) + }) + + t.Run("forgetting a pod lets the same node be published again", func(t *testing.T) { + client := newClient() + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + p.forgetShadowNodeName(pod) + p.publishShadowNodeName(t.Context(), pod, "node042.hpc.example.org") + + assert.Len(t, configMapPatches(client), 2) + }) +} + +func TestCleanupShadowResourcesRemovesNodeConfigMap(t *testing.T) { + const name = "pod-default" + ns := testNamespaceDefault + + client := fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: shadowNodeConfigMapName(name), Namespace: ns}, + }) + p := tunnelProvider(client) + + p.cleanupShadowResources(t.Context(), name, ns) + + _, err := client.CoreV1().ConfigMaps(ns).Get(t.Context(), shadowNodeConfigMapName(name), metav1.GetOptions{}) + assert.True(t, apierrors.IsNotFound(err), "the node configmap should be deleted with the rest of the shadow") +} + +// TestIsValidComputeNodeName pins what a plugin is allowed to report. The shadow +// puts this in front of ssh, so the interesting cases are the ones that would +// change the meaning of a command rather than merely look odd. +func TestIsValidComputeNodeName(t *testing.T) { + valid := []string{ + "j18n3", + "j18n3.liza.surf.nl", + "hpc-cloud-test001.cern.ch", + "as01r2b14", + "10.0.0.7", + "fe80::1", + "node_7", + } + for _, name := range valid { + assert.True(t, isValidComputeNodeName(name), "expected %q to be accepted", name) + } + + hostile := map[string]string{ + "whitespace injects an ssh option": "h -oProxyCommand=touch", + "tab does the same": "h\t-oProxyCommand=touch", + "newline": "h\nsecond", + "quote breaks the relay command": `x" ; touch /tmp/pwned ; echo "`, + "command substitution": "x$(id)y", + "backticks": "x`id`y", + "semicolon": "h;id", + "glob": "h*", + "leading hyphen reads as a flag": "-oProxyCommand=touch", + "empty": "", + } + for why, name := range hostile { + assert.False(t, isValidComputeNodeName(name), "expected %q to be refused (%s)", name, why) + } + + assert.False(t, isValidComputeNodeName(strings.Repeat("a", maxComputeNodeNameLen+1)), + "a name longer than a DNS name is not one") +} + +// TestPublishShadowNodeNameRefusesHostileNames makes sure a bad value never reaches +// the ConfigMap the shadow mounts, so neither forward mode can be made to run +// something other than ssh. +func TestPublishShadowNodeNameRefusesHostileNames(t *testing.T) { + pod := podWithPort("nb", testNamespaceDefault, "uid-hostile") + identity, err := computeShadowResourceIdentity(pod) + assert.NoError(t, err) + cmName := shadowNodeConfigMapName(identity.Name) + + for _, nodeName := range []string{"h -oProxyCommand=touch", `x" ; id ; echo "`, "x$(id)y"} { + client := fake.NewSimpleClientset(&v1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: cmName, Namespace: identity.Namespace}, + Data: map[string]string{shadowNodeNameKey: ""}, + }) + p := tunnelProvider(client) + + p.publishShadowNodeName(t.Context(), pod, nodeName) + + assert.Empty(t, configMapPatches(client), "must not patch for %q", nodeName) + cm, err := client.CoreV1().ConfigMaps(identity.Namespace).Get(t.Context(), cmName, metav1.GetOptions{}) + assert.NoError(t, err) + assert.Equal(t, "", cm.Data[shadowNodeNameKey]) + } +} diff --git a/pkg/virtualkubelet/templates/shadow-ssh-template.yaml b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml new file mode 100644 index 00000000..159a016a --- /dev/null +++ b/pkg/virtualkubelet/templates/shadow-ssh-template.yaml @@ -0,0 +1,355 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{.Name}} + namespace: {{.Namespace}} +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/component: {{.Name}} + template: + metadata: + labels: + app.kubernetes.io/component: {{.Name}} + spec: + # Never schedule the shadow onto the virtual node it shadows: real nodes are + # kubernetes.io/os=linux, the virtual one is kubernetes.io/os=virtual-kubelet. + nodeSelector: + kubernetes.io/os: linux + tolerations: [] + initContainers: + # The remote job is usually still queued when the shadow starts, so the pod + # sits in Init until the plugin reports which node it landed on. That is the + # expected path, not a failure: it reads as "waiting for the allocation" + # instead of a crash-looping tunnel. + - name: wait-for-node + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - | + set -eu + deadline=$(( $(date +%s) + {{.SSHNodeWaitSeconds}} )) + while :; do + node=$(tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}} 2>/dev/null || true) + [ -n "$node" ] && break + if [ "$(date +%s)" -ge "$deadline" ]; then + echo "timed out after {{.SSH.NodeWaitTimeout}} waiting for the plugin to report a compute node." >&2 + echo "The plugin must return PodStatus.NodeName once the job runs; plugins that do not are unsupported by the ssh shadow." >&2 + exit 1 + fi + sleep 5 + done + echo "compute node: $node" + volumeMounts: + - name: shadow-node + mountPath: /interlink/node + readOnly: true + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + {{- if eq .SSH.Auth "kerberos" }} + # Runs after the node is known so the ticket is fresh when ssh starts, however + # long the job waited in the queue. + - name: kinit + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - kinit -k -t /interlink/keytab/{{.SSH.KeytabSecretKey}} {{.SSH.Principal}} + env: + - name: KRB5CCNAME + value: FILE:/interlink/krb5cc/ccache + volumeMounts: + - name: shadow-keytab + mountPath: /interlink/keytab + readOnly: true + - name: shadow-krb5cc + mountPath: /interlink/krb5cc + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + mountPath: /etc/krb5.conf + subPath: krb5.conf + readOnly: true + {{- end }} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + {{- end }} + containers: + {{- if eq .SSH.Auth "kerberos" }} + - name: kinit-renew + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - | + set -u + while :; do + sleep 21600 + kinit -k -t /interlink/keytab/{{.SSH.KeytabSecretKey}} {{.SSH.Principal}} || true + done + env: + - name: KRB5CCNAME + value: FILE:/interlink/krb5cc/ccache + volumeMounts: + - name: shadow-keytab + mountPath: /interlink/keytab + readOnly: true + - name: shadow-krb5cc + mountPath: /interlink/krb5cc + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + mountPath: /etc/krb5.conf + subPath: krb5.conf + readOnly: true + {{- end }} + resources: + requests: + cpu: 10m + memory: 32Mi + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + {{- end }} + - name: ssh-forward + image: {{.SSH.Image}} + command: ["/bin/sh", "-c"] + args: + - | + set -u + read_node() { tr -d '[:space:]' < /interlink/node/{{.NodeNameKey}} 2>/dev/null || true; } + + target={{.SSH.User}}@{{.SSH.LoginHost}} + ssh_opts="\ + -p {{.SSH.Port}} \ + -o BatchMode=yes \ + -o ServerAliveInterval=30 \ + -o ServerAliveCountMax=3 \ + {{- if eq .SSH.Auth "kerberos" }} + -o GSSAPIAuthentication=yes \ + -o GSSAPIDelegateCredentials=yes \ + -o PreferredAuthentications=gssapi-with-mic \ + {{- else }} + -i /interlink/ssh/{{.SSH.KeySecretKey}} \ + -o IdentitiesOnly=yes \ + -o PreferredAuthentications=publickey \ + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + -o StrictHostKeyChecking=yes \ + -o UserKnownHostsFile=/interlink/known-hosts/known_hosts \ + {{- else }} + -o StrictHostKeyChecking=accept-new \ + {{- end }} + {{- range .SSH.ExtraOptions }} + -o {{.}} \ + {{- end }} + " + + pids= + + {{- if eq .SSH.ForwardMode "exec" }} + # One local listener per port, each accepted connection relayed by a + # command run on the login node. Nothing here asks sshd to forward + # anything, which is the point: it works where AllowTcpForwarding is off. + forward_dir=/tmp/interlink-forward + control="$forward_dir/master" + connect_cmd='{{.SSH.ExecConnectCommand}}' + + start_forwarders() { + mkdir -p "$forward_dir" + rm -f "$control" + + # Every relayed stream rides one multiplexed SSH connection, so opening a + # connection costs a channel rather than a full handshake, and the login + # node sees a single session instead of one per connection. + # shellcheck disable=SC2086 + ssh -M -N -o ControlMaster=yes -o ControlPath="$control" -o ControlPersist=no $ssh_opts "$target" & + pids=$! + + # Quoted for the login node's shell, which re-parses whatever ssh sends it. + remote_cmd="$connect_cmd '$node'" + + waited=0 + # shellcheck disable=SC2086 + until ssh -O check -o ControlPath="$control" $ssh_opts "$target" >/dev/null 2>&1; do + waited=$((waited + 1)) + if [ "$waited" -ge 30 ]; then + echo "the ssh control master did not come up" >&2 + return 1 + fi + sleep 1 + done + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + + printf '#!/bin/sh\nexec ssh -q -o ControlMaster=auto -o ControlPath=%s %s %s "%s {{.Port}}"\n' \ + "$control" "$ssh_opts" "$target" "$remote_cmd" > "$forward_dir/{{.Port}}" + chmod +x "$forward_dir/{{.Port}}" + socat TCP-LISTEN:{{.Port}},fork,reuseaddr,bind=0.0.0.0 EXEC:"$forward_dir/{{.Port}}" & + pids="$pids $!" + {{- end }} + {{- end }} + } + {{- else }} + # One -L per port. The login node must set AllowTcpForwarding yes for this + # account, or every channel is refused with "administratively prohibited". + start_forwarders() { + # shellcheck disable=SC2086 + ssh -N -o ExitOnForwardFailure=yes $ssh_opts \ + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + -L "0.0.0.0:{{.Port}}:$node:{{.Port}}" \ + {{- end }} + {{- end }} + "$target" & + pids=$! + } + {{- end }} + + all_alive() { + for pid in $pids; do + kill -0 "$pid" 2>/dev/null || return 1 + done + return 0 + } + + stop_forwarders() { + for pid in $pids; do + kill "$pid" 2>/dev/null || true + done + wait 2>/dev/null || true + pids= + } + + node=$(read_node) + [ -n "$node" ] || { echo "no compute node recorded" >&2; exit 1; } + + # A requeued job comes back on a different host and the virtual kubelet + # republishes it into the mounted ConfigMap, but the running forwarders + # still hold the old name. Watch for the change and rebuild them. + while :; do + echo "forwarding{{range .ExposedPorts}}{{if ne .Protocol "UDP"}} {{.Port}}{{end}}{{end}} to $node via {{.SSH.LoginHost}} ({{.SSH.ForwardMode}} mode)" + if ! start_forwarders; then + stop_forwarders + exit 1 + fi + + changed=false + while all_alive; do + current=$(read_node) + if [ -n "$current" ] && [ "$current" != "$node" ]; then + echo "compute node changed from $node to $current, rebuilding the tunnel" + node=$current + changed=true + break + fi + sleep 10 + done + + stop_forwarders + $changed && continue + + current=$(read_node) + if [ -n "$current" ] && [ "$current" != "$node" ]; then + node=$current + continue + fi + echo "the tunnel exited; letting Kubernetes restart the container" >&2 + exit 1 + done + {{- if eq .SSH.Auth "kerberos" }} + env: + - name: KRB5CCNAME + value: FILE:/interlink/krb5cc/ccache + {{- end }} + ports: + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + - containerPort: {{.Port}} + name: {{if .Name}}{{.Name}}{{else}}port-{{.Port}}{{end}} + protocol: {{.Protocol}} + {{- end }} + {{- end }} + volumeMounts: + - name: shadow-node + mountPath: /interlink/node + readOnly: true + {{- if eq .SSH.Auth "kerberos" }} + - name: shadow-krb5cc + mountPath: /interlink/krb5cc + readOnly: true + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + mountPath: /etc/krb5.conf + subPath: krb5.conf + readOnly: true + {{- end }} + {{- else }} + - name: shadow-ssh-key + mountPath: /interlink/ssh + readOnly: true + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + - name: shadow-known-hosts + mountPath: /interlink/known-hosts + readOnly: true + {{- end }} + resources: + requests: + cpu: 100m + memory: 64Mi + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + volumes: + # Mounted as a directory, not subPath: only directory mounts are refreshed in + # place when the VK writes the compute node into this ConfigMap. + - name: shadow-node + configMap: + name: {{.NodeConfigMap}} + {{- if eq .SSH.Auth "kerberos" }} + - name: shadow-keytab + secret: + secretName: {{.SSH.KeytabSecret}} + defaultMode: 0400 + - name: shadow-krb5cc + emptyDir: + medium: Memory + {{- if .SSH.Krb5ConfigMap }} + - name: shadow-krb5-config + configMap: + name: {{.SSH.Krb5ConfigMap}} + {{- end }} + {{- else }} + - name: shadow-ssh-key + secret: + secretName: {{.SSH.KeySecret}} + defaultMode: 0400 + {{- end }} + {{- if .SSH.KnownHostsConfigMap }} + - name: shadow-known-hosts + configMap: + name: {{.SSH.KnownHostsConfigMap}} + {{- end }} +--- +apiVersion: v1 +kind: Service +metadata: + name: {{.Name}} + namespace: {{.Namespace}} +spec: + type: ClusterIP + selector: + app.kubernetes.io/component: {{.Name}} + ports: + {{- range .ExposedPorts }} + {{- if ne .Protocol "UDP" }} + - port: {{.Port}} + targetPort: {{.Port}} + name: {{if .Name}}{{.Name}}{{else}}port-{{.Port}}{{end}} + protocol: {{.Protocol}} + {{- end }} + {{- end }} diff --git a/pkg/virtualkubelet/virtualkubelet.go b/pkg/virtualkubelet/virtualkubelet.go index e55b831b..9ba0591d 100644 --- a/pkg/virtualkubelet/virtualkubelet.go +++ b/pkg/virtualkubelet/virtualkubelet.go @@ -44,8 +44,8 @@ import ( k8stypes "k8s.io/apimachinery/pkg/types" ) -//go:embed templates/wstunnel-template.yaml templates/wstunnel-wireguard-template.yaml -var defaultWstunnelTemplate embed.FS +//go:embed templates/wstunnel-template.yaml templates/wstunnel-wireguard-template.yaml templates/shadow-ssh-template.yaml +var defaultShadowTemplates embed.FS //go:embed all:templates/mesh.sh var meshScriptTemplate embed.FS @@ -86,7 +86,7 @@ const ( annMeshNetworkDisabled = "interlink.eu/mesh-network" // set to "disabled" to opt out of mesh networking ) -type WstunnelTemplateData struct { +type ShadowTemplateData struct { Name string Namespace string RandomPassword string @@ -105,6 +105,17 @@ type WstunnelTemplateData struct { IngressTLS bool IngressClusterIssuer string FullMesh bool + // NodeConfigMap is the ConfigMap the shadow can mount to learn which remote + // compute node the workload landed on. It exists from the moment the shadow is + // created but holds an empty NodeNameKey until the plugin reports the node. + NodeConfigMap string + // NodeNameKey is the key inside NodeConfigMap holding the node name. + NodeNameKey string + // SSH carries the SSH port-forward settings, used by the ssh shadow template. + SSH SSHTunnel + // SSHNodeWaitSeconds is SSH.NodeWaitTimeout in whole seconds, for the shell loop + // that waits on NodeConfigMap. + SSHNodeWaitSeconds int } type PortMapping struct { @@ -147,6 +158,9 @@ type Provider struct { clientSet kubernetes.Interface clientHTTPTransport *http.Transport podIPs []string + // shadowNodeNames caches, per pod UID, the last compute node published to + // that pod's shadow, so the status loop only writes on change. + shadowNodeNames sync.Map } // Increment the given IP address @@ -580,6 +594,10 @@ func LoadConfig(ctx context.Context, providerConfig string) (config Config, err // config = configMap SetDefaultResource(&config) + if err = NormalizeShadowConfig(&config); err != nil { + return config, err + } + if _, err = resource.ParseQuantity(config.Resources.CPU); err != nil { return config, fmt.Errorf("invalid CPU value %v", config.Resources.CPU) } @@ -891,16 +909,16 @@ func copyPodLabelsAndAnnotations(pod *v1.Pod) (map[string]string, map[string]str return labels, annotations } -// createDummyPod creates wstunnel infrastructure from template for containers with exposed ports -func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1.Pod, *WstunnelTemplateData, error) { - log.G(ctx).Infof("Creating wstunnel infrastructure for %s/%s with exposed ports", originalPod.Namespace, originalPod.Name) +// createShadowPod creates shadow infrastructure from template for containers with exposed ports +func (p *Provider) createShadowPod(ctx context.Context, originalPod *v1.Pod) (*v1.Pod, *ShadowTemplateData, error) { + log.G(ctx).Infof("Creating shadow infrastructure for %s/%s with exposed ports", originalPod.Namespace, originalPod.Name) - // If not exists, create the namespace for wstunnel + // If not exists, create the namespace for the shadow if originalPod.Namespace == "" { return nil, nil, fmt.Errorf("pod namespace is empty") } - identity, err := computeWstunnelResourceIdentity(originalPod) + identity, err := computeShadowResourceIdentity(originalPod) if err != nil { return nil, nil, err } @@ -912,7 +930,7 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 _, err = p.clientSet.CoreV1().Namespaces().Get(ctx, identity.Namespace, metav1.GetOptions{}) if err != nil { if !apierrors.IsNotFound(err) { - return nil, nil, fmt.Errorf("failed to get wstunnel namespace %s: %w", identity.Namespace, err) + return nil, nil, fmt.Errorf("failed to get shadow namespace %s: %w", identity.Namespace, err) } // Create the namespace if it doesn't exist ns := &v1.Namespace{ @@ -922,9 +940,9 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 } _, err = p.clientSet.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) if err != nil { - return nil, nil, fmt.Errorf("failed to create wstunnel namespace %s: %w", identity.Namespace, err) + return nil, nil, fmt.Errorf("failed to create shadow namespace %s: %w", identity.Namespace, err) } - log.G(ctx).Infof("Created wstunnel namespace %s", identity.Namespace) + log.G(ctx).Infof("Created shadow namespace %s", identity.Namespace) } // Reuse existing random path prefix if the Deployment already exists, otherwise generate it once @@ -943,6 +961,18 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 // log the path prefix log.G(ctx).Infof("Using wstunnel path prefix %s for %s/%s", pathPrefix, originalPod.Namespace, originalPod.Name) + // The remote node is unknown at this point - the job has not been submitted, let + // alone scheduled - so publish an empty ConfigMap the shadow can already mount. + if err := p.resetShadowNodeConfigMap(ctx, identity); err != nil { + return nil, nil, err + } + + if p.isSSHShadow() { + if err := p.replicateShadowCredentials(ctx, identity); err != nil { + return nil, nil, err + } + } + localContainers := getLocalContainers(originalPod) localInitContainers := getLocalInitContainers(originalPod) @@ -950,7 +980,7 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 log.G(ctx).Infof("Copied %d labels and %d annotations from original pod to shadow pod", len(podLabels), len(podAnnotations)) fullMeshEnabledForPod := p.config.Network.FullMesh && !isMeshNetworkingDisabled(originalPod) - templateData := WstunnelTemplateData{ + templateData := ShadowTemplateData{ Name: identity.Name, Namespace: identity.Namespace, RandomPassword: pathPrefix, @@ -964,6 +994,14 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 IngressTLS: p.config.Network.IngressTLS, IngressClusterIssuer: p.config.Network.IngressClusterIssuer, FullMesh: fullMeshEnabledForPod, + NodeConfigMap: shadowNodeConfigMapName(identity.Name), + NodeNameKey: shadowNodeNameKey, + SSH: p.config.Network.SSH, + SSHNodeWaitSeconds: p.sshNodeWaitSeconds(), + } + + if p.isSSHShadow() { + warnOnUDPPorts(ctx, templateData.ExposedPorts) } log.G(ctx).Debugf("LocalInitContainers count: %d", len(templateData.LocalInitContainers)) @@ -980,22 +1018,22 @@ func (p *Provider) createDummyPod(ctx context.Context, originalPod *v1.Pod) (*v1 } } - manifestYAML, err := p.executeWstunnelTemplate(ctx, templateData) + manifestYAML, err := p.executeShadowTemplate(ctx, templateData) if err != nil { - return nil, nil, fmt.Errorf("failed to execute wstunnel template: %w", err) + return nil, nil, fmt.Errorf("failed to execute shadow template: %w", err) } - createdPod, err := p.applyWstunnelManifests(ctx, manifestYAML, &templateData) + createdPod, err := p.applyShadowManifests(ctx, manifestYAML, &templateData) if err != nil { - return nil, nil, fmt.Errorf("failed to apply wstunnel manifests: %w", err) + return nil, nil, fmt.Errorf("failed to apply shadow manifests: %w", err) } - log.G(ctx).Infof("Created wstunnel infrastructure for %s/%s", originalPod.Namespace, originalPod.Name) + log.G(ctx).Infof("Created shadow infrastructure for %s/%s", originalPod.Namespace, originalPod.Name) return createdPod, &templateData, nil } // setupWireGuardConfig populates WireGuard-related fields on templateData using annotations from the original pod. -func (p *Provider) setupWireGuardConfig(ctx context.Context, originalPod *v1.Pod, templateData *WstunnelTemplateData) error { +func (p *Provider) setupWireGuardConfig(ctx context.Context, originalPod *v1.Pod, templateData *ShadowTemplateData) error { wgMTU := 1280 if v := strings.TrimSpace(originalPod.Annotations[annWGMTU]); v != "" { if n, err := strconv.Atoi(v); err == nil && n > 0 { @@ -1064,8 +1102,8 @@ func mergeMaps(dst, src map[string]string) map[string]string { return dst } -// executeWstunnelTemplate loads and executes the wstunnel template -func (p *Provider) executeWstunnelTemplate(ctx context.Context, data WstunnelTemplateData) (string, error) { +// executeShadowTemplate loads and executes the shadow template +func (p *Provider) executeShadowTemplate(ctx context.Context, data ShadowTemplateData) (string, error) { var templateContent string // Try to load from custom path first @@ -1081,10 +1119,13 @@ func (p *Provider) executeWstunnelTemplate(ctx context.Context, data WstunnelTem // Fall back to embedded template if templateContent == "" { templatePath := "templates/wstunnel-template.yaml" - if data.FullMesh { + switch { + case p.isSSHShadow(): + templatePath = "templates/shadow-ssh-template.yaml" + case data.FullMesh: templatePath = "templates/wstunnel-wireguard-template.yaml" } - content, err := defaultWstunnelTemplate.ReadFile(templatePath) + content, err := defaultShadowTemplates.ReadFile(templatePath) if err != nil { return "", fmt.Errorf("failed to read embedded template: %w", err) } @@ -1092,7 +1133,7 @@ func (p *Provider) executeWstunnelTemplate(ctx context.Context, data WstunnelTem } // Parse and execute template - tmpl, err := template.New("wstunnel").Parse(templateContent) + tmpl, err := template.New("shadow").Parse(templateContent) if err != nil { return "", fmt.Errorf("failed to parse template: %w", err) } @@ -1144,7 +1185,7 @@ func prependContainers(dst []v1.Container, add []v1.Container) []v1.Container { } // applyOrUpdateDeployment creates or updates a Deployment, returning its name and namespace. -func (p *Provider) applyOrUpdateDeployment(ctx context.Context, o *appsv1.Deployment, td *WstunnelTemplateData) (string, string, error) { +func (p *Provider) applyOrUpdateDeployment(ctx context.Context, o *appsv1.Deployment, td *ShadowTemplateData) (string, string, error) { if td != nil { ps := &o.Spec.Template.Spec ps.Volumes = mergeVolumes(ps.Volumes, td.Volumes) @@ -1281,8 +1322,8 @@ func (p *Provider) applyOrUpdateSecret(ctx context.Context, o *v1.Secret) error return nil } -// applyWstunnelManifests applies the generated manifests and returns the first created pod -func (p *Provider) applyWstunnelManifests(ctx context.Context, manifestYAML string, td *WstunnelTemplateData) (*v1.Pod, error) { +// applyShadowManifests applies the generated manifests and returns the first created pod +func (p *Provider) applyShadowManifests(ctx context.Context, manifestYAML string, td *ShadowTemplateData) (*v1.Pod, error) { resources := strings.Split(manifestYAML, "---") decoder := serializer.NewCodecFactory(scheme.Scheme).UniversalDeserializer() var deploymentName string @@ -1305,7 +1346,7 @@ func (p *Provider) applyWstunnelManifests(ctx context.Context, manifestYAML stri case *appsv1.Deployment: name, ns, err := p.applyOrUpdateDeployment(ctx, o, td) if err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } deploymentName = name @@ -1314,28 +1355,28 @@ func (p *Provider) applyWstunnelManifests(ctx context.Context, manifestYAML stri case *v1.Service: if err := p.applyOrUpdateService(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "service:"+o.Name) case *networkingv1.Ingress: if err := p.applyOrUpdateIngress(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "ingress:"+o.Name) case *v1.ConfigMap: if err := p.applyOrUpdateConfigMap(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "configmap:"+o.Name) case *v1.Secret: if err := p.applyOrUpdateSecret(ctx, o); err != nil { - p.cleanupPartialWstunnelResources(ctx, createdResources, o.Namespace) + p.cleanupPartialShadowResources(ctx, createdResources, o.Namespace) return nil, err } createdResources = append(createdResources, "secret:"+o.Name) @@ -1377,57 +1418,66 @@ func (p *Provider) waitForDeploymentPod(ctx context.Context, deploymentName, nam return nil, fmt.Errorf("no pod found for deployment %s within timeout", deploymentName) } -// cleanupWstunnelResources removes all wstunnel resources for a given name and namespace -func (p *Provider) cleanupWstunnelResources(ctx context.Context, wstunnelName, namespace string) { - log.G(ctx).Infof("Cleaning up wstunnel resources for %s/%s", namespace, wstunnelName) +// cleanupShadowResources removes all shadow resources for a given name and namespace +func (p *Provider) cleanupShadowResources(ctx context.Context, shadowName, namespace string) { + log.G(ctx).Infof("Cleaning up shadow resources for %s/%s", namespace, shadowName) // Delete deployment - err := p.clientSet.AppsV1().Deployments(namespace).Delete(ctx, wstunnelName, metav1.DeleteOptions{}) + err := p.clientSet.AppsV1().Deployments(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel deployment %s/%s: %v", namespace, wstunnelName, err) + log.G(ctx).Warningf("Failed to delete shadow deployment %s/%s: %v", namespace, shadowName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel deployment %s/%s", namespace, wstunnelName) + log.G(ctx).Infof("Successfully deleted shadow deployment %s/%s", namespace, shadowName) } // Delete service - err = p.clientSet.CoreV1().Services(namespace).Delete(ctx, wstunnelName, metav1.DeleteOptions{}) + err = p.clientSet.CoreV1().Services(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel service %s/%s: %v", namespace, wstunnelName, err) + log.G(ctx).Warningf("Failed to delete shadow service %s/%s: %v", namespace, shadowName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel service %s/%s", namespace, wstunnelName) + log.G(ctx).Infof("Successfully deleted shadow service %s/%s", namespace, shadowName) } // Delete ingress - err = p.clientSet.NetworkingV1().Ingresses(namespace).Delete(ctx, wstunnelName, metav1.DeleteOptions{}) + err = p.clientSet.NetworkingV1().Ingresses(namespace).Delete(ctx, shadowName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel ingress %s/%s: %v", namespace, wstunnelName, err) + log.G(ctx).Warningf("Failed to delete shadow ingress %s/%s: %v", namespace, shadowName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel ingress %s/%s", namespace, wstunnelName) + log.G(ctx).Infof("Successfully deleted shadow ingress %s/%s", namespace, shadowName) } // Delete configmap - err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, wstunnelName+"-wg-config", metav1.DeleteOptions{}) + err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, shadowName+"-wg-config", metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel configmap %s/%s: %v", namespace, wstunnelName+"-wg-config", err) + log.G(ctx).Warningf("Failed to delete shadow configmap %s/%s: %v", namespace, shadowName+"-wg-config", err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel configmap %s/%s", namespace, wstunnelName+"-wg-config") + log.G(ctx).Infof("Successfully deleted shadow configmap %s/%s", namespace, shadowName+"-wg-config") + } + + // Delete the compute node configmap + nodeCM := shadowNodeConfigMapName(shadowName) + err = p.clientSet.CoreV1().ConfigMaps(namespace).Delete(ctx, nodeCM, metav1.DeleteOptions{}) + if err != nil { + log.G(ctx).Warningf("Failed to delete shadow configmap %s/%s: %v", namespace, nodeCM, err) + } else { + log.G(ctx).Infof("Successfully deleted shadow configmap %s/%s", namespace, nodeCM) } // Delete cert-manager-provisioned TLS secret. if p.config.Network.IngressTLS { - secretName := wstunnelName + "-tls" + secretName := shadowName + "-tls" err = p.clientSet.CoreV1().Secrets(namespace).Delete(ctx, secretName, metav1.DeleteOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to delete wstunnel TLS secret %s/%s: %v", namespace, secretName, err) + log.G(ctx).Warningf("Failed to delete shadow TLS secret %s/%s: %v", namespace, secretName, err) } else { - log.G(ctx).Infof("Successfully deleted wstunnel TLS secret %s/%s", namespace, secretName) + log.G(ctx).Infof("Successfully deleted shadow TLS secret %s/%s", namespace, secretName) } } } -// cleanupPartialWstunnelResources removes specific resources that were created before a failure -func (p *Provider) cleanupPartialWstunnelResources(ctx context.Context, createdResources []string, namespace string) { - log.G(ctx).Infof("Cleaning up partial wstunnel resources in namespace %s", namespace) +// cleanupPartialShadowResources removes specific resources that were created before a failure +func (p *Provider) cleanupPartialShadowResources(ctx context.Context, createdResources []string, namespace string) { + log.G(ctx).Infof("Cleaning up partial shadow resources in namespace %s", namespace) for _, resource := range createdResources { parts := strings.Split(resource, ":") @@ -1602,8 +1652,8 @@ func hasExtraPortsAnnotation(pod *v1.Pod) bool { return exists && strings.TrimSpace(extraPorts) != "" } -// shouldCreateWstunnel checks if wstunnel infrastructure should be created -func (p *Provider) shouldCreateWstunnel(pod *v1.Pod) bool { +// shouldCreateShadow checks if shadow infrastructure should be created +func (p *Provider) shouldCreateShadow(pod *v1.Pod) bool { return p.config.Network.EnableTunnel && (hasExposedPorts(pod) || hasExtraPortsAnnotation(pod)) && pod.Annotations["interlink.eu/pod-vpn"] == "" } @@ -1617,23 +1667,23 @@ func isMeshNetworkingDisabled(pod *v1.Pod) bool { return strings.EqualFold(strings.TrimSpace(pod.Annotations[annMeshNetworkDisabled]), "disabled") } -// handleWstunnelCreation creates wstunnel infrastructure and returns the pod IP -func (p *Provider) handleWstunnelCreation(ctx context.Context, pod *v1.Pod) (string, error) { - identity, err := computeWstunnelResourceIdentity(pod) +// handleShadowCreation creates shadow infrastructure and returns the pod IP +func (p *Provider) handleShadowCreation(ctx context.Context, pod *v1.Pod) (string, error) { + identity, err := computeShadowResourceIdentity(pod) if err != nil { return "", err } - // Create wstunnel infrastructure outside virtual node for port exposure - dummyPod, templateData, err := p.createDummyPod(ctx, pod) + // Create shadow infrastructure outside virtual node for port exposure + shadowPod, templateData, err := p.createShadowPod(ctx, pod) if err != nil { - log.G(ctx).Errorf("Failed to create wstunnel infrastructure for %s/%s: %v", pod.Namespace, pod.Name, err) + log.G(ctx).Errorf("Failed to create shadow infrastructure for %s/%s: %v", pod.Namespace, pod.Name, err) // Clean up any partially created resources - p.cleanupWstunnelResources(ctx, identity.Name, identity.Namespace) - return "", fmt.Errorf("failed to create wstunnel infrastructure for exposed ports: %w", err) + p.cleanupShadowResources(ctx, identity.Name, identity.Namespace) + return "", fmt.Errorf("failed to create shadow infrastructure for exposed ports: %w", err) } - // Wait for wstunnel pod to get an IP with timeout + // Wait for shadow pod to get an IP with timeout timeout := 30 * time.Second // Configurable timeout if timeoutStr := pod.Annotations["interlink.virtual-kubelet.io/wstunnel-timeout"]; timeoutStr != "" { if parsedTimeout, err := time.ParseDuration(timeoutStr); err == nil { @@ -1641,50 +1691,54 @@ func (p *Provider) handleWstunnelCreation(ctx context.Context, pod *v1.Pod) (str } } - podIP, err := p.waitForWstunnelPodIP(ctx, dummyPod, timeout, wstunnelResourceIdentity{ + podIP, err := p.waitForShadowPodIP(ctx, shadowPod, timeout, shadowResourceIdentity{ Name: templateData.Name, Namespace: templateData.Namespace, }) if err != nil { - log.G(ctx).Errorf("Failed to get wstunnel pod IP for %s/%s: %v", pod.Namespace, pod.Name, err) + log.G(ctx).Errorf("Failed to get shadow pod IP for %s/%s: %v", pod.Namespace, pod.Name, err) return "", err } - // Add wstunnel client command annotation to the original pod - if err := p.addWstunnelClientAnnotation(ctx, pod, templateData); err != nil { - log.G(ctx).Warningf("Failed to add wstunnel client annotation to pod %s/%s: %v", pod.Namespace, pod.Name, err) - // Note: We don't clean up here since the wstunnel infrastructure is working, - // just the annotation failed (non-critical) + // The SSH shadow dials in to the login node, so the workload has nothing to set + // up on its side: no wstunnel client to launch, no WireGuard config to apply. + if !p.isSSHShadow() { + // Add wstunnel client command annotation to the original pod + if err := p.addWstunnelClientAnnotation(ctx, pod, templateData); err != nil { + log.G(ctx).Warningf("Failed to add wstunnel client annotation to pod %s/%s: %v", pod.Namespace, pod.Name, err) + // Note: We don't clean up here since the shadow infrastructure is working, + // just the annotation failed (non-critical) + } } return podIP, nil } -// waitForWstunnelPodIP waits for wstunnel pod to get an IP -func (p *Provider) waitForWstunnelPodIP(ctx context.Context, dummyPod *v1.Pod, timeout time.Duration, identity wstunnelResourceIdentity) (string, error) { - log.G(ctx).Infof("Waiting up to %v for wstunnel pod %s/%s to get an IP", timeout, dummyPod.Namespace, dummyPod.Name) +// waitForShadowPodIP waits for shadow pod to get an IP +func (p *Provider) waitForShadowPodIP(ctx context.Context, shadowPod *v1.Pod, timeout time.Duration, identity shadowResourceIdentity) (string, error) { + log.G(ctx).Infof("Waiting up to %v for shadow pod %s/%s to get an IP", timeout, shadowPod.Namespace, shadowPod.Name) start := time.Now() for time.Since(start) < timeout { - updatedDummyPod, err := p.clientSet.CoreV1().Pods(dummyPod.Namespace).Get(ctx, dummyPod.Name, metav1.GetOptions{}) + updatedDummyPod, err := p.clientSet.CoreV1().Pods(shadowPod.Namespace).Get(ctx, shadowPod.Name, metav1.GetOptions{}) if err != nil { - log.G(ctx).Warningf("Failed to get wstunnel pod status: %v", err) + log.G(ctx).Warningf("Failed to get shadow pod status: %v", err) time.Sleep(1 * time.Second) continue } if updatedDummyPod.Status.PodIP != "" { podIP := updatedDummyPod.Status.PodIP - log.G(ctx).Infof("Using wstunnel pod IP %s for virtual pod %s/%s", podIP, identity.Namespace, dummyPod.Name) + log.G(ctx).Infof("Using shadow pod IP %s for virtual pod %s/%s", podIP, identity.Namespace, shadowPod.Name) return podIP, nil } time.Sleep(1 * time.Second) } - // Clean up the wstunnel infrastructure since it didn't get an IP - p.cleanupWstunnelResources(ctx, identity.Name, identity.Namespace) - return "", fmt.Errorf("wstunnel pod %s/%s failed to get an IP within %v timeout", dummyPod.Namespace, dummyPod.Name, timeout) + // Clean up the shadow infrastructure since it didn't get an IP + p.cleanupShadowResources(ctx, identity.Name, identity.Namespace) + return "", fmt.Errorf("shadow pod %s/%s failed to get an IP within %v timeout", shadowPod.Namespace, shadowPod.Name, timeout) } // buildTerminatedContainerStatuses builds a slice of ContainerStatus entries where @@ -1830,10 +1884,10 @@ func (p *Provider) CreatePod(ctx context.Context, pod *v1.Pod) error { podIP := "127.0.0.1" - // Handle wstunnel creation if needed - if p.shouldCreateWstunnel(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { + // Handle shadow creation if needed + if p.hasShadow(pod) { var err error - podIP, err = p.handleWstunnelCreation(ctx, pod) + podIP, err = p.handleShadowCreation(ctx, pod) if err != nil { return err } @@ -1920,14 +1974,15 @@ func (p *Provider) DeletePod(ctx context.Context, pod *v1.Pod) (err error) { return errdefs.NotFound("pod not found") } - // Clean up wstunnel resources if tunnel is enabled and they exist and no VPN annotation - if p.shouldCreateWstunnel(pod) || (p.config.Network.FullMesh && !isMeshNetworkingDisabled(pod)) { - identity, identityErr := computeWstunnelResourceIdentity(pod) + // Clean up shadow resources if tunnel is enabled and they exist and no VPN annotation + if p.hasShadow(pod) { + identity, identityErr := computeShadowResourceIdentity(pod) if identityErr != nil { - log.G(ctx).Warningf("Failed to compute wstunnel resource identity for %s/%s: %v", pod.Namespace, pod.Name, identityErr) + log.G(ctx).Warningf("Failed to compute shadow resource identity for %s/%s: %v", pod.Namespace, pod.Name, identityErr) } else { - p.cleanupWstunnelResources(ctx, identity.Name, identity.Namespace) + p.cleanupShadowResources(ctx, identity.Name, identity.Namespace) } + p.forgetShadowNodeName(pod) } now := metav1.Now() diff --git a/pkg/virtualkubelet/virtualkubelet_mesh_test.go b/pkg/virtualkubelet/virtualkubelet_mesh_test.go index 1d75be4c..4c612ed6 100644 --- a/pkg/virtualkubelet/virtualkubelet_mesh_test.go +++ b/pkg/virtualkubelet/virtualkubelet_mesh_test.go @@ -61,11 +61,11 @@ func TestIsMeshNetworkingDisabled(t *testing.T) { } } -func TestExecuteWstunnelTemplateIngressTLS(t *testing.T) { +func TestExecuteShadowTemplateIngressTLS(t *testing.T) { p := &Provider{} - manifest, err := p.executeWstunnelTemplate(t.Context(), WstunnelTemplateData{ + manifest, err := p.executeShadowTemplate(t.Context(), ShadowTemplateData{ Name: "pod-default", - Namespace: "default-wstunnel", + Namespace: "default-shadow", RandomPassword: testPathPrefix, WildcardDNS: "tunnel.example.com", IngressTLS: true, @@ -74,17 +74,17 @@ func TestExecuteWstunnelTemplateIngressTLS(t *testing.T) { assert.NoError(t, err) assert.Contains(t, manifest, "cert-manager.io/cluster-issuer: lets-issuer") - assert.Contains(t, manifest, "- pod-default-default-wstunnel.tunnel.example.com") - assert.Contains(t, manifest, "host: pod-default-default-wstunnel.tunnel.example.com") + assert.Contains(t, manifest, "- pod-default-default-shadow.tunnel.example.com") + assert.Contains(t, manifest, "host: pod-default-default-shadow.tunnel.example.com") assert.NotContains(t, manifest, "host: ws-pod-default.tunnel.example.com") assert.Equal(t, 1, strings.Count(manifest, "secretName: pod-default-tls")) } -func TestExecuteWstunnelTemplateFullMeshSelectsWireGuardTemplate(t *testing.T) { +func TestExecuteShadowTemplateFullMeshSelectsWireGuardTemplate(t *testing.T) { p := &Provider{} - manifest, err := p.executeWstunnelTemplate(t.Context(), WstunnelTemplateData{ + manifest, err := p.executeShadowTemplate(t.Context(), ShadowTemplateData{ Name: "pod-default", - Namespace: "default-wstunnel", + Namespace: "default-shadow", RandomPassword: testPathPrefix, WildcardDNS: "tunnel.example.com", FullMesh: true, @@ -98,9 +98,9 @@ func TestExecuteWstunnelTemplateFullMeshSelectsWireGuardTemplate(t *testing.T) { assert.Contains(t, manifest, "number: 28080") } -func TestComputeWstunnelResourceIdentityUsesFinalNamespace(t *testing.T) { +func TestComputeShadowResourceIdentityUsesFinalNamespace(t *testing.T) { t.Run("default shadow namespace", func(t *testing.T) { - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "my-pod", Namespace: testNamespaceDefault, @@ -109,11 +109,11 @@ func TestComputeWstunnelResourceIdentityUsesFinalNamespace(t *testing.T) { assert.NoError(t, err) assert.Equal(t, "my-pod-default", identity.Name) - assert.Equal(t, "default-wstunnel", identity.Namespace) + assert.Equal(t, "default-shadow", identity.Namespace) }) t.Run("same namespace keeps original namespace", func(t *testing.T) { - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "my-pod", Namespace: testNamespaceDefault, @@ -124,19 +124,19 @@ func TestComputeWstunnelResourceIdentityUsesFinalNamespace(t *testing.T) { }) assert.NoError(t, err) - assert.Equal(t, "wstunnel-my-pod-default", identity.Name) + assert.Equal(t, "shadow-my-pod-default", identity.Name) assert.Equal(t, testNamespaceDefault, identity.Namespace) }) } -func TestComputeWstunnelResourceIdentitySameNamespaceLongNames(t *testing.T) { +func TestComputeShadowResourceIdentitySameNamespaceLongNames(t *testing.T) { t.Run("long pod name preserves full namespace and stays within 63 chars", func(t *testing.T) { // A real, long-lived namespace that must never be truncated in same-namespace // mode (resources are created in the pod's actual namespace). namespace := strings.Repeat("a", 40) podName := strings.Repeat("b", 80) - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: podName, Namespace: namespace, @@ -160,7 +160,7 @@ func TestComputeWstunnelResourceIdentitySameNamespaceLongNames(t *testing.T) { // label, so the identity must error rather than silently truncate the namespace. namespace := strings.Repeat("a", 62) - identity, err := computeWstunnelResourceIdentity(&v1.Pod{ + identity, err := computeShadowResourceIdentity(&v1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: "web", Namespace: namespace, @@ -188,13 +188,13 @@ func TestGenerateFullMeshScriptIncludesRetryAndReadinessLogic(t *testing.T) { }, } - script, err := p.generateFullMeshScript(t.Context(), &WstunnelTemplateData{ + script, err := p.generateFullMeshScript(t.Context(), &ShadowTemplateData{ RandomPassword: testPathPrefix, WGPrivateKey: serverPriv, ClientPrivateKey: "client-private-key", WGMTU: 1280, KeepaliveSecs: 25, - }, "pod-default-default-wstunnel.tunnel.example.com", "1234567890abcdef") + }, "pod-default-default-shadow.tunnel.example.com", "1234567890abcdef") assert.NoError(t, err) assert.Contains(t, script, "download_with_retry") @@ -202,11 +202,11 @@ func TestGenerateFullMeshScriptIncludesRetryAndReadinessLogic(t *testing.T) { assert.Contains(t, script, "ensure_wstunnel_running") assert.Contains(t, script, "wait_for_wireguard_interface") assert.Contains(t, script, `readiness_protocol="https"`) - assert.Contains(t, script, "$readiness_protocol://pod-default-default-wstunnel.tunnel.example.com:443/path-prefix") - assert.Contains(t, script, "wss://pod-default-default-wstunnel.tunnel.example.com:443") + assert.Contains(t, script, "$readiness_protocol://pod-default-default-shadow.tunnel.example.com:443/path-prefix") + assert.Contains(t, script, "wss://pod-default-default-shadow.tunnel.example.com:443") } -func TestShouldCreateWstunnel(t *testing.T) { +func TestShouldCreateShadow(t *testing.T) { basePod := &v1.Pod{ Spec: v1.PodSpec{ Containers: []v1.Container{ @@ -251,7 +251,7 @@ func TestShouldCreateWstunnel(t *testing.T) { expected: true, }, { - name: "pod vpn annotation disables wstunnel", + name: "pod vpn annotation disables the shadow", network: Network{EnableTunnel: true}, pod: &v1.Pod{ ObjectMeta: metav1.ObjectMeta{ @@ -285,12 +285,12 @@ func TestShouldCreateWstunnel(t *testing.T) { Network: tt.network, }, } - assert.Equal(t, tt.expected, p.shouldCreateWstunnel(tt.pod)) + assert.Equal(t, tt.expected, p.shouldCreateShadow(tt.pod)) }) } } -func TestCleanupWstunnelResources(t *testing.T) { +func TestCleanupShadowResources(t *testing.T) { const ( name = "pod-default" ns = testNamespaceDefault @@ -312,7 +312,7 @@ func TestCleanupWstunnelResources(t *testing.T) { client := newClient() p := &Provider{clientSet: client, config: Config{Network: Network{IngressTLS: true}}} - p.cleanupWstunnelResources(t.Context(), name, ns) + p.cleanupShadowResources(t.Context(), name, ns) _, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{}) assert.True(t, apierrors.IsNotFound(err), "deployment should be deleted") @@ -330,7 +330,7 @@ func TestCleanupWstunnelResources(t *testing.T) { client := newClient() p := &Provider{clientSet: client, config: Config{Network: Network{IngressTLS: false}}} - p.cleanupWstunnelResources(t.Context(), name, ns) + p.cleanupShadowResources(t.Context(), name, ns) // Core resources are still removed regardless of TLS... _, err := client.AppsV1().Deployments(ns).Get(t.Context(), name, metav1.GetOptions{})