git-csi-driver

1package main23// arming.go finds the claim a writeable volume is bound to and reads4// the class that arms it. Plan 04 records the answer, and plan 05 acts5// on it.67import (8	"context"9	"fmt"10	"log/slog"11	"time"1213	corev1 "k8s.io/api/core/v1"14	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"15	"k8s.io/client-go/kubernetes"16)1718// defaultResync is how long the driver waits before it reads the claim19// again. It covers a watch that ended and a claim that arrived after20// the volume did.21const defaultResync = 30 * time.Second2223// claimReference is the claim a PersistentVolume is bound to. It labels24// the volume's metrics and takes its Events.25type claimReference struct {26	namespace string27	name      string28}2930// arming reads the cluster for one node's volumes. A driver outside a31// cluster holds no client and arms nothing.32type arming struct {33	node   *node34	client kubernetes.Interface35	logger *slog.Logger36	resync time.Duration37}3839func newArming(answering *node, client kubernetes.Interface, logger *slog.Logger) *arming {40	return &arming{node: answering, client: client, logger: logger, resync: defaultResync}41}4243// arm starts the loop that reads the volume's claim. The caller holds44// the node's lock.45func (n *node) arm(staged *volume) {46	if n.arms.client == nil {47		return48	}49	ctx, cancel := context.WithCancel(n.base)50	n.armings[staged.id] = cancel51	go n.arms.follow(ctx, staged)52}5354// disarm ends that loop and takes the volume off the gauges. The caller55// holds the node's lock.56func (n *node) disarm(staged *volume) {57	cancel, found := n.armings[staged.id]58	if !found {59		return60	}61	delete(n.armings, staged.id)62	cancel()63	n.readings.forget(staged)64}6566// follow reads the claim, then waits for the claim to change or for the67// resync, until the driver stops.68func (a *arming) follow(ctx context.Context, staged *volume) {69	for ctx.Err() == nil {70		a.pass(ctx, staged)71	}72}7374// pass finds the claim, reads the class, and holds a watch open until75// the claim changes or the resync says to start again.76func (a *arming) pass(ctx context.Context, staged *volume) {77	claim, err := a.claimOf(ctx, staged.id)78	if err != nil {79		a.logger.WarnContext(ctx, "the claim was not found",80			"volume", staged.id, "error", err)81		a.rest(ctx)82		return83	}84	a.read(ctx, staged, claim)8586	watching, err := a.client.CoreV1().PersistentVolumeClaims(claim.namespace).87		Watch(ctx, metav1.ListOptions{FieldSelector: "metadata.name=" + claim.name})88	if err != nil {89		a.logger.WarnContext(ctx, "the claim is not watched",90			"claim", claim.namespace+"/"+claim.name, "error", err)91		a.rest(ctx)92		return93	}94	defer watching.Stop()9596	resync := time.NewTimer(a.resync)97	defer resync.Stop()98	for {99		select {100		case <-ctx.Done():101			return102		case <-resync.C:103			return104		case _, open := <-watching.ResultChan():105			if !open {106				return107			}108			a.read(ctx, staged, claim)109		}110	}111}112113// rest waits out the resync after a read that failed, so a claim that114// is not there yet costs one call per resync.115func (a *arming) rest(ctx context.Context) {116	timer := time.NewTimer(a.resync)117	defer timer.Stop()118	select {119	case <-ctx.Done():120	case <-timer.C:121	}122}123124// claimOf finds the claim through the PersistentVolume that carries the125// handle. The kubelet passes the handle and never the object's name, so126// the driver lists the volumes and matches on the handle.127func (a *arming) claimOf(ctx context.Context, handle string) (claimReference, error) {128	volumes, err := a.client.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})129	if err != nil {130		return claimReference{}, err131	}132	for _, held := range volumes.Items {133		source := held.Spec.CSI134		if source == nil || source.Driver != driverName || source.VolumeHandle != handle {135			continue136		}137		if held.Spec.ClaimRef == nil {138			return claimReference{}, fmt.Errorf("the PersistentVolume %s is bound to no claim", held.Name)139		}140		return claimReference{141			namespace: held.Spec.ClaimRef.Namespace,142			name:      held.Spec.ClaimRef.Name,143		}, nil144	}145	return claimReference{}, fmt.Errorf("no PersistentVolume of %s carries the handle %s", driverName, handle)146}147148// read takes the class the claim names and arms the volume when that149// class belongs to this driver.150func (a *arming) read(ctx context.Context, staged *volume, claim claimReference) {151	held, err := a.client.CoreV1().PersistentVolumeClaims(claim.namespace).152		Get(ctx, claim.name, metav1.GetOptions{})153	if err != nil {154		a.logger.WarnContext(ctx, "the claim was not read",155			"claim", claim.namespace+"/"+claim.name, "error", err)156		return157	}158159	name, armed := className(held), false160	if name != "" {161		class, err := a.client.StorageV1().VolumeAttributesClasses().162			Get(ctx, name, metav1.GetOptions{})163		switch {164		case err != nil:165			a.logger.WarnContext(ctx, "the class was not read",166				"class", name, "error", err)167		default:168			armed = class.DriverName == driverName169		}170	}171	a.node.armed(ctx, staged, claim, name, armed)172}173174// className is the class in force: the one the claim's status carries,175// or the one its spec names until a resizer records the modify. Without176// a resizer the status is never filled, so the spec has to count or177// nothing ever arms.178func className(held *corev1.PersistentVolumeClaim) string {179	if current := held.Status.CurrentVolumeAttributesClassName; current != nil && *current != "" {180		return *current181	}182	if asked := held.Spec.VolumeAttributesClassName; asked != nil {183		return *asked184	}185	return ""186}187188// armed records the answer and posts an Event on the pod and the claim189// when the volume moved between armed and unarmed.190func (n *node) armed(191	ctx context.Context, staged *volume, claim claimReference, class string, armed bool,192) {193	if staged.reportArmed(claim, class, armed) {194		reason, message := reasonArmed, fmt.Sprintf("armed by the class %s", class)195		if !armed {196			reason, message = reasonUnarmed, "unarmed: the claim names no class of "+driverName197		}198		n.report(ctx, staged, claim, corev1.EventTypeNormal, reason, message)199	}200	n.readings.record(staged)201}

1package main23// attributes.go reads a volume's attributes and refuses what the driver4// cannot serve, before any git runs.56import (7	"regexp"8	"strconv"9	"strings"10	"time"1112	"github.com/container-storage-interface/spec/lib/go/csi"13	"google.golang.org/grpc/codes"14	"google.golang.org/grpc/status"15)1617// The keys the kubelet adds to the volume context itself. podInfoOnMount18// on the CSIDriver object turns them on.19const (20	ephemeralKey      = "csi.storage.k8s.io/ephemeral"21	podNameKey        = "csi.storage.k8s.io/pod.name"22	podNamespaceKey   = "csi.storage.k8s.io/pod.namespace"23	podUIDKey         = "csi.storage.k8s.io/pod.uid"24	serviceAccountKey = "csi.storage.k8s.io/serviceAccount.name"25)2627// offlinePolicy is what a volume does when the fetch at publish fails.28type offlinePolicy string2930const (31	offlineRefuse     offlinePolicy = "refuse"32	offlineAllowStale offlinePolicy = "allowStale"33)3435// The defaults for a read-only volume.36const (37	defaultRef  = "main"38	defaultPull = 5 * time.Minute39)4041// podReference is the pod the kubelet named, where an Event goes.42type podReference struct {43	name      string44	namespace string45	uid       string46}4748// attributes is one read-only volume as its attributes describe it.49type attributes struct {50	url       string51	ref       string52	pull      time.Duration53	depth     int54	offline   offlinePolicy55	ephemeral bool56	pod       podReference57}5859// parseAttributes reads the volume context. It refuses an unknown60// attribute and a malformed value with InvalidArgument and the61// attribute's name, so the pod's events say what to fix.62func parseAttributes(request *csi.NodePublishVolumeRequest) (*attributes, error) {63	parsed, err := parseVolumeContext(request.GetVolumeContext())64	if err != nil {65		return nil, err66	}67	if parsed.ephemeral && !request.GetReadonly() {68		return nil, status.Error(codes.InvalidArgument,69			"readOnly: an inline volume of this driver has to be read-only")70	}71	return parsed, nil72}7374// readOnlyAttributes are the attributes a read-only volume alone75// accepts. A writeable volume follows its ref at stage and never after.76var readOnlyAttributes = []string{"pull", "depth", "offline"}7778// parseStageAttributes reads a persistent volume's attributes and79// refuses the read-only ones.80func parseStageAttributes(context map[string]string) (*attributes, error) {81	for _, key := range readOnlyAttributes {82		if _, found := context[key]; found {83			return nil, status.Errorf(codes.InvalidArgument,84				"%s: a writeable volume follows its ref at stage alone", key)85		}86	}87	return parseVolumeContext(context)88}8990// parseVolumeContext reads the attributes both a stage call and a91// publish call carry.92func parseVolumeContext(context map[string]string) (*attributes, error) {93	parsed := &attributes{ref: defaultRef, pull: defaultPull, offline: offlineRefuse}9495	for key, value := range context {96		switch key {97		case "url":98			parsed.url = value99		case "ref":100			parsed.ref = value101		case "pull":102			pull, err := parsePull(value)103			if err != nil {104				return nil, err105			}106			parsed.pull = pull107		case "depth":108			depth, err := parseDepth(value)109			if err != nil {110				return nil, err111			}112			parsed.depth = depth113		case "offline":114			switch offlinePolicy(value) {115			case offlineRefuse, offlineAllowStale:116				parsed.offline = offlinePolicy(value)117			default:118				return nil, status.Errorf(codes.InvalidArgument,119					"offline: %q is not refuse or allowStale", value)120			}121		case ephemeralKey:122			parsed.ephemeral = value == "true"123		case podNameKey:124			parsed.pod.name = value125		case podNamespaceKey:126			parsed.pod.namespace = value127		case podUIDKey:128			parsed.pod.uid = value129		case serviceAccountKey:130		default:131			return nil, status.Errorf(codes.InvalidArgument, "%s: unknown attribute", key)132		}133	}134135	if parsed.url == "" {136		return nil, status.Error(codes.InvalidArgument, "url: an attribute the volume must set")137	}138	if err := checkURL(parsed.url); err != nil {139		return nil, err140	}141	if parsed.ref == "" {142		return nil, status.Error(codes.InvalidArgument, "ref: an empty ref names nothing")143	}144	return parsed, nil145}146147// checkURL refuses two URL shapes that would run a command. git accepts148// options after the repository, so a URL that starts with a dash is an149// option. A <transport>::<address> URL names a helper program git runs.150// The driver fetches as root on the node and the URL comes from a pod151// spec, so both are refused before git sees them.152func checkURL(url string) error {153	if strings.HasPrefix(url, "-") {154		return status.Errorf(codes.InvalidArgument, "url: %q reads as an option to git", url)155	}156	if transportHelper.MatchString(url) {157		return status.Errorf(codes.InvalidArgument,158			"url: %q names a transport helper, and the driver runs none", url)159	}160	return nil161}162163// transportHelper matches git's <transport>::<address> form. No https,164// ssh, git, or file URL has two colons after its first word.165var transportHelper = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9+.\-]*::`)166167// parsePull reads pull. never is zero: no fetch after the publish.168func parsePull(value string) (time.Duration, error) {169	if value == "never" {170		return 0, nil171	}172	pull, err := time.ParseDuration(value)173	if err != nil {174		return 0, status.Errorf(codes.InvalidArgument,175			"pull: %q is not a duration or never", value)176	}177	if pull <= 0 {178		return 0, status.Errorf(codes.InvalidArgument,179			"pull: %q is not longer than zero", value)180	}181	return pull, nil182}183184// parseDepth reads depth. Zero is a full clone.185func parseDepth(value string) (int, error) {186	depth, err := strconv.Atoi(value)187	if err != nil || depth < 0 {188		return 0, status.Errorf(codes.InvalidArgument,189			"depth: %q is not a whole number of commits", value)190	}191	return depth, nil192}

1package main23// credentials.go turns a Secret's data into the environment one git4// invocation runs under.56import (7	"os"8	"path/filepath"9	"strings"1011	"google.golang.org/grpc/codes"12	"google.golang.org/grpc/status"13)1415// The keys the driver reads out of a nodePublishSecretRef Secret.16const (17	privateKeyKey = "ssh-privatekey"18	knownHostsKey = "known_hosts"19	tokenKey      = "token"20	usernameKey   = "username"21)2223// defaultUsername is the user a token authenticates as when the Secret24// names none. Forges accept any user with a token; git is the custom.25const defaultUsername = "git"2627// The file names the credentials take under the volume's directory.28const (29	privateKeyFile = "ssh-privatekey"30	knownHostsFile = "known_hosts"31	helperFile     = "credential-helper"32)3334// credentials is what the kubelet passed from the Secret. It stays in35// memory on the volume, so a later fetch can use it, and reaches the36// disk only around a git invocation.37type credentials struct {38	privateKey string39	knownHosts string40	token      string41	username   string42}4344// parseCredentials reads the Secret. No Secret is fine. A Secret with45// neither key is refused, because the person who named it meant one.46func parseCredentials(secrets map[string]string) (*credentials, error) {47	if len(secrets) == 0 {48		return nil, nil49	}50	parsed := &credentials{51		privateKey: secrets[privateKeyKey],52		knownHosts: secrets[knownHostsKey],53		token:      secrets[tokenKey],54		username:   secrets[usernameKey],55	}56	if parsed.privateKey == "" && parsed.token == "" {57		return nil, status.Errorf(codes.InvalidArgument,58			"nodePublishSecretRef: the Secret carries no %s and no %s", privateKeyKey, tokenKey)59	}60	if parsed.username == "" {61		parsed.username = defaultUsername62	}63	return parsed, nil64}6566// use writes the credential files under dir and returns the environment67// that names them and the function that removes them. A private key on68// the node's disk lives no longer than the git invocation that reads it.69// The token goes through a helper script, not the command line, so it70// never appears in the process table.71func (c *credentials) use(dir string) ([]string, func(), error) {72	if c == nil {73		return nil, func() {}, nil74	}75	written := []string{}76	remove := func() {77		for _, path := range written {78			os.Remove(path)79		}80	}8182	var env []string83	if c.privateKey != "" {84		key := filepath.Join(dir, privateKeyFile)85		if err := os.WriteFile(key, []byte(endWithNewline(c.privateKey)), 0o600); err != nil {86			remove()87			return nil, func() {}, err88		}89		written = append(written, key)9091		hosts := filepath.Join(dir, knownHostsFile)92		if err := os.WriteFile(hosts, []byte(c.knownHosts), 0o600); err != nil {93			remove()94			return nil, func() {}, err95		}96		written = append(written, hosts)97		env = append(env, "GIT_SSH_COMMAND="+sshCommand(key, hosts, c.knownHosts != ""))98	}99100	if c.token != "" {101		helper := filepath.Join(dir, helperFile)102		if err := os.WriteFile(helper, []byte(credentialHelper(c.username, c.token)), 0o700); err != nil {103			remove()104			return nil, func() {}, err105		}106		written = append(written, helper)107		env = append(env,108			"GIT_CONFIG_COUNT=1",109			"GIT_CONFIG_KEY_0=credential.helper",110			"GIT_CONFIG_VALUE_0="+helper,111		)112	}113	return env, remove, nil114}115116// sshCommand makes ssh read the key and the hosts file the driver wrote117// and nothing of the node's own. A Secret with known_hosts can check118// the host key, so that one demands a match. Without it, ssh accepts119// the first key it sees and refuses a later change.120func sshCommand(key, hosts string, knownHosts bool) string {121	checking := "accept-new"122	if knownHosts {123		checking = "yes"124	}125	return strings.Join([]string{126		"ssh",127		"-i", quote(key),128		"-o", "IdentitiesOnly=yes",129		"-o", "BatchMode=yes",130		"-o", "UserKnownHostsFile=" + quote(hosts),131		"-o", "StrictHostKeyChecking=" + checking,132	}, " ")133}134135// credentialHelper is the script git runs for a password. git reads the136// answer from its standard output.137func credentialHelper(username, token string) string {138	return strings.Join([]string{139		"#!/bin/sh",140		"cat <<'GIT_CSI_CREDENTIAL'",141		"username=" + username,142		"password=" + token,143		"GIT_CSI_CREDENTIAL",144		"",145	}, "\n")146}147148// quote makes a path safe inside GIT_SSH_COMMAND, which git splits with149// a shell's rules.150func quote(path string) string {151	return "'" + strings.ReplaceAll(path, "'", `'\''`) + "'"152}153154// endWithNewline appends the newline ssh requires at the end of a key155// file. A Secret authored by hand often lacks it.156func endWithNewline(text string) string {157	if strings.HasSuffix(text, "\n") {158		return text159	}160	return text + "\n"161}

1package main23// events.go posts an Event on the pod that mounts a volume. Events are4// what kubectl describe shows, so a refused or stale mount is explained5// where a person looks first.67import (8	"context"9	"log/slog"10	"time"1112	corev1 "k8s.io/api/core/v1"13	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"14	"k8s.io/apimachinery/pkg/types"15	"k8s.io/client-go/kubernetes"16	"k8s.io/client-go/rest"17)1819// The reasons, one per state change a person has to see.20const (21	reasonRefused = "GitVolumeRefused"22	reasonStale   = "GitVolumeStale"23	reasonFailed  = "GitFetchFailed"24	// The three a writeable volume adds: a class armed it, the class left25	// it, and the tree holds work the driver has not committed.26	reasonArmed   = "GitVolumeArmed"27	reasonUnarmed = "GitVolumeUnarmed"28	reasonPending = "GitVolumePending"29)3031// events posts Events through the cluster's API, or posts nothing when32// the driver runs outside a cluster.33type events struct {34	client kubernetes.Interface35	node   string36	logger *slog.Logger37	now    func() time.Time38}3940// newEvents reads the driver's own credentials from the pod it runs in.41func newEvents(nodeID string, logger *slog.Logger) *events {42	return eventsFrom(nodeID, logger, rest.InClusterConfig)43}4445// eventsFrom builds the client from the configuration load returns. A46// driver that finds no cluster still serves volumes and says so once,47// because a mount is worth more than an Event.48func eventsFrom(nodeID string, logger *slog.Logger, load func() (*rest.Config, error)) *events {49	posting := &events{node: nodeID, logger: logger, now: time.Now}50	config, err := load()51	if err != nil {52		logger.Warn("no events", "reason", err)53		return posting54	}55	client, err := kubernetes.NewForConfig(config)56	if err != nil {57		logger.Warn("no events", "reason", err)58		return posting59	}60	posting.client = client61	return posting62}6364// post creates one Event on the pod. A failure to post is logged and65// nothing more, because a mount must never fail on the API server.66func (e *events) post(ctx context.Context, pod podReference, kind, reason, message string) {67	if pod.name == "" || pod.namespace == "" {68		return69	}70	e.create(ctx, corev1.ObjectReference{71		Kind:       "Pod",72		APIVersion: "v1",73		Name:       pod.name,74		Namespace:  pod.namespace,75		UID:        types.UID(pod.uid),76	}, kind, reason, message)77}7879// postClaim creates the same Event on the claim, where a person who80// describes the claim learns whether the volume is armed.81func (e *events) postClaim(ctx context.Context, claim claimReference, kind, reason, message string) {82	if claim.name == "" || claim.namespace == "" {83		return84	}85	e.create(ctx, corev1.ObjectReference{86		Kind:       "PersistentVolumeClaim",87		APIVersion: "v1",88		Name:       claim.name,89		Namespace:  claim.namespace,90	}, kind, reason, message)91}9293// create posts one Event on the object it names.94func (e *events) create(95	ctx context.Context, involved corev1.ObjectReference, kind, reason, message string,96) {97	if e == nil || e.client == nil {98		return99	}100	now := metav1.NewTime(e.now())101	event := &corev1.Event{102		ObjectMeta: metav1.ObjectMeta{103			GenerateName: involved.Name + ".",104			Namespace:    involved.Namespace,105		},106		InvolvedObject: involved,107		Reason:         reason,108		Message:        message,109		Type:           kind,110		Source:         corev1.EventSource{Component: driverName, Host: e.node},111		FirstTimestamp: now,112		LastTimestamp:  now,113		Count:          1,114	}115	if _, err := e.client.CoreV1().Events(involved.Namespace).116		Create(ctx, event, metav1.CreateOptions{}); err != nil {117		e.logger.WarnContext(ctx, "the event was not posted",118			"object", involved.Namespace+"/"+involved.Name, "reason", reason, "error", err)119	}120}121122// report posts one fact in both places a person looks: on the pod that123// mounts the volume and on the claim that binds it.124func (n *node) report(125	ctx context.Context, held *volume, claim claimReference, kind, reason, message string,126) {127	n.events.post(ctx, held.podRef(), kind, reason, message)128	n.events.postClaim(ctx, claim, kind, reason, message)129}

1package main23// follow.go holds the fetch loop that keeps a published tree on the ref4// it follows.56import (7	"context"8	"sync"9	"time"1011	corev1 "k8s.io/api/core/v1"12)1314// follower is one fetch loop for one repository, shared by every volume15// of that URL on this node. The fetch is the repository's work, not the16// volume's, so ten pods on one repository cost one fetch.17type follower struct {18	node       *node19	repository *repository20	cancel     context.CancelFunc21	wake       chan struct{}2223	mu      sync.Mutex24	volumes map[string]*volume25}2627// follow adds a volume to its repository's loop, starting the loop on28// the first volume. The caller holds the node's lock. A volume with29// pull never joins no loop, so a repository every volume pins fetches30// nothing.31func (n *node) follow(mounting *volume) {32	if mounting.attributes.pull == 0 {33		return34	}35	repo := n.store.repository(mounting.attributes.url)36	loop, found := n.followers[repo.name]37	if !found {38		ctx, cancel := context.WithCancel(n.base)39		loop = &follower{40			node:       n,41			repository: repo,42			cancel:     cancel,43			wake:       make(chan struct{}, 1),44			volumes:    map[string]*volume{},45		}46		n.followers[repo.name] = loop47		go loop.run(ctx)48	}49	loop.add(mounting)50}5152// unfollow removes a volume from its loop and stops the loop when the53// last volume of the repository goes. The caller holds the node's lock.54func (n *node) unfollow(published *volume) {55	if published.attributes.pull == 0 {56		return57	}58	repo := n.store.repository(published.attributes.url)59	loop, found := n.followers[repo.name]60	if !found {61		return62	}63	if loop.remove(published) == 0 {64		loop.cancel()65		delete(n.followers, repo.name)66	}67}6869func (f *follower) add(mounting *volume) {70	f.mu.Lock()71	f.volumes[mounting.id] = mounting72	f.mu.Unlock()73	f.nudge()74}7576func (f *follower) remove(published *volume) int {77	f.mu.Lock()78	delete(f.volumes, published.id)79	left := len(f.volumes)80	f.mu.Unlock()81	f.nudge()82	return left83}8485// nudge wakes the loop so it reads the interval again. The channel has86// one slot and the send never blocks, so a publish never waits on a87// loop that is fetching.88func (f *follower) nudge() {89	select {90	case f.wake <- struct{}{}:91	default:92	}93}9495// interval is the shortest pull among the volumes that share the96// repository.97func (f *follower) interval() time.Duration {98	f.mu.Lock()99	defer f.mu.Unlock()100	shortest := time.Duration(0)101	for _, held := range f.volumes {102		if shortest == 0 || held.attributes.pull < shortest {103			shortest = held.attributes.pull104		}105	}106	if shortest == 0 {107		shortest = defaultPull108	}109	return shortest110}111112// run fetches on the interval until the context ends. The context113// descends from the driver's run, so the pod's stop ends every loop.114func (f *follower) run(ctx context.Context) {115	timer := time.NewTimer(f.interval())116	defer timer.Stop()117	for {118		select {119		case <-ctx.Done():120			return121		case <-f.wake:122			timer.Reset(f.interval())123		case <-timer.C:124			f.tick(ctx)125			timer.Reset(f.interval())126		}127	}128}129130// tick is one pass over the volumes of this repository, under the131// repository's lock, so a fetch never races a publish.132func (f *follower) tick(ctx context.Context) {133	defer f.repository.lock()()134	for _, held := range f.snapshot() {135		f.refresh(ctx, held)136	}137}138139func (f *follower) snapshot() []*volume {140	f.mu.Lock()141	defer f.mu.Unlock()142	held := make([]*volume, 0, len(f.volumes))143	for _, one := range f.volumes {144		held = append(held, one)145	}146	return held147}148149// refresh fetches the volume's ref and, when it moved, places the new150// commit in the published tree.151func (f *follower) refresh(ctx context.Context, held *volume) {152	env, remove, err := held.credentials.use(held.directory)153	if err != nil {154		f.trouble(ctx, held, err.Error())155		return156	}157	fetchErr := f.repository.fetch(ctx, env, held.attributes.ref, 0)158	remove()159	if fetchErr != nil {160		f.trouble(ctx, held, fetchErr.Error())161		return162	}163	commit, err := f.repository.resolve(ctx, held.attributes.ref)164	if err != nil {165		f.trouble(ctx, held, err.Error())166		return167	}168	if standing, _ := held.condition(); standing == commit {169		held.reportCommit(commit)170		return171	}172173	if err := f.repository.place(ctx, commit, held.directory, held.tree); err != nil {174		f.trouble(ctx, held, err.Error())175		return176	}177	held.reportCommit(commit)178	f.node.logger.InfoContext(ctx, "the tree moved",179		"volume", held.id, "ref", held.attributes.ref, "commit", short(commit))180}181182// trouble records a failed fetch. The first failure after a success183// posts one Event, and the condition carries the failure until a fetch184// works again.185func (f *follower) trouble(ctx context.Context, held *volume, message string) {186	if held.reportTrouble(message) {187		f.node.events.post(ctx, held.attributes.pod, corev1.EventTypeWarning, reasonFailed, message)188	}189	f.node.logger.WarnContext(ctx, "the fetch failed",190		"volume", held.id, "ref", held.attributes.ref, "error", message)191}

1package main23// git.go holds the one function every git invocation goes through, so4// every call has the same deadline, the same environment, and the same5// error shape.67import (8	"bytes"9	"context"10	"fmt"11	"os"12	"os/exec"13	"strings"14	"time"15)1617// gitDeadline bounds one git invocation. A remote that never answers18// must not hold a kubelet call open forever.19const gitDeadline = 60 * time.Second2021// gitWaitDelay is how long a killed git has to exit before its output22// pipes are abandoned.23const gitWaitDelay = 5 * time.Second2425// gitOutput is what one git invocation answers.26type gitOutput struct {27	stdout string28	stderr string29	code   int30}3132// runGit runs git in dir, with env added to the hermetic environment33// below, under the deadline. The error carries git's own last line of34// stderr, which is where git says why it failed.35func runGit(ctx context.Context, dir string, env []string, args ...string) (gitOutput, error) {36	ctx, cancel := context.WithTimeout(ctx, gitDeadline)37	defer cancel()3839	command := exec.CommandContext(ctx, "git", args...)40	command.Dir = dir41	command.Env = append(gitEnvironment(), env...)42	command.WaitDelay = gitWaitDelay4344	var stdout, stderr bytes.Buffer45	command.Stdout = &stdout46	command.Stderr = &stderr4748	err := command.Run()49	output := gitOutput{stdout: stdout.String(), stderr: stderr.String(), code: -1}50	if command.ProcessState != nil {51		output.code = command.ProcessState.ExitCode()52	}53	if err != nil {54		return output, fmt.Errorf("git %s: %s", strings.Join(args, " "), gitReason(output, err))55	}56	return output, nil57}5859// gitEnvironment is the environment every invocation starts from.60// Nothing of the node's own reaches git: no HOME, no agent socket, no61// global or system configuration, and no prompt, so a fetch behaves the62// same on every node and never waits for a password.63func gitEnvironment() []string {64	return []string{65		"PATH=" + os.Getenv("PATH"),66		"GIT_CONFIG_GLOBAL=/dev/null",67		"GIT_CONFIG_SYSTEM=/dev/null",68		"GIT_TERMINAL_PROMPT=0",69		"LC_ALL=C",70	}71}7273// gitReason is the last non-empty line of stderr, where git states its74// reason, or the exec error when git wrote nothing.75func gitReason(output gitOutput, err error) string {76	lines := strings.Split(strings.TrimSpace(output.stderr), "\n")77	if last := strings.TrimSpace(lines[len(lines)-1]); last != "" {78		return last79	}80	return err.Error()81}

1package main23// identity.go holds the CSI Identity service: the three calls a plugin4// answers before it holds any volume.56import (7	"context"8	"os"910	"github.com/container-storage-interface/spec/lib/go/csi"11	"google.golang.org/protobuf/types/known/wrapperspb"12)1314// driverName is the name every volume and the CSIDriver object use to15// select this driver.16const driverName = "git.liken.sh"1718// identity answers the Identity service. Its only state is the store19// path, because readiness is whether the store takes a write.20type identity struct {21	csi.UnimplementedIdentityServer22	store string23}2425func (i *identity) GetPluginInfo(26	context.Context, *csi.GetPluginInfoRequest,27) (*csi.GetPluginInfoResponse, error) {28	return &csi.GetPluginInfoResponse{Name: driverName, VendorVersion: version}, nil29}3031// GetPluginCapabilities declares nothing. The controller service32// arrives with plan 05, and the driver has no topology, because a33// checkout is made on whichever node publishes it.34func (i *identity) GetPluginCapabilities(35	context.Context, *csi.GetPluginCapabilitiesRequest,36) (*csi.GetPluginCapabilitiesResponse, error) {37	return &csi.GetPluginCapabilitiesResponse{}, nil38}3940// Probe reports ready when the store takes a write. Every repository41// and work tree lives in the store, so a driver that cannot write there42// can do nothing.43func (i *identity) Probe(context.Context, *csi.ProbeRequest) (*csi.ProbeResponse, error) {44	return &csi.ProbeResponse{Ready: wrapperspb.Bool(i.storeIsWriteable())}, nil45}4647// storeIsWriteable creates and removes a file instead of reading the48// directory's mode. A mode says what a user may do, and a write says49// what this process did.50func (i *identity) storeIsWriteable() bool {51	file, err := os.CreateTemp(i.store, ".probe-")52	if err != nil {53		return false54	}55	file.Close()56	os.Remove(file.Name())57	return true58}

1// git-csi-driver is the CSI driver named git.liken.sh. It mounts git2// repositories as volumes. This file holds the command line and the3// run that serves the socket until the pod stops.4package main56import (7	"context"8	"errors"9	"flag"10	"fmt"11	"io"12	"log/slog"13	"os"14	"os/signal"15	"syscall"16)1718// version is the release the binary was built from. The Dockerfile sets19// it with -ldflags "-X main.version=...", and every other build reports20// dev.21var version = "dev"2223func main() {24	os.Exit(run(context.Background(), os.Args[1:], os.Stdout))25}2627// run parses the command line, serves the socket, and returns the exit28// code. The context is the run's life: a signal ends it in the pod, and29// a test ends it the same way.30func run(ctx context.Context, args []string, out io.Writer) int {31	cfg, err := parse(args, out)32	if err != nil {33		return 134	}35	if cfg == nil {36		return 037	}3839	ctx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM)40	defer stop()4142	server, err := newServer(ctx, cfg, slog.Default())43	if err != nil {44		fmt.Fprintln(out, err)45		return 146	}47	if err := server.serve(ctx); err != nil {48		fmt.Fprintln(out, err)49		return 150	}51	return 052}5354// config is what the command line resolves to.55type config struct {56	endpoint string57	nodeID   string58	store    string59	metrics  string60}6162// parse reads the command line and reports every problem to out. When63// the arguments ask for the version alone, it prints it and answers a64// nil config with a nil error, because there is nothing to run.65func parse(args []string, out io.Writer) (*config, error) {66	flags := flag.NewFlagSet("git-csi-driver", flag.ContinueOnError)67	flags.SetOutput(out)6869	endpoint := flags.String("endpoint", "unix:///csi/csi.sock",70		"the address the CSI socket listens on")71	nodeID := flags.String("node-id", "",72		"the name of the node this plugin runs on")73	store := flags.String("store", "/var/lib/liken/pod-storage/git-csi",74		"the directory that holds the repositories and work trees")75	// An empty --metrics serves no metrics.76	metrics := flags.String("metrics", ":9808",77		"the address the metrics listener takes; empty serves none")78	showVersion := flags.Bool("version", false, "print the version and exit")7980	if err := flags.Parse(args); err != nil {81		return nil, err82	}83	if *showVersion {84		fmt.Fprintln(out, version)85		return nil, nil86	}87	if *nodeID == "" {88		err := errors.New("--node-id is required")89		fmt.Fprintln(out, err)90		return nil, err91	}92	return &config{93		endpoint: *endpoint,94		nodeID:   *nodeID,95		store:    *store,96		metrics:  *metrics,97	}, nil98}

1package main23// metrics.go holds the gauges the node plugin exports and the listener4// that serves them. Every fact the driver reports reaches the5// condition, the Events, and these numbers.67import (8	"context"9	"log/slog"10	"net"11	"net/http"12	"time"1314	"github.com/prometheus/client_golang/prometheus"15	"github.com/prometheus/client_golang/prometheus/promhttp"16)1718// metricsDeadline bounds a request's headers and the listener's stop.19const metricsDeadline = 30 * time.Second2021// metrics is the registry the listener serves and the two gauges plan22// 04 fills. Plan 05 adds the rest of the design's list.23type metrics struct {24	registry *prometheus.Registry25	armed    *prometheus.GaugeVec26	pending  *prometheus.GaugeVec27}2829// metricLabels name the claim a person would look up.30var metricLabels = []string{"namespace", "claim"}3132func newMetrics() *metrics {33	readings := &metrics{34		registry: prometheus.NewRegistry(),35		// One when a class of this driver arms the volume, zero when none36		// does.37		armed: prometheus.NewGaugeVec(38			prometheus.GaugeOpts{Name: "git_csi_armed", Help: "One when a class of the driver arms the volume, zero when none does."}, metricLabels),39		// How many paths the last scan found that the driver has not40		// committed.41		pending: prometheus.NewGaugeVec(42			prometheus.GaugeOpts{Name: "git_csi_pending_paths", Help: "Paths the last scan found that the driver has not committed."}, metricLabels),43	}44	readings.registry.MustRegister(readings.armed, readings.pending)45	return readings46}4748// record puts the volume's state on the gauges. A volume whose claim49// the driver has not found has no labels, so it reports nothing yet.50func (m *metrics) record(held *volume) {51	claim, armed, pending := held.reading()52	if m == nil || claim.name == "" {53		return54	}55	value := 0.056	if armed {57		value = 158	}59	m.armed.WithLabelValues(claim.namespace, claim.name).Set(value)60	m.pending.WithLabelValues(claim.namespace, claim.name).Set(float64(pending))61}6263// forget takes a volume off the gauges, so a claim that is gone stops64// being reported.65func (m *metrics) forget(held *volume) {66	claim, _, _ := held.reading()67	if m == nil || claim.name == "" {68		return69	}70	m.armed.DeleteLabelValues(claim.namespace, claim.name)71	m.pending.DeleteLabelValues(claim.namespace, claim.name)72}7374// listen opens the address --metrics names. An empty address serves no75// metrics, which is what a driver under test does.76func (m *metrics) listen(address string) (net.Listener, error) {77	if address == "" {78		return nil, nil79	}80	return net.Listen("tcp", address)81}8283// handler serves the registry at /metrics and nothing else.84func (m *metrics) handler() http.Handler {85	served := http.NewServeMux()86	served.Handle("/metrics", promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{}))87	return served88}8990// serveMetrics answers on the listener until the run ends.91func serveMetrics(ctx context.Context, listener net.Listener, readings *metrics, logger *slog.Logger) {92	serving := &http.Server{93		Handler:           readings.handler(),94		ReadHeaderTimeout: metricsDeadline,95	}96	go func() {97		<-ctx.Done()98		_ = serving.Close()99	}()100	if err := serving.Serve(listener); err != nil && ctx.Err() == nil {101		logger.WarnContext(ctx, "the metrics listener stopped", "error", err)102	}103}

1package main23// mount.go holds the bind mount that puts a tree at the kubelet's4// target path.56import (7	"errors"8	"fmt"910	"golang.org/x/sys/unix"11)1213// mountSyscalls is the two syscalls the driver makes. It is an interface14// so a test without privilege can watch the calls the driver would make.15// Everything else the driver does to a tree is tested for real.16type mountSyscalls interface {17	Mount(source, target, filesystem string, flags uintptr, data string) error18	Unmount(target string, flags int) error19}2021// kernelMounts is the kernel's own syscalls, which the driver uses in22// its pod.23type kernelMounts struct{}2425func (kernelMounts) Mount(source, target, filesystem string, flags uintptr, data string) error {26	return unix.Mount(source, target, filesystem, flags, data)27}2829func (kernelMounts) Unmount(target string, flags int) error {30	return unix.Unmount(target, flags)31}3233// bindReadOnly binds source onto target and makes it read-only. Two34// steps, because the kernel makes the bind first and reads MS_RDONLY35// only on a remount of it. A failed remount unbinds, so a target never36// stays writeable by accident.37func bindReadOnly(calls mountSyscalls, source, target string) error {38	if err := calls.Mount(source, target, "", unix.MS_BIND, ""); err != nil {39		return fmt.Errorf("bind %s onto %s: %w", source, target, err)40	}41	if err := calls.Mount(source, target, "",42		unix.MS_BIND|unix.MS_REMOUNT|unix.MS_RDONLY, ""); err != nil {43		_ = calls.Unmount(target, unix.MNT_DETACH)44		return fmt.Errorf("remount %s read-only: %w", target, err)45	}46	return nil47}4849// unbind detaches the mount at target. A target that holds no mount is50// not an error, so an unpublish the kubelet repeats answers the same51// way twice.52func unbind(calls mountSyscalls, target string) error {53	err := calls.Unmount(target, unix.MNT_DETACH)54	if err == nil || errors.Is(err, unix.EINVAL) || errors.Is(err, unix.ENOENT) {55		return nil56	}57	return fmt.Errorf("unmount %s: %w", target, err)58}5960// bindReadWrite binds source onto target with one call. The pod writes61// the tree, and the driver reads what it wrote.62func bindReadWrite(calls mountSyscalls, source, target string) error {63	if err := calls.Mount(source, target, "", unix.MS_BIND, ""); err != nil {64		return fmt.Errorf("bind %s onto %s: %w", source, target, err)65	}66	return nil67}

1package main23// node.go holds the CSI Node service: the calls the kubelet makes to4// put a volume under a pod and take it away.56import (7	"context"8	"fmt"9	"log/slog"10	"os"11	"path/filepath"12	"strings"13	"sync"14	"time"1516	"github.com/container-storage-interface/spec/lib/go/csi"17	"golang.org/x/sys/unix"18	"google.golang.org/grpc/codes"19	"google.golang.org/grpc/status"20	corev1 "k8s.io/api/core/v1"21)2223// node answers the Node service and holds what this node has published.24type node struct {25	csi.UnimplementedNodeServer26	nodeID string27	store  *store28	mounts mountSyscalls29	events *events30	arms   *arming31	// The gauges every volume reports itself on.32	readings *metrics33	logger   *slog.Logger34	base     context.Context35	// How long a tree rests before the driver reads what is pending, and36	// how often it reads that anyway.37	quiesce time.Duration38	sweep   time.Duration39	// mounted asks the kernel whether a path is still a mount. A restarted40	// driver asks it about every record.41	mounted func(string) bool42	// inotify opens the file the watch reads. It is a seam, so a test can43	// drive a driver whose kernel refused one.44	inotify func(int) (int, error)4546	mu        sync.Mutex47	volumes   map[string]*volume48	followers map[string]*follower49	// The writeable volumes this node has staged, and the loops that watch50	// and arm them.51	staged   map[string]*volume52	watchers map[string]*watcher53	armings  map[string]context.CancelFunc54}5556// newNode builds the service. base is the driver's run, so every fetch57// loop ends when the pod stops.58func newNode(base context.Context, cfg *config, posting *events, readings *metrics, logger *slog.Logger) *node {59	answering := &node{60		nodeID:    cfg.nodeID,61		store:     newStore(cfg.store),62		mounts:    kernelMounts{},63		events:    posting,64		readings:  readings,65		logger:    logger,66		base:      base,67		quiesce:   defaultQuiesce,68		sweep:     defaultSweep,69		mounted:   mountedNow,70		inotify:   unix.InotifyInit1,71		volumes:   map[string]*volume{},72		followers: map[string]*follower{},73		staged:    map[string]*volume{},74		watchers:  map[string]*watcher{},75		armings:   map[string]context.CancelFunc{},76	}77	answering.arms = newArming(answering, posting.client, logger)78	return answering79}8081// NodeGetInfo names the node and no topology. A checkout is made on82// whichever node publishes it, so no node is closer to a volume than83// another.84func (n *node) NodeGetInfo(85	context.Context, *csi.NodeGetInfoRequest,86) (*csi.NodeGetInfoResponse, error) {87	return &csi.NodeGetInfoResponse{NodeId: n.nodeID}, nil88}8990// NodeGetCapabilities declares what the kubelet may ask of this node.91// STAGE_UNSTAGE_VOLUME makes the kubelet stage a persistent volume92// once per node before it publishes it per pod. GET_VOLUME_STATS makes93// it poll NodeGetVolumeStats, and VOLUME_CONDITION makes it read the94// condition in that answer.95func (n *node) NodeGetCapabilities(96	context.Context, *csi.NodeGetCapabilitiesRequest,97) (*csi.NodeGetCapabilitiesResponse, error) {98	declared := []csi.NodeServiceCapability_RPC_Type{99		csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,100		csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,101		csi.NodeServiceCapability_RPC_VOLUME_CONDITION,102	}103	capabilities := make([]*csi.NodeServiceCapability, 0, len(declared))104	for _, rpc := range declared {105		capabilities = append(capabilities, &csi.NodeServiceCapability{106			Type: &csi.NodeServiceCapability_Rpc{107				Rpc: &csi.NodeServiceCapability_RPC{Type: rpc},108			},109		})110	}111	return &csi.NodeGetCapabilitiesResponse{Capabilities: capabilities}, nil112}113114// NodePublishVolume binds a tree under the pod: a read-only volume's115// checkout of the ref, made here, or a writeable volume's work tree,116// made at stage. A repeated call for a published volume answers117// success, because the kubelet retries.118func (n *node) NodePublishVolume(119	ctx context.Context, request *csi.NodePublishVolumeRequest,120) (*csi.NodePublishVolumeResponse, error) {121	id := request.GetVolumeId()122	target := request.GetTargetPath()123	switch {124	case id == "":125		return nil, status.Error(codes.InvalidArgument, "volume_id: the call names no volume")126	case strings.ContainsRune(id, filepath.Separator):127		return nil, status.Error(codes.InvalidArgument, "volume_id: a volume id is one path element")128	case target == "":129		return nil, status.Error(codes.InvalidArgument, "target_path: the call names no path")130	}131132	parsed, err := parseAttributes(request)133	if err != nil {134		n.refused(ctx, podOf(request.GetVolumeContext()), err)135		return nil, err136	}137	holder, err := parseCredentials(request.GetSecrets())138	if err != nil {139		n.refused(ctx, parsed.pod, err)140		return nil, err141	}142	if !parsed.ephemeral {143		if err := n.publishStaged(ctx, request, parsed, holder); err != nil {144			n.refused(ctx, parsed.pod, err)145			return nil, err146		}147		return &csi.NodePublishVolumeResponse{}, nil148	}149150	n.mu.Lock()151	published, found := n.volumes[id]152	n.mu.Unlock()153	if found {154		if published.target != target {155			return nil, status.Errorf(codes.FailedPrecondition,156				"volume_id: %s is published at %s", id, published.target)157		}158		return &csi.NodePublishVolumeResponse{}, nil159	}160161	directory := n.store.volumeDir(id)162	mounting := &volume{163		id:          id,164		attributes:  parsed,165		credentials: holder,166		directory:   directory,167		tree:        filepath.Join(directory, "tree"),168		target:      target,169		pod:         parsed.pod,170	}171	if err := n.publish(ctx, mounting); err != nil {172		n.refused(ctx, parsed.pod, err)173		_ = os.RemoveAll(directory)174		return nil, err175	}176	n.record(ctx, mounting, request.GetVolumeContext())177178	n.mu.Lock()179	n.volumes[id] = mounting180	n.follow(mounting)181	n.mu.Unlock()182	return &csi.NodePublishVolumeResponse{}, nil183}184185// publish makes the volume's directory, stages the ref into it, and186// binds the tree onto the target path.187func (n *node) publish(ctx context.Context, mounting *volume) error {188	if err := os.MkdirAll(mounting.directory, 0o700); err != nil {189		return status.Error(codes.Internal, err.Error())190	}191	if err := n.stage(ctx, mounting); err != nil {192		return err193	}194	if err := os.MkdirAll(mounting.target, 0o755); err != nil {195		return status.Error(codes.Internal, err.Error())196	}197	// A driver that restarted left its mounts behind, so the target198	// comes away before the new bind goes on.199	if err := unbind(n.mounts, mounting.target); err != nil {200		return status.Error(codes.Internal, err.Error())201	}202	if err := bindReadOnly(n.mounts, mounting.tree, mounting.target); err != nil {203		return status.Error(codes.Internal, err.Error())204	}205	return nil206}207208// stage fetches the ref into the shared bare repository and places it209// in the volume's own tree. offline decides what a failed fetch means:210// refuse fails the publish, and allowStale publishes what the store211// holds and reports the failure. A repository the store never fetched212// is refused under both, because there is nothing to publish.213func (n *node) stage(ctx context.Context, mounting *volume) error {214	repo := n.store.repository(mounting.attributes.url)215	defer repo.lock()()216217	first := !repo.exists()218	if first {219		if err := repo.create(ctx); err != nil {220			return status.Error(codes.Internal, err.Error())221		}222	}223	// depth applies to the first fetch of a repository. A later volume224	// with a depth of its own reuses what is there.225	depth := 0226	if first {227		depth = mounting.attributes.depth228	}229230	env, remove, err := mounting.credentials.use(mounting.directory)231	if err != nil {232		return status.Error(codes.Internal, err.Error())233	}234	fetchErr := repo.fetch(ctx, env, mounting.attributes.ref, depth)235	remove()236237	commit, resolveErr := repo.resolve(ctx, mounting.attributes.ref)238	switch {239	case fetchErr != nil && mounting.attributes.offline == offlineRefuse:240		return status.Error(codes.Unavailable, fetchErr.Error())241	case fetchErr != nil && resolveErr != nil:242		return status.Errorf(codes.Unavailable, "%s, and the node holds no copy of %s",243			fetchErr, mounting.attributes.ref)244	case resolveErr != nil:245		return status.Error(codes.Internal, resolveErr.Error())246	}247248	if err := repo.place(ctx, commit, mounting.directory, mounting.tree); err != nil {249		return status.Error(codes.Internal, err.Error())250	}251	mounting.reportCommit(commit)252	if fetchErr != nil {253		mounting.reportTrouble(fetchErr.Error())254		n.events.post(ctx, mounting.attributes.pod, corev1.EventTypeWarning, reasonStale,255			fmt.Sprintf("%s is published from the node's copy at %s: %s",256				mounting.attributes.ref, short(commit), fetchErr))257	}258	return nil259}260261// NodeUnpublishVolume takes the mount away, and a read-only volume's262// directory with it. A writeable volume's work tree stays for the next263// pod on this node, and the bare repository stays for the next volume264// of the same URL.265func (n *node) NodeUnpublishVolume(266	ctx context.Context, request *csi.NodeUnpublishVolumeRequest,267) (*csi.NodeUnpublishVolumeResponse, error) {268	id := request.GetVolumeId()269	target := request.GetTargetPath()270	switch {271	case id == "":272		return nil, status.Error(codes.InvalidArgument, "volume_id: the call names no volume")273	case target == "":274		return nil, status.Error(codes.InvalidArgument, "target_path: the call names no path")275	}276277	n.mu.Lock()278	published, found := n.volumes[id]279	if found {280		delete(n.volumes, id)281		n.unfollow(published)282		n.unwatch(published)283	}284	n.mu.Unlock()285286	if err := unbind(n.mounts, target); err != nil {287		return nil, status.Error(codes.Internal, err.Error())288	}289	if err := os.RemoveAll(target); err != nil {290		return nil, status.Error(codes.Internal, err.Error())291	}292	// A writeable volume gives up only its record of the mount. It stays293	// on the gauges while it is staged, because it is still this node's294	// volume.295	if found && published.writeable {296		n.forget(published)297	}298	if found && !published.writeable {299		if err := os.RemoveAll(published.directory); err != nil {300			return nil, status.Error(codes.Internal, err.Error())301		}302	}303	n.logger.InfoContext(ctx, "unpublished", "volume", id)304	return &csi.NodeUnpublishVolumeResponse{}, nil305}306307// NodeGetVolumeStats reports the tree's size as used. A git volume has308// no free space to report, so available is zero.309//310// The condition is what the volume reports, in its order: trouble, then311// unarmed work, then the commit.312func (n *node) NodeGetVolumeStats(313	_ context.Context, request *csi.NodeGetVolumeStatsRequest,314) (*csi.NodeGetVolumeStatsResponse, error) {315	id := request.GetVolumeId()316	switch {317	case id == "":318		return nil, status.Error(codes.InvalidArgument, "volume_id: the call names no volume")319	case request.GetVolumePath() == "":320		return nil, status.Error(codes.InvalidArgument, "volume_path: the call names no path")321	}322323	n.mu.Lock()324	published, found := n.volumes[id]325	n.mu.Unlock()326	if !found {327		return nil, status.Errorf(codes.NotFound, "volume_id: %s is not published on this node", id)328	}329330	size, err := treeSize(published.tree)331	if err != nil {332		return nil, status.Error(codes.Internal, err.Error())333	}334	abnormal, message := published.report()335	return &csi.NodeGetVolumeStatsResponse{336		Usage: []*csi.VolumeUsage{{337			Unit:      csi.VolumeUsage_BYTES,338			Used:      size,339			Available: 0,340			Total:     size,341		}},342		VolumeCondition: &csi.VolumeCondition{Abnormal: abnormal, Message: message},343	}, nil344}345346func (n *node) NodeExpandVolume(347	context.Context, *csi.NodeExpandVolumeRequest,348) (*csi.NodeExpandVolumeResponse, error) {349	return nil, unimplemented("NodeExpandVolume", "never; git volumes have no size")350}351352// refused posts the refusal on the pod, so a person who describes the353// pod sees why it stays in ContainerCreating.354func (n *node) refused(ctx context.Context, pod podReference, err error) {355	n.events.post(ctx, pod, corev1.EventTypeWarning, reasonRefused, status.Convert(err).Message())356}357358// podOf reads the pod straight from the volume context, so a refused359// parse still knows where its Event goes.360func podOf(context map[string]string) podReference {361	return podReference{362		name:      context[podNameKey],363		namespace: context[podNamespaceKey],364		uid:       context[podUIDKey],365	}366}367368// short is the seven characters git itself prints for a commit.369func short(commit string) string {370	if len(commit) > 7 {371		return commit[:7]372	}373	return commit374}375376// unimplemented answers a call the driver does not serve yet. The377// message names the plan that adds it, so a reader of the log learns378// when the call will work, not only that it does not.379func unimplemented(rpc, when string) error {380	return status.Errorf(codes.Unimplemented, "%s: %s", rpc, when)381}

1package main23// records.go is what a restarted driver reads to find the volumes it4// published. The mounts belong to the kernel and survive the driver;5// its own set does not.67import (8	"bufio"9	"context"10	"encoding/json"11	"io"12	"os"13	"path/filepath"14	"strings"15)1617// recordFile is the file each volume's directory carries beside its18// tree.19const recordFile = "volume.json"2021// record is one published volume as the store holds it. The credentials22// are not in it. A Secret reaches the driver through the kubelet, and23// the node's disk is not where it belongs.24type record struct {25	VolumeID    string            `json:"volumeId"`26	Attributes  map[string]string `json:"attributes"`27	Target      string            `json:"targetPath"`28	Staging     string            `json:"stagingPath,omitempty"`29	Ephemeral   bool              `json:"ephemeral"`30	Credentials bool              `json:"credentials"`31}3233// record writes the volume's own record. A record the driver cannot34// write is logged and nothing more, because a mount is worth more than35// the record of it.36func (n *node) record(ctx context.Context, published *volume, attributes map[string]string) {37	written := &record{38		VolumeID:    published.id,39		Attributes:  attributes,40		Target:      published.target,41		Staging:     published.staging,42		Ephemeral:   !published.writeable,43		Credentials: published.credentials != nil,44	}45	content, err := json.Marshal(written)46	if err == nil {47		err = os.WriteFile(filepath.Join(published.directory, recordFile), content, 0o600)48	}49	if err != nil {50		n.logger.WarnContext(ctx, "the volume's record was not written",51			"volume", published.id, "error", err)52	}53}5455// forget removes the record, so a driver that starts after this56// unpublish does not resume a mount the kubelet took away.57func (n *node) forget(published *volume) {58	if err := os.Remove(filepath.Join(published.directory, recordFile)); err != nil {59		n.logger.Warn("the volume's record was not removed", "volume", published.id, "error", err)60	}61}6263// resume rebuilds the driver's set from the store. A target that is64// still a mount is a volume a pod still reads, so its loops start65// again. A target that is not is a read-only volume the store may drop,66// or a work tree that may hold work nobody has pushed.67func (n *node) resume(ctx context.Context) {68	volumes := filepath.Join(n.store.root, "volumes")69	entries, err := os.ReadDir(volumes)70	if err != nil {71		return72	}73	for _, entry := range entries {74		directory := filepath.Join(volumes, entry.Name())75		held, err := readRecord(filepath.Join(directory, recordFile))76		if err != nil {77			continue78		}79		if !n.mounted(held.Target) {80			n.drop(ctx, held, directory)81			continue82		}83		n.resumeOne(ctx, held, directory)84	}85}8687// drop removes a read-only volume whose mount is gone, because it88// leaves nothing worth keeping. A writeable one keeps its tree until89// plan 06's sweep or a person takes it.90func (n *node) drop(ctx context.Context, held *record, directory string) {91	if !held.Ephemeral {92		return93	}94	if err := os.RemoveAll(directory); err != nil {95		n.logger.WarnContext(ctx, "the volume's directory stayed",96			"volume", held.VolumeID, "error", err)97	}98}99100// resumeOne rebuilds one volume and starts the loops it had. It holds101// no credential: the Secret came with a call the kubelet makes again102// only when the pod restarts, so a fetch that needs one fails and the103// condition says so.104func (n *node) resumeOne(ctx context.Context, held *record, directory string) {105	parsed, err := parseVolumeContext(held.Attributes)106	if err != nil {107		n.logger.WarnContext(ctx, "the volume's record was not read",108			"volume", held.VolumeID, "error", err)109		return110	}111	resumed := &volume{112		id:         held.VolumeID,113		attributes: parsed,114		directory:  directory,115		tree:       filepath.Join(directory, "tree"),116		target:     held.Target,117		staging:    held.Staging,118		writeable:  !held.Ephemeral,119		pod:        parsed.pod,120	}121	if held.Credentials {122		resumed.reportTrouble("the driver restarted and holds no credential for this volume")123	}124125	n.mu.Lock()126	defer n.mu.Unlock()127	n.volumes[resumed.id] = resumed128	if held.Ephemeral {129		n.follow(resumed)130		return131	}132	resumed.work = n.store.workTree(n.store.repository(parsed.url), resumed.id)133	if commit, err := resumed.work.head(ctx); err == nil {134		resumed.commit = commit135	}136	n.staged[resumed.id] = resumed137	n.arm(resumed)138	n.watch(resumed)139}140141// readRecord reads one volume's record.142func readRecord(path string) (*record, error) {143	content, err := os.ReadFile(path)144	if err != nil {145		return nil, err146	}147	held := &record{}148	if err := json.Unmarshal(content, held); err != nil {149		return nil, err150	}151	return held, nil152}153154// mountedNow asks the kernel whether the path is still a mount. Only155// the mount table says what is still there.156func mountedNow(path string) bool {157	table, err := os.Open("/proc/self/mountinfo")158	if err != nil {159		return false160	}161	defer table.Close()162	return mountedIn(table, path)163}164165// mountedIn reads mountinfo. The fifth field of every line is the mount166// point, with a space, a tab, a newline, and a backslash written as167// octal escapes.168func mountedIn(table io.Reader, path string) bool {169	lines := bufio.NewScanner(table)170	for lines.Scan() {171		fields := strings.Fields(lines.Text())172		if len(fields) > 4 && mountEscapes.Replace(fields[4]) == path {173			return true174		}175	}176	return false177}178179var mountEscapes = strings.NewReplacer(`\040`, " ", `\011`, "\t", `\012`, "\n", `\134`, `\`)

1package main23// server.go holds the socket the kubelet connects to, the gRPC server4// that answers on it, and the log line every call writes.56import (7	"context"8	"errors"9	"fmt"10	"io/fs"11	"log/slog"12	"net"13	"os"14	"strings"1516	"github.com/container-storage-interface/spec/lib/go/csi"17	"google.golang.org/grpc"18	"google.golang.org/grpc/status"19)2021// endpointScheme is the only scheme --endpoint accepts. The kubelet22// reaches a CSI plugin through a Unix socket in its plugins directory,23// and nothing else connects to this server.24const endpointScheme = "unix://"2526// server is the listening socket and the gRPC server registered on it.27type server struct {28	grpc     *grpc.Server29	listener net.Listener30	// The gauges and the listener that serves them. An empty --metrics31	// leaves the listener nil.32	readings *metrics33	metrics  net.Listener34	logger   *slog.Logger35}3637// newServer makes the store, takes the socket, and registers the38// Identity and Node services. It fails before it listens when the store39// cannot be made, because a driver with no store can hold no volume.40//41// newServer makes the store, takes the socket, and registers the42// Identity and Node services. ctx is the driver's run, and every fetch43// loop the Node service starts ends with it.44func newServer(ctx context.Context, cfg *config, logger *slog.Logger) (*server, error) {45	socket, found := strings.CutPrefix(cfg.endpoint, endpointScheme)46	if !found {47		return nil, fmt.Errorf("--endpoint %q does not begin with %s", cfg.endpoint, endpointScheme)48	}49	if err := os.MkdirAll(cfg.store, 0o755); err != nil {50		return nil, err51	}5253	// A pod that was killed leaves its socket file on the node, and the54	// next pod has to bind the same path. The file is removed, not55	// reported, because no other process ever owns it.56	if err := os.Remove(socket); err != nil && !errors.Is(err, fs.ErrNotExist) {57		return nil, err58	}59	listener, err := net.Listen("unix", socket)60	if err != nil {61		return nil, err62	}6364	readings := newMetrics()65	answering := newNode(ctx, cfg, newEvents(cfg.nodeID, logger), readings, logger)66	// The mounts outlive the driver, so a driver that starts takes back67	// the volumes its store still records.68	answering.resume(ctx)6970	registered := grpc.NewServer(grpc.UnaryInterceptor(logCalls(logger)))71	csi.RegisterIdentityServer(registered, &identity{store: cfg.store})72	csi.RegisterNodeServer(registered, answering)7374	metricsListener, err := readings.listen(cfg.metrics)75	if err != nil {76		_ = listener.Close()77		return nil, err78	}79	return &server{80		grpc:     registered,81		listener: listener,82		readings: readings,83		metrics:  metricsListener,84		logger:   logger,85	}, nil86}8788// serve blocks until the context ends, then stops the server and lets89// a call in flight finish.90func (s *server) serve(ctx context.Context) error {91	served := make(chan error, 1)92	go func() { served <- s.grpc.Serve(s.listener) }()93	if s.metrics != nil {94		go serveMetrics(ctx, s.metrics, s.readings, s.logger)95	}96	select {97	case err := <-served:98		return err99	case <-ctx.Done():100		s.grpc.GracefulStop()101		// A context that is over before Serve reaches the socket makes102		// Serve return ErrServerStopped. That is the stop this run asked103		// for, not a failure.104		if err := <-served; err != nil && !errors.Is(err, grpc.ErrServerStopped) {105			return err106		}107		return nil108	}109}110111// logCalls writes one line per RPC with its name and its status code.112// The kubelet's calls are the driver's whole input, and a person who113// reads the log has to see them.114func logCalls(logger *slog.Logger) grpc.UnaryServerInterceptor {115	return func(116		ctx context.Context,117		request any,118		call *grpc.UnaryServerInfo,119		handle grpc.UnaryHandler,120	) (any, error) {121		answer, err := handle(ctx, request)122		logger.InfoContext(ctx, "call",123			"rpc", call.FullMethod,124			"code", status.Code(err).String())125		return answer, err126	}127}

1package main23// stage.go holds the calls the kubelet makes for a writeable volume:4// the stage that brings the work tree to the ref, and the publish that5// binds it under the pod.67import (8	"context"9	"fmt"10	"os"11	"path/filepath"12	"strings"1314	"github.com/container-storage-interface/spec/lib/go/csi"15	"google.golang.org/grpc/codes"16	"google.golang.org/grpc/status"17)1819// NodeStageVolume fetches the ref, makes the work tree on the volume's20// first stage on this node, and starts the loop that reads the claim.21func (n *node) NodeStageVolume(22	ctx context.Context, request *csi.NodeStageVolumeRequest,23) (*csi.NodeStageVolumeResponse, error) {24	id := request.GetVolumeId()25	staging := request.GetStagingTargetPath()26	switch {27	case id == "":28		return nil, status.Error(codes.InvalidArgument, "volume_id: the call names no volume")29	case strings.ContainsRune(id, filepath.Separator):30		return nil, status.Error(codes.InvalidArgument, "volume_id: a volume id is one path element")31	case staging == "":32		return nil, status.Error(codes.InvalidArgument, "staging_target_path: the call names no path")33	}34	if err := checkAccessMode(request.GetVolumeCapability()); err != nil {35		return nil, err36	}37	parsed, err := parseStageAttributes(request.GetVolumeContext())38	if err != nil {39		return nil, err40	}41	holder, err := parseCredentials(request.GetSecrets())42	if err != nil {43		return nil, err44	}4546	n.mu.Lock()47	staged, found := n.staged[id]48	n.mu.Unlock()49	if found {50		if staged.staging != staging {51			return nil, status.Errorf(codes.FailedPrecondition,52				"volume_id: %s is staged at %s", id, staged.staging)53		}54		return &csi.NodeStageVolumeResponse{}, nil55	}5657	directory := n.store.volumeDir(id)58	if err := os.MkdirAll(directory, 0o700); err != nil {59		return nil, status.Error(codes.Internal, err.Error())60	}61	repo := n.store.repository(parsed.url)62	arriving := &volume{63		id:          id,64		attributes:  parsed,65		credentials: holder,66		directory:   directory,67		tree:        filepath.Join(directory, "tree"),68		work:        n.store.workTree(repo, id),69		staging:     staging,70		writeable:   true,71	}72	if err := n.stageTree(ctx, arriving, repo); err != nil {73		return nil, err74	}7576	n.mu.Lock()77	n.staged[id] = arriving78	n.arm(arriving)79	n.mu.Unlock()80	return &csi.NodeStageVolumeResponse{}, nil81}8283// checkAccessMode refuses every mode but the one ReadWriteOncePod asks84// for. The driver pushes what one writer wrote, and ReadWriteOnce85// allows two pods on one node.86func checkAccessMode(capability *csi.VolumeCapability) error {87	mode := capability.GetAccessMode().GetMode()88	if mode != csi.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER {89		return status.Errorf(codes.InvalidArgument,90			"access_mode: %s is not SINGLE_NODE_SINGLE_WRITER, which is ReadWriteOncePod", mode)91	}92	return nil93}9495// stageTree fetches the ref into the shared bare repository and brings96// the work tree to it. A tree that already exists is left as the last97// pod left it. A ref that moved under it is reported in the condition;98// plan 06 reconciles it.99func (n *node) stageTree(ctx context.Context, staging *volume, repo *repository) error {100	defer repo.lock()()101102	if !repo.exists() {103		if err := repo.create(ctx); err != nil {104			return status.Error(codes.Internal, err.Error())105		}106	}107	env, remove, err := staging.credentials.use(staging.directory)108	if err != nil {109		return status.Error(codes.Internal, err.Error())110	}111	fetchErr := repo.fetch(ctx, env, staging.attributes.ref, 0)112	remove()113	if fetchErr != nil {114		return status.Error(codes.Unavailable, fetchErr.Error())115	}116	commit, err := repo.resolve(ctx, staging.attributes.ref)117	if err != nil {118		return status.Error(codes.Internal, err.Error())119	}120121	if !staging.work.exists() {122		if err := staging.work.create(ctx, staging.attributes.ref, commit); err != nil {123			return status.Error(codes.Internal, err.Error())124		}125		staging.reportCommit(commit)126		return nil127	}128	head, err := staging.work.head(ctx)129	if err != nil {130		return status.Error(codes.Internal, err.Error())131	}132	staging.reportCommit(head)133	if head != commit {134		staging.reportTrouble(fmt.Sprintf("upstream moved: %s is at %s and the tree is at %s",135			staging.attributes.ref, short(commit), short(head)))136	}137	return nil138}139140// NodeUnstageVolume stops the loops and keeps the work tree, because141// the next stage on this node starts from what the pod wrote.142func (n *node) NodeUnstageVolume(143	ctx context.Context, request *csi.NodeUnstageVolumeRequest,144) (*csi.NodeUnstageVolumeResponse, error) {145	id := request.GetVolumeId()146	switch {147	case id == "":148		return nil, status.Error(codes.InvalidArgument, "volume_id: the call names no volume")149	case request.GetStagingTargetPath() == "":150		return nil, status.Error(codes.InvalidArgument, "staging_target_path: the call names no path")151	}152153	n.mu.Lock()154	staged, found := n.staged[id]155	if found {156		delete(n.staged, id)157		n.disarm(staged)158	}159	n.mu.Unlock()160	n.logger.InfoContext(ctx, "unstaged", "volume", id)161	return &csi.NodeUnstageVolumeResponse{}, nil162}163164// publishStaged binds the work tree read-write under the pod. A volume165// the kubelet never staged is refused, because the tree it would bind166// does not exist.167func (n *node) publishStaged(168	ctx context.Context,169	request *csi.NodePublishVolumeRequest,170	parsed *attributes,171	holder *credentials,172) error {173	id, target := request.GetVolumeId(), request.GetTargetPath()174	n.mu.Lock()175	staged, found := n.staged[id]176	published, standing := n.volumes[id]177	n.mu.Unlock()178	if !found {179		return status.Errorf(codes.FailedPrecondition,180			"volume_id: %s is not staged on this node", id)181	}182	if standing {183		if published.target != target {184			return status.Errorf(codes.FailedPrecondition,185				"volume_id: %s is published at %s", id, published.target)186		}187		return nil188	}189190	staged.setPod(parsed.pod)191	// A Secret named on the publish reaches the driver here and nowhere192	// else, so it replaces what the stage held.193	if holder != nil {194		staged.credentials = holder195	}196	if err := os.MkdirAll(target, 0o755); err != nil {197		return status.Error(codes.Internal, err.Error())198	}199	if err := unbind(n.mounts, target); err != nil {200		return status.Error(codes.Internal, err.Error())201	}202	if err := bindReadWrite(n.mounts, staged.tree, target); err != nil {203		return status.Error(codes.Internal, err.Error())204	}205	staged.target = target206	n.record(ctx, staged, request.GetVolumeContext())207208	n.mu.Lock()209	n.volumes[id] = staged210	n.watch(staged)211	n.mu.Unlock()212	return nil213}

1package main23// store.go holds the driver's directories on the node: one bare4// repository per URL under repos/, and one published tree per volume5// under volumes/.67import (8	"context"9	"crypto/sha256"10	"encoding/hex"11	"os"12	"path/filepath"13	"strconv"14	"sync"15)1617// refPrefix is where a store repository keeps the refs it follows. The18// driver's own namespace keeps them apart from anything the remote19// names, and lets one bare repository serve volumes on different refs.20const refPrefix = "refs/git-csi/"2122// store is the root directory and one lock per repository, made on23// first use.24type store struct {25	root string2627	mu    sync.Mutex28	locks map[string]*sync.Mutex29}3031func newStore(root string) *store {32	return &store{root: root, locks: map[string]*sync.Mutex{}}33}3435// repository is one bare repository, shared by every volume of the same36// URL on this node.37type repository struct {38	store *store39	url   string40	name  string41	dir   string42}4344// repository names the directory by the sha256 of the URL, because a URL45// is not a file name. create writes the URL beside it for a reader.46func (s *store) repository(url string) *repository {47	sum := sha256.Sum256([]byte(url))48	name := hex.EncodeToString(sum[:])49	return &repository{50		store: s,51		url:   url,52		name:  name,53		dir:   filepath.Join(s.root, "repos", name),54	}55}5657// volumeDir is where a volume keeps its published tree and, around each58// git invocation, its credential files.59func (s *store) volumeDir(id string) string {60	return filepath.Join(s.root, "volumes", id)61}6263// lock takes this repository's lock and returns the release. A fetch64// and a publish of the same URL never run at once.65func (r *repository) lock() func() {66	r.store.mu.Lock()67	held, found := r.store.locks[r.name]68	if !found {69		held = &sync.Mutex{}70		r.store.locks[r.name] = held71	}72	r.store.mu.Unlock()7374	held.Lock()75	return held.Unlock76}7778// exists reports whether the store already holds this repository.79func (r *repository) exists() bool {80	_, err := os.Stat(filepath.Join(r.dir, "HEAD"))81	return err == nil82}8384// create makes the bare repository and writes the URL beside it, so a85// person can read the store without hashing anything.86func (r *repository) create(ctx context.Context) error {87	if err := os.MkdirAll(r.dir, 0o755); err != nil {88		return err89	}90	if _, err := runGit(ctx, r.dir, nil, "init", "--quiet", "--bare"); err != nil {91		return err92	}93	return os.WriteFile(filepath.Join(r.dir, "url"), []byte(r.url+"\n"), 0o644)94}9596// fetch moves the driver's own ref to what the remote holds now. depth97// applies only when the caller asks, which is the first fetch of a new98// repository.99func (r *repository) fetch(ctx context.Context, env []string, ref string, depth int) error {100	args := []string{"fetch", "--quiet", "--no-tags"}101	if depth > 0 {102		args = append(args, "--depth="+strconv.Itoa(depth))103	}104	args = append(args, r.url, "+"+ref+":"+refPrefix+ref)105	_, err := runGit(ctx, r.dir, env, args...)106	return err107}108109// resolve is the commit the driver's own ref names, or an error when110// the store holds no copy of the ref.111func (r *repository) resolve(ctx context.Context, ref string) (string, error) {112	output, err := runGit(ctx, r.dir, nil, "rev-parse", "--verify", "--end-of-options",113		refPrefix+ref+"^{commit}")114	if err != nil {115		return "", err116	}117	return trimLine(output.stdout), nil118}119120// checkout writes the commit into dir. The index file lives outside the121// tree and is removed after, so the tree holds only what the commit122// holds and no pod ever sees a git file.123func (r *repository) checkout(ctx context.Context, commit, dir string) error {124	if err := os.MkdirAll(dir, 0o755); err != nil {125		return err126	}127	index := dir + ".index"128	defer os.Remove(index)129130	_, err := runGit(ctx, r.dir, []string{"GIT_INDEX_FILE=" + index},131		"--work-tree="+dir, "-c", "core.bare=false",132		"checkout", "--force", commit, "--", ".")133	return err134}135136// nextTree is the checkout a placement makes beside the published tree.137const nextTree = "next"138139// place puts the commit in the published tree. A tree that is not there140// yet arrives whole, with one rename. A tree a pod already reads is141// replaced entry by entry, because a bind mount follows the directory142// the driver bound and not its name: a renamed directory leaves every143// reader on the old tree. So the new tree arrives inside the directory144// the pod holds, and each file appears in one rename.145func (r *repository) place(ctx context.Context, commit, directory, tree string) error {146	fresh := filepath.Join(directory, nextTree)147	if err := os.RemoveAll(fresh); err != nil {148		return err149	}150	if err := r.checkout(ctx, commit, fresh); err != nil {151		return err152	}153	if _, err := os.Stat(tree); err != nil {154		return os.Rename(fresh, tree)155	}156	if err := replaceTree(fresh, tree); err != nil {157		return err158	}159	// The pod already reads the new tree, and the next placement removes160	// this checkout again, so a removal that fails is not a failure of161	// the placement.162	_ = os.RemoveAll(fresh)163	return nil164}165166// trimLine removes the newline from one line of git output.167func trimLine(out string) string {168	for len(out) > 0 && (out[len(out)-1] == '\n' || out[len(out)-1] == '\r') {169		out = out[:len(out)-1]170	}171	return out172}173174// replaceTree makes published hold exactly what fresh holds. Entries175// upstream removed go first. A directory present on both sides is176// recursed into, so its inode and the pod's view of it stay. Every other177// entry moves with one rename, so a reader sees the old file or the178// new one and never a partial write.179func replaceTree(fresh, published string) error {180	arriving, err := os.ReadDir(fresh)181	if err != nil {182		return err183	}184	standing, err := os.ReadDir(published)185	if err != nil {186		return err187	}188189	keep := make(map[string]bool, len(arriving))190	for _, entry := range arriving {191		keep[entry.Name()] = true192	}193	for _, entry := range standing {194		if !keep[entry.Name()] {195			if err := os.RemoveAll(filepath.Join(published, entry.Name())); err != nil {196				return err197			}198		}199	}200201	for _, entry := range arriving {202		from := filepath.Join(fresh, entry.Name())203		to := filepath.Join(published, entry.Name())204		there, err := os.Lstat(to)205		switch {206		case err == nil && entry.IsDir() && there.IsDir():207			if err := replaceTree(from, to); err != nil {208				return err209			}210			continue211		case err == nil && entry.IsDir() != there.IsDir():212			if err := os.RemoveAll(to); err != nil {213				return err214			}215		}216		if err := os.Rename(from, to); err != nil {217			return err218		}219	}220	return nil221}222223// treeSize is the bytes the tree's regular files hold, which224// NodeGetVolumeStats reports as used.225func treeSize(dir string) (int64, error) {226	var total int64227	err := filepath.Walk(dir, func(_ string, info os.FileInfo, err error) error {228		if err != nil {229			return err230		}231		if info.Mode().IsRegular() {232			total += info.Size()233		}234		return nil235	})236	return total, err237}

1package main23// volume.go holds one volume this node has and everything the driver4// reports about it.56import (7	"fmt"8	"sync"9)1011// volume is one volume this node holds: the commit its tree stands on,12// the trouble since the last good fetch, what the pod wrote and the13// driver has not committed, and the claim and class that say whether it14// may commit.15type volume struct {16	id          string17	attributes  *attributes18	credentials *credentials19	directory   string20	tree        string21	target      string22	// A writeable volume has a git directory and a staging path. A read-23	// only volume has neither.24	work      *workTree25	staging   string26	writeable bool2728	mu      sync.Mutex29	commit  string30	trouble string31	// The pod the kubelet named at publish. A stage call names no pod.32	pod podReference33	// What the pod wrote and the driver has not committed, and the claim34	// and class that say whether it may.35	pending []change36	claim   claimReference37	class   string38	armed   bool39}4041// setPod records the pod a publish named, so an Event from a loop that42// started at stage reaches it.43func (v *volume) setPod(pod podReference) {44	v.mu.Lock()45	defer v.mu.Unlock()46	v.pod = pod47}4849func (v *volume) podRef() podReference {50	v.mu.Lock()51	defer v.mu.Unlock()52	return v.pod53}5455// reportPending records what the last scan found and reports whether56// the set went from empty to not empty, which is the one moment an57// Event is worth posting.58func (v *volume) reportPending(found []change) bool {59	v.mu.Lock()60	defer v.mu.Unlock()61	first := len(v.pending) == 0 && len(found) > 062	v.pending = found63	return first64}6566// reportArmed records the claim and the class and reports whether the67// volume moved between armed and unarmed.68func (v *volume) reportArmed(claim claimReference, class string, armed bool) bool {69	v.mu.Lock()70	defer v.mu.Unlock()71	changed := armed != v.armed72	v.claim = claim73	v.class = class74	v.armed = armed75	return changed76}7778// reading is what the gauges carry: the claim that labels them, whether79// the volume is armed, and how many paths are pending.80func (v *volume) reading() (claimReference, bool, int) {81	v.mu.Lock()82	defer v.mu.Unlock()83	return v.claim, v.armed, len(v.pending)84}8586// report is the condition every NodeGetVolumeStats answer carries. A87// failure comes first, then an unarmed volume with work the driver may88// not commit, then the commit the tree stands on.89func (v *volume) report() (bool, string) {90	v.mu.Lock()91	defer v.mu.Unlock()92	switch {93	case v.trouble != "":94		return true, v.trouble95	case v.writeable && !v.armed && len(v.pending) > 0:96		return true, fmt.Sprintf("unarmed: %d paths pending, no class on claim %s/%s",97			len(v.pending), v.claim.namespace, v.claim.name)98	case len(v.pending) > 0:99		return false, fmt.Sprintf("%s at %s, %d paths pending",100			v.attributes.ref, short(v.commit), len(v.pending))101	}102	return false, fmt.Sprintf("%s at %s", v.attributes.ref, short(v.commit))103}104105// reportCommit records that the tree holds commit and nothing is wrong.106func (v *volume) reportCommit(commit string) {107	v.mu.Lock()108	defer v.mu.Unlock()109	v.commit = commit110	v.trouble = ""111}112113// reportTrouble records a failure and reports whether it is the first114// since the last success, which is when an Event is worth posting.115func (v *volume) reportTrouble(message string) bool {116	v.mu.Lock()117	defer v.mu.Unlock()118	first := v.trouble == ""119	v.trouble = message120	return first121}122123func (v *volume) condition() (string, string) {124	v.mu.Lock()125	defer v.mu.Unlock()126	return v.commit, v.trouble127}

1package main23// watch.go holds the inotify watch on a published work tree and the4// sweep that backs it up. Together they decide when the driver reads5// what the pod wrote.67import (8	"context"9	"encoding/binary"10	"fmt"11	"io/fs"12	"os"13	"path/filepath"14	"sync"15	"time"1617	"golang.org/x/sys/unix"18	corev1 "k8s.io/api/core/v1"19)2021// defaultQuiesce is how long a tree rests before the driver reads what22// is pending. defaultSweep is how often it reads that anyway. The class23// sets the quiesce in plan 05.24const (25	defaultQuiesce = 30 * time.Second26	defaultSweep   = time.Minute27)2829// watchMask is the events that mean the pod changed the tree: a write,30// a create, a delete, and both halves of a rename.31const watchMask = unix.IN_CREATE | unix.IN_DELETE | unix.IN_MODIFY |32	unix.IN_MOVED_FROM | unix.IN_MOVED_TO | unix.IN_CLOSE_WRITE3334// watcher is the inotify watch and the sweep of one published work35// tree.36type watcher struct {37	node    *node38	volume  *volume39	quiesce time.Duration40	sweep   time.Duration41	cancel  context.CancelFunc42	changes chan struct{}43	running sync.WaitGroup4445	// The raw descriptor adds watches, and the file reads events. Calling46	// Fd on the file would take it out of the runtime's poller and make47	// the read blocking, so both are kept.48	descriptor int49	inotify    *os.File50	watched    map[int32]string51}5253// watch starts the loops that read one published work tree. The caller54// holds the node's lock.55func (n *node) watch(published *volume) {56	if !published.writeable {57		return58	}59	ctx, cancel := context.WithCancel(n.base)60	seeing := &watcher{61		node:    n,62		volume:  published,63		quiesce: n.quiesce,64		sweep:   n.sweep,65		cancel:  cancel,66		changes: make(chan struct{}, 1),67		watched: map[int32]string{},68	}69	n.watchers[published.id] = seeing70	seeing.running.Add(2)71	seeing.open(ctx)72	go seeing.read(ctx)73	go seeing.run(ctx)74}7576// unwatch ends both loops and waits for them, so a volume the kubelet77// unpublished holds no file open. The caller holds the node's lock.78func (n *node) unwatch(published *volume) {79	seeing, found := n.watchers[published.id]80	if !found {81		return82	}83	delete(n.watchers, published.id)84	seeing.cancel()85	seeing.running.Wait()86}8788// open takes an inotify file and adds a watch for the tree and every89// directory under it. A watch the kernel refuses, for example at the90// inotify limit, leaves the sweep as the whole watch, which is why the91// sweep exists.92func (w *watcher) open(ctx context.Context) {93	fd, err := w.node.inotify(unix.IN_CLOEXEC | unix.IN_NONBLOCK)94	if err != nil {95		w.node.logger.WarnContext(ctx, "the watch did not start",96			"volume", w.volume.id, "error", err)97		return98	}99	w.descriptor = fd100	w.inotify = os.NewFile(uintptr(fd), "inotify")101	w.add(ctx, w.volume.tree)102}103104// add watches the directory and everything under it, because inotify105// watches one directory at a time.106func (w *watcher) add(ctx context.Context, dir string) {107	err := filepath.WalkDir(dir, func(path string, entry fs.DirEntry, err error) error {108		if err != nil || !entry.IsDir() {109			return nil110		}111		watched, err := unix.InotifyAddWatch(w.descriptor, path, watchMask)112		if err != nil {113			return err114		}115		w.watched[int32(watched)] = path116		return nil117	})118	if err != nil {119		w.node.logger.WarnContext(ctx, "the watch missed a directory",120			"volume", w.volume.id, "directory", dir, "error", err)121	}122}123124// read turns every batch of inotify events into one nudge and adds a125// watch for each directory the pod created. It ends when run closes the126// file.127func (w *watcher) read(ctx context.Context) {128	defer w.running.Done()129	if w.inotify == nil {130		return131	}132	buffer := make([]byte, 16*1024)133	for {134		count, err := w.inotify.Read(buffer)135		if err != nil {136			return137		}138		for _, event := range w.events(buffer[:count]) {139			if event.directory != "" {140				w.add(ctx, event.directory)141			}142		}143		w.nudge()144	}145}146147// arrival is one inotify event: the directory the pod created, or148// nothing.149type arrival struct {150	directory string151}152153// events reads the kernel's own record: a watch descriptor, a mask, a154// cookie, the name's length, and the name.155func (w *watcher) events(buffer []byte) []arrival {156	found := []arrival{}157	for offset := 0; offset+unix.SizeofInotifyEvent <= len(buffer); {158		descriptor := int32(binary.NativeEndian.Uint32(buffer[offset:]))159		mask := binary.NativeEndian.Uint32(buffer[offset+4:])160		length := int(binary.NativeEndian.Uint32(buffer[offset+12:]))161		name := ""162		if length > 0 {163			name = string(trimZeros(buffer[offset+unix.SizeofInotifyEvent : offset+unix.SizeofInotifyEvent+length]))164		}165		one := arrival{}166		if mask&unix.IN_CREATE != 0 && mask&unix.IN_ISDIR != 0 {167			one.directory = filepath.Join(w.watched[descriptor], name)168		}169		found = append(found, one)170		offset += unix.SizeofInotifyEvent + length171	}172	return found173}174175// trimZeros removes the zero bytes inotify pads a name with.176func trimZeros(name []byte) []byte {177	for len(name) > 0 && name[len(name)-1] == 0 {178		name = name[:len(name)-1]179	}180	return name181}182183// nudge restarts the quiesce timer. The channel has one slot and the184// send never blocks, so a burst of writes costs one nudge.185func (w *watcher) nudge() {186	select {187	case w.changes <- struct{}{}:188	default:189	}190}191192// run waits until the tree has been quiet for the quiesce, then reads193// what is pending. The sweep reads it anyway on a timer, because an194// inotify watch the kernel refused reports nothing.195func (w *watcher) run(ctx context.Context) {196	defer w.running.Done()197	quiesce := time.NewTimer(w.quiesce)198	quiesce.Stop()199	defer quiesce.Stop()200	sweep := time.NewTicker(w.sweep)201	defer sweep.Stop()202203	w.scan(ctx)204	for {205		select {206		case <-ctx.Done():207			w.close()208			return209		case <-w.changes:210			quiesce.Reset(w.quiesce)211		case <-quiesce.C:212			w.scan(ctx)213		case <-sweep.C:214			w.scan(ctx)215		}216	}217}218219// close ends the read loop, because a read of a closed file answers an220// error.221func (w *watcher) close() {222	if w.inotify != nil {223		_ = w.inotify.Close()224	}225}226227// scan records what git finds in the tree. An unarmed volume commits228// none of it. This is the report a person reads before a class arms the229// volume.230func (w *watcher) scan(ctx context.Context) {231	found, err := w.volume.work.pending(ctx)232	if err != nil {233		w.node.logger.WarnContext(ctx, "the tree was not read",234			"volume", w.volume.id, "error", err)235		return236	}237	if w.volume.reportPending(found) {238		claim, _, count := w.volume.reading()239		w.node.logger.InfoContext(ctx, "the tree holds work",240			"volume", w.volume.id, "paths", count)241		w.node.report(ctx, w.volume, claim, corev1.EventTypeNormal, reasonPending,242			pendingMessage(count))243	}244	w.node.readings.record(w.volume)245}246247// pendingMessage is what the Event says about a tree the driver has not248// committed.249func pendingMessage(count int) string {250	return fmt.Sprintf("%d paths pending", count)251}

1package main23// worktree.go holds the tree a writeable volume's pod writes, the git4// directory beside it, and the changes git finds in it.56import (7	"context"8	"os"9	"path/filepath"10	"strings"11	"sync"12)1314// alternatesFile names the bare repository whose objects this work tree15// reads, so history is stored once per URL.16const alternatesFile = "objects/info/alternates"1718// workTree is one volume's git directory and the checkout beside it.19// The checkout holds no .git of its own, so the pod cannot commit or20// push around the driver.21type workTree struct {22	repository *repository23	directory  string24	gitDir     string25	tree       string2627	mu sync.Mutex28}2930// workTree is the work tree of a volume, sharing the bare repository of31// its URL.32func (s *store) workTree(repo *repository, id string) *workTree {33	directory := s.volumeDir(id)34	return &workTree{35		repository: repo,36		directory:  directory,37		gitDir:     filepath.Join(directory, "git"),38		tree:       filepath.Join(directory, "tree"),39	}40}4142// exists reports whether create finished. HEAD is what create writes43// last.44func (w *workTree) exists() bool {45	_, err := os.Stat(filepath.Join(w.gitDir, "HEAD"))46	return err == nil47}4849// create makes the git directory beside the tree, shares the bare50// repository's objects through the alternates file, points HEAD at the51// ref, and resets the tree to the commit. reset sets HEAD, the index,52// and the tree in one call, so git status is meaningful from the first53// stage.54func (w *workTree) create(ctx context.Context, ref, commit string) error {55	if err := os.MkdirAll(w.tree, 0o755); err != nil {56		return err57	}58	if _, err := w.git(ctx, "init", "--quiet"); err != nil {59		return err60	}61	alternates := filepath.Join(w.gitDir, alternatesFile)62	if err := os.MkdirAll(filepath.Dir(alternates), 0o755); err != nil {63		return err64	}65	objects := filepath.Join(w.repository.dir, "objects")66	if err := os.WriteFile(alternates, []byte(objects+"\n"), 0o644); err != nil {67		return err68	}69	if _, err := w.git(ctx, "symbolic-ref", "HEAD", "refs/heads/"+ref); err != nil {70		return err71	}72	_, err := w.git(ctx, "reset", "--hard", "--quiet", commit)73	return err74}7576// head is the commit the tree stands on.77func (w *workTree) head(ctx context.Context) (string, error) {78	output, err := w.git(ctx, "rev-parse", "--verify", "--end-of-options", "HEAD")79	if err != nil {80		return "", err81	}82	return trimLine(output.stdout), nil83}8485// change is one path git reports and the bytes it holds now.86type change struct {87	path string88	size int6489}9091// pending is what the pod wrote and the driver has not committed. Every92// untracked file is named, so three files under a new directory count93// as three paths and not one.94func (w *workTree) pending(ctx context.Context) ([]change, error) {95	output, err := w.git(ctx, "status", "--porcelain", "-z", "--untracked-files=all")96	if err != nil {97		return nil, err98	}99	return w.changes(output.stdout), nil100}101102// changes reads git's -z report: two status letters, a space, the path,103// and a second path after a rename or a copy.104func (w *workTree) changes(report string) []change {105	entries := strings.Split(report, "\x00")106	found := []change{}107	for i := 0; i < len(entries); i++ {108		entry := entries[i]109		if len(entry) < 4 {110			continue111		}112		if entry[0] == 'R' || entry[0] == 'C' {113			i++114		}115		found = append(found, change{path: entry[3:], size: w.sizeOf(entry[3:])})116	}117	return found118}119120// sizeOf is the size of a path git named, and zero for a path the pod121// deleted.122func (w *workTree) sizeOf(path string) int64 {123	info, err := os.Lstat(filepath.Join(w.tree, filepath.FromSlash(path)))124	if err != nil || !info.Mode().IsRegular() {125		return 0126	}127	return info.Size()128}129130// git runs git against the work tree with the git directory beside it,131// so the pod never sees a .git. The lock keeps a stage and a status of132// the same tree apart.133func (w *workTree) git(ctx context.Context, args ...string) (gitOutput, error) {134	w.mu.Lock()135	defer w.mu.Unlock()136	return runGit(ctx, w.directory, nil,137		append([]string{"--git-dir=" + w.gitDir, "--work-tree=" + w.tree}, args...)...)138}