feat: add Kubernetes backend (#1260)

Adds a Kubernetes backend. A `name:kubernetes://image` label runs the job as one pod through the Kubernetes API, with services as containers of the same pod. Step env is exported by a script sent over exec stdin, keeping it out of the pod spec and process arguments, and service env goes into a per-job Secret. Docker actions and container options are rejected. See `docs/kubernetes.md`. Operators shape job pods with `kubernetes.pod_template`, with per-label `pod_templates` merged in `runs-on` order and a `$services` entry for service containers. The idle cleanup removes pods and Secrets a runner left behind. The client calls the Kubernetes API directly and runs step commands over `coder/websocket` (same dependency used in gitea) instead of pulling in the heavyweight `kubernetes/client-go`. Includes an e2e test against k3s. Replaces: https://gitea.com/gitea/runner/pulls/1000 Fixes: https://gitea.com/gitea/runner/issues/31 Assisted-by: Claude Code:claude-opus-5-5 Co-authored-by: Zettat123 <[email protected]> Co-authored-by: bircni <[email protected]> Reviewed-on: https://gitea.com/gitea/runner/pulls/1260 Reviewed-by: Lunny Xiao <[email protected]> Reviewed-by: Zettat123 <[email protected]> Co-authored-by: silverwind <[email protected]>
2026-11-13 11:23:42 +00:00 · 2026-10-01 06:59:15 +00:00
1 parent 07ca4bfa7f
commit 0cedd98051
@@ -0,0 +1,400 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package container
+
+import (
+	"bytes"
+	"cmp"
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/coder/websocket"
+	"go.yaml.in/yaml/v4"
+)
+
+const (
+	serviceAccountDir = "/var/run/secrets/kubernetes.io/serviceaccount"
+	execProtocol      = "v5.channel.k8s.io"
+)
+
+type kubeClient struct {
+	server, namespace, token string
+	tokenFile                string // re-read on every request, service account tokens rotate
+	inCluster                bool
+	http                     *http.Client
+}
+
+type apiError struct {
+	code    int
+	message string
+}
+
+func (e *apiError) Error() string { return e.message }
+
+func isNotFound(err error) bool {
+	var apiErr *apiError
+	return errors.As(err, &apiErr) && apiErr.code == http.StatusNotFound
+}
+
+func newKubeClient(kubeconfig, namespace string) (*kubeClient, error) {
+	client := &kubeClient{namespace: namespace}
+	transport := &http.Transport{Proxy: http.ProxyFromEnvironment, TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}}
+	if host := os.Getenv("KUBERNETES_SERVICE_HOST"); host != "" && kubeconfig == "" {
+		if err := client.loadInCluster(host, transport.TLSClientConfig); err != nil {
+			return nil, err
+		}
+	} else if err := client.loadKubeconfig(kubeconfig, transport); err != nil {
+		return nil, err
+	}
+	client.http = &http.Client{Transport: transport}
+	return client, nil
+}
+
+func (c *kubeClient) loadInCluster(host string, tlsConfig *tls.Config) error {
+	c.server = "https://" + net.JoinHostPort(host, os.Getenv("KUBERNETES_SERVICE_PORT"))
+	c.tokenFile = serviceAccountDir + "/token"
+	ca, err := os.ReadFile(serviceAccountDir + "/ca.crt")
+	if err != nil {
+		return err
+	}
+	tlsConfig.RootCAs = x509.NewCertPool()
+	tlsConfig.RootCAs.AppendCertsFromPEM(ca)
+	if c.namespace == "" {
+		namespace, err := os.ReadFile(serviceAccountDir + "/namespace")
+		if err != nil {
+			return err
+		}
+		c.namespace = strings.TrimSpace(string(namespace))
+	}
+	c.inCluster = true
+	return nil
+}
+
+func (c *kubeClient) loadKubeconfig(path string, transport *http.Transport) error {
+	if paths := filepath.SplitList(os.Getenv("KUBECONFIG")); path == "" && len(paths) > 0 {
+		path = paths[0]
+	}
+	if path == "" {
+		home, err := os.UserHomeDir()
+		if err != nil {
+			return err
+		}
+		path = filepath.Join(home, ".kube", "config")
+	}
+	content, err := os.ReadFile(path)
+	if err != nil {
+		return fmt.Errorf("no in-cluster service account and no kubeconfig: %w", err)
+	}
+	var config struct {
+		CurrentContext string `yaml:"current-context"`
+		Contexts       []struct {
+			Name    string
+			Context struct{ Cluster, User, Namespace string }
+		}
+		Clusters []struct {
+			Name    string
+			Cluster struct {
+				Server                   string
+				TLSServerName            string `yaml:"tls-server-name"`
+				InsecureSkipTLSVerify    bool   `yaml:"insecure-skip-tls-verify"`
+				ProxyURL                 string `yaml:"proxy-url"`
+				CertificateAuthority     string `yaml:"certificate-authority"`
+				CertificateAuthorityData string `yaml:"certificate-authority-data"`
+			}
+		}
+		Users []struct {
+			Name string
+			User struct {
+				Token                 string
+				TokenFile             string `yaml:"tokenFile"`
+				ClientCertificate     string `yaml:"client-certificate"`
+				ClientCertificateData string `yaml:"client-certificate-data"`
+				ClientKey             string `yaml:"client-key"`
+				ClientKeyData         string `yaml:"client-key-data"`
+				Exec                  any
+				AuthProvider          any `yaml:"auth-provider"`
+			}
+		}
+	}
+	if err := yaml.Unmarshal(content, &config); err != nil {
+		return fmt.Errorf("parse kubeconfig %s: %w", path, err)
+	}
+	var current struct{ Cluster, User, Namespace string }
+	for _, entry := range config.Contexts {
+		if entry.Name == config.CurrentContext {
+			current = entry.Context
+		}
+	}
+	if current.Cluster == "" {
+		return fmt.Errorf("kubeconfig %s has no current context", path)
+	}
+	c.namespace = cmp.Or(c.namespace, current.Namespace, "default")
+
+	resolve := func(file string) string {
+		if file == "" || filepath.IsAbs(file) {
+			return file
+		}
+		return filepath.Join(filepath.Dir(path), file)
+	}
+	read := func(data, file string) ([]byte, error) {
+		if data != "" {
+			return base64.StdEncoding.DecodeString(data)
+		}
+		if file == "" {
+			return nil, nil
+		}
+		return os.ReadFile(resolve(file))
+	}
+
+	tlsConfig := transport.TLSClientConfig
+	for _, cluster := range config.Clusters {
+		if cluster.Name != current.Cluster {
+			continue
+		}
+		c.server = strings.TrimSuffix(cluster.Cluster.Server, "/")
+		tlsConfig.ServerName, tlsConfig.InsecureSkipVerify = cluster.Cluster.TLSServerName, cluster.Cluster.InsecureSkipTLSVerify
+		if cluster.Cluster.ProxyURL != "" {
+			proxy, err := url.Parse(cluster.Cluster.ProxyURL)
+			if err != nil {
+				return fmt.Errorf("kubeconfig %s: %w", path, err)
+			}
+			transport.Proxy = http.ProxyURL(proxy)
+		}
+		ca, err := read(cluster.Cluster.CertificateAuthorityData, cluster.Cluster.CertificateAuthority)
+		if err != nil {
+			return err
+		}
+		if ca != nil {
+			tlsConfig.RootCAs = x509.NewCertPool()
+			tlsConfig.RootCAs.AppendCertsFromPEM(ca)
+		}
+	}
+	for _, user := range config.Users {
+		if user.Name != current.User {
+			continue
+		}
+		if user.User.Exec != nil || user.User.AuthProvider != nil {
+			return fmt.Errorf("kubeconfig %s: user %q authenticates through a credential plugin, which the runner does not support, use a token or client certificate", path, user.Name)
+		}
+		c.token, c.tokenFile = user.User.Token, resolve(user.User.TokenFile)
+		cert, err := read(user.User.ClientCertificateData, user.User.ClientCertificate)
+		if err != nil {
+			return err
+		}
+		key, err := read(user.User.ClientKeyData, user.User.ClientKey)
+		if err != nil {
+			return err
+		}
+		if cert != nil {
+			pair, err := tls.X509KeyPair(cert, key)
+			if err != nil {
+				return err
+			}
+			tlsConfig.Certificates = []tls.Certificate{pair}
+		}
+	}
+	if c.server == "" {
+		return fmt.Errorf("kubeconfig %s names no server for context %q", path, config.CurrentContext)
+	}
+	return nil
+}
+
+func (c *kubeClient) authorize(header http.Header) error {
+	token := c.token
+	if c.tokenFile != "" {
+		content, err := os.ReadFile(c.tokenFile)
+		if err != nil {
+			return err
+		}
+		token = strings.TrimSpace(string(content))
+	}
+	if token != "" {
+		header.Set("Authorization", "Bearer "+token)
+	}
+	return nil
+}
+
+func (c *kubeClient) path(resource string) string {
+	return "/api/v1/namespaces/" + c.namespace + "/" + resource
+}
+
+func (c *kubeClient) do(ctx context.Context, method, path string, in, out any) error {
+	var body io.Reader
+	if in != nil {
+		content, err := json.Marshal(in)
+		if err != nil {
+			return err
+		}
+		body = bytes.NewReader(content)
+	}
+	req, err := http.NewRequestWithContext(ctx, method, c.server+path, body)
+	if err != nil {
+		return err
+	}
+	if err := c.authorize(req.Header); err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode >= http.StatusMultipleChoices {
+		return statusError(resp)
+	}
+	if out == nil {
+		out = io.Discard // draining lets the connection be reused
+	}
+	if writer, ok := out.(io.Writer); ok {
+		_, err = io.Copy(writer, resp.Body)
+		return err
+	}
+	return json.NewDecoder(resp.Body).Decode(out)
+}
+
+func (c *kubeClient) delete(ctx context.Context, resource, name string) error {
+	if name == "" {
+		return nil
+	}
+	if err := c.do(ctx, http.MethodDelete, c.path(resource+"/"+name), nil, nil); err != nil && !isNotFound(err) {
+		return fmt.Errorf("delete %s %s: %w", resource, name, err)
+	}
+	return nil
+}
+
+func statusError(resp *http.Response) error {
+	content, _ := io.ReadAll(resp.Body)
+	var status struct{ Message string }
+	if json.Unmarshal(content, &status) != nil || status.Message == "" {
+		status.Message = strings.TrimSpace(string(content))
+	}
+	return &apiError{code: resp.StatusCode, message: status.Message}
+}
+
+func (c *kubeClient) exec(ctx context.Context, pod, containerName string, command []string, stdin io.Reader, stdout, stderr io.Writer) error {
+	query := url.Values{"container": {containerName}, "command": command, "stdout": {"true"}, "stderr": {"true"}}
+	if stdin != nil {
+		query.Set("stdin", "true")
+	}
+	header := http.Header{}
+	if err := c.authorize(header); err != nil {
+		return err
+	}
+	ctx, cancel := context.WithCancelCause(ctx)
+	defer cancel(nil)
+	conn, resp, err := websocket.Dial(ctx, c.server+c.path("pods/"+pod+"/exec?")+query.Encode(), &websocket.DialOptions{
+		HTTPClient:   c.http,
+		HTTPHeader:   header,
+		Subprotocols: []string{execProtocol},
+	})
+	if err != nil {
+		if resp != nil && resp.StatusCode >= http.StatusMultipleChoices {
+			defer resp.Body.Close()
+			err = statusError(resp)
+		}
+		return fmt.Errorf("exec in pod %s: %w", pod, err)
+	}
+	defer func() { _ = conn.CloseNow() }()
+	if conn.Subprotocol() != execProtocol {
+		return fmt.Errorf("the Kubernetes API server does not speak %s, which needs Kubernetes 1.30 or later", execProtocol)
+	}
+	conn.SetReadLimit(-1)
+
+	if stdin != nil {
+		go func() {
+			if err := sendStdin(ctx, conn, stdin); err != nil {
+				cancel(err)
+			}
+		}()
+	}
+	go func() {
+		ticker := time.NewTicker(5 * time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				if conn.Ping(ctx) != nil {
+					return
+				}
+			}
+		}
+	}()
+
+	for {
+		_, frame, err := conn.Read(ctx)
+		if err != nil {
+			return cmp.Or(context.Cause(ctx), fmt.Errorf("exec in pod %s ended without an exit status: %w", pod, err))
+		}
+		if len(frame) < 2 {
+			continue
+		}
+		switch frame[0] {
+		case 1:
+			_, err = stdout.Write(frame[1:])
+		case 2:
+			_, err = stderr.Write(frame[1:])
+		case 3:
+			return cmp.Or(context.Cause(ctx), execStatus(frame[1:]))
+		}
+		if err != nil {
+			return err
+		}
+	}
+}
+
+func sendStdin(ctx context.Context, conn *websocket.Conn, stdin io.Reader) error {
+	frame := make([]byte, 1+32*1024) // frame[0] is the stdin channel, 0
+	for {
+		n, err := stdin.Read(frame[1:])
+		if n > 0 {
+			if err := conn.Write(ctx, websocket.MessageBinary, frame[:1+n]); err != nil {
+				return err
+			}
+		}
+		if errors.Is(err, io.EOF) {
+			return conn.Write(ctx, websocket.MessageBinary, []byte{255, 0}) // v5 close signal for stdin
+		}
+		if err != nil {
+			return err
+		}
+	}
+}
+
+func execStatus(content []byte) error {
+	var status struct {
+		Status, Message string
+		Details         struct {
+			Causes []struct{ Reason, Message string }
+		}
+	}
+	if err := json.Unmarshal(content, &status); err != nil {
+		return fmt.Errorf("decode exec status: %w", err)
+	}
+	if status.Status == "Success" {
+		return nil
+	}
+	for _, cause := range status.Details.Causes {
+		if code, err := strconv.Atoi(cause.Message); cause.Reason == "ExitCode" && err == nil {
+			return ExitCodeError(code)
+		}
+	}
+	return fmt.Errorf("exec failed: %s", status.Message)
+}