feat: add --job-context-dir to configure job coordination file locations

agent/agent_configuration.go

@@ -45,6 +45,7 @@ type AgentConfiguration struct {
 	RunInPty                        bool
 	KubernetesExec                  bool
 	KubernetesContainerStartTimeout time.Duration
+	JobContextDir                   string
 
 	SigningJWKSFile  string // Where to find the key to sign pipeline uploads with (passed through to jobs, they might be uploading pipelines)
 	SigningJWKSKeyID string // The key ID to sign pipeline uploads with

agent/agent_worker.go

@@ -553,6 +553,7 @@ func (a *AgentWorker) RunJob(ctx context.Context, acceptResponse *api.Job, ignor
 		AgentStdout:                     a.agentStdout,
 		KubernetesExec:                  a.agentConfiguration.KubernetesExec,
 		KubernetesContainerStartTimeout: a.agentConfiguration.KubernetesContainerStartTimeout,
+		JobContextDir:                   a.agentConfiguration.JobContextDir,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to initialize job: %w", err)

agent/job_runner.go

@@ -90,6 +90,14 @@ type JobRunnerConfig struct {
 	// It's useful to be configured in situations like huge container image cold download.
 	KubernetesContainerStartTimeout time.Duration
 
+	// JobContextDir is the directory for files the agent uses to coordinate
+	// with the processes running the job: the job env files, the job timeout
+	// marker file, and, in Kubernetes mode, the coordination socket. If
+	// empty, it defaults to kubernetes.DefaultContextDir in Kubernetes mode,
+	// or the system temporary directory otherwise. In Kubernetes mode it must
+	// be on a volume shared by all containers in the pod.
+	JobContextDir string
+
 	// Stdout of the parent agent process. Used for job log stdout writing arg, for simpler containerized log collection.
 	AgentStdout io.Writer
 }
@@ -212,7 +220,9 @@ func NewJobRunner(ctx context.Context, l logger.Logger, apiClient *api.Client, c
 		},
 	)
 
-	r.envShellFile, r.envJSONFile, err = createJobEnvFiles(r.agentLogger, r.conf.Job.ID, conf.KubernetesExec)
+	contextDir := jobContextDir(conf)
+
+	r.envShellFile, r.envJSONFile, err = createJobEnvFiles(r.agentLogger, r.conf.Job.ID, contextDir)
 	if err != nil {
 		return nil, err
 	}
@@ -221,7 +231,7 @@ func NewJobRunner(ctx context.Context, l logger.Logger, apiClient *api.Client, c
 	// disk if the job is cancelled because of a Buildkite job-level timeout
 	// (see Cancel). The bootstrap subprocess reads this path from the
 	// BUILDKITE_AGENT_JOB_TIMEOUT_FILE env var.
-	r.jobTimeoutFilePath = jobTimeoutFilePath(r.conf.Job.ID, conf.KubernetesExec)
+	r.jobTimeoutFilePath = jobTimeoutFilePath(r.conf.Job.ID, contextDir)
 
 	env, err := r.createEnvironment(ctx)
 	if err != nil {
@@ -338,6 +348,7 @@ func NewJobRunner(ctx context.Context, l logger.Logger, apiClient *api.Client, c
 			return nil, fmt.Errorf("failed to parse BUILDKITE_CONTAINER_COUNT: %w", err)
 		}
 		r.process = kubernetes.NewRunner(r.agentLogger, kubernetes.RunnerConfig{
+			SocketPath:         kubernetes.SocketPath(jobContextDir(conf)),
 			Stdout:             r.jobLogs,
 			Stderr:             r.jobLogs,
 			ClientCount:        containerCount,
@@ -934,13 +945,22 @@ func (r *JobRunner) onUploadHeaderTime(ctx context.Context, cursor, total int, t
 	}
 }
 
-func createJobEnvFiles(l logger.Logger, jobID string, kubernetesExec bool) (shellFile, jsonFile *os.File, err error) {
-	// Use /workspace in Kubernetes mode for shared volume access between containers
-	tempDir := os.TempDir()
-	if kubernetesExec {
-		tempDir = "/workspace"
+// jobContextDir returns the directory for files the agent creates to
+// coordinate with the processes running the job (the env files, the job
+// timeout marker file, and, in Kubernetes mode, the coordination socket). In
+// Kubernetes mode these files are read by other containers in the pod, so the
+// directory must be on a volume shared by all containers in the pod.
+func jobContextDir(conf JobRunnerConfig) string {
+	if conf.JobContextDir != "" {
+		return conf.JobContextDir
+	}
+	if conf.KubernetesExec {
+		return kubernetes.DefaultContextDir
 	}
+	return os.TempDir()
+}
 
+func createJobEnvFiles(l logger.Logger, jobID, tempDir string) (shellFile, jsonFile *os.File, err error) {
 	// tempDir is not guaranteed to exist
 	if _, err := os.Stat(tempDir); os.IsNotExist(err) {
 		// Actual file permissions will be reduced by umask, and won't be 0o777 unless the user has manually changed the umask to 000
@@ -971,10 +991,6 @@ func createJobEnvFiles(l logger.Logger, jobID string, kubernetesExec bool) (shel
 // because of a Buildkite job-level timeout. The file is not created until
 // the timeout actually fires; the bootstrap detects the timeout by checking
 // whether the file exists.
-func jobTimeoutFilePath(jobID string, kubernetesExec bool) string {
-	tempDir := os.TempDir()
-	if kubernetesExec {
-		tempDir = "/workspace"
-	}
+func jobTimeoutFilePath(jobID, tempDir string) string {
 	return filepath.Join(tempDir, fmt.Sprintf("job-timeout-%s", jobID))
 }

agent/job_runner_test.go

@@ -66,14 +66,55 @@ func TestValidateJobValue(t *testing.T) {
 func TestJobTimeoutFilePath(t *testing.T) {
 	t.Parallel()
 
-	got := jobTimeoutFilePath("abc123", false)
+	got := jobTimeoutFilePath("abc123", jobContextDir(JobRunnerConfig{}))
 	want := filepath.Join(os.TempDir(), "job-timeout-abc123")
 	if got != want {
-		t.Errorf("jobTimeoutFilePath(%q, false) = %q, want %q", "abc123", got, want)
+		t.Errorf("jobTimeoutFilePath(%q, jobContextDir({})) = %q, want %q", "abc123", got, want)
 	}
 
-	if got, want := jobTimeoutFilePath("abc123", true), filepath.Join("/workspace", "job-timeout-abc123"); got != want {
-		t.Errorf("jobTimeoutFilePath(%q, true) = %q, want %q", "abc123", got, want)
+	k8sDir := jobContextDir(JobRunnerConfig{KubernetesExec: true})
+	if got, want := jobTimeoutFilePath("abc123", k8sDir), filepath.Join("/workspace", "job-timeout-abc123"); got != want {
+		t.Errorf("jobTimeoutFilePath(%q, %q) = %q, want %q", "abc123", k8sDir, got, want)
+	}
+}
+
+func TestJobContextDir(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		conf JobRunnerConfig
+		want string
+	}{
+		{
+			name: "default",
+			conf: JobRunnerConfig{},
+			want: os.TempDir(),
+		},
+		{
+			name: "explicit_dir",
+			conf: JobRunnerConfig{JobContextDir: "/var/lib/buildkite/job"},
+			want: "/var/lib/buildkite/job",
+		},
+		{
+			name: "kubernetes_default",
+			conf: JobRunnerConfig{KubernetesExec: true},
+			want: "/workspace",
+		},
+		{
+			name: "kubernetes_explicit_dir",
+			conf: JobRunnerConfig{
+				KubernetesExec: true,
+				JobContextDir:  "/buildkite-shared",
+			},
+			want: "/buildkite-shared",
+		},
+	}
+
+	for _, tc := range tests {
+		if got := jobContextDir(tc.conf); got != tc.want {
+			t.Errorf("%s: jobContextDir(%+v) = %q, want %q", tc.name, tc.conf, got, tc.want)
+		}
 	}
 }
 

clicommand/agent_start.go

@@ -206,6 +206,7 @@ type AgentStartConfig struct {
 	StrictSingleHooks               bool          `cli:"strict-single-hooks"`
 	KubernetesExec                  bool          `cli:"kubernetes-exec"`
 	KubernetesContainerStartTimeout time.Duration `cli:"kubernetes-container-start-timeout"`
+	JobContextDir                   string        `cli:"job-context-dir" normalize:"filepath"`
 	TraceContextEncoding            string        `cli:"trace-context-encoding"`
 	NoMultipartArtifactUpload       bool          `cli:"no-multipart-artifact-upload"`
 	ArtifactUploadConcurrency       int           `cli:"artifact-upload-concurrency"`
@@ -778,6 +779,7 @@ var AgentStartCommand = cli.Command{
 			EnvVar: "BUILDKITE_KUBERNETES_CONTAINER_START_TIMEOUT",
 			Value:  5 * time.Minute,
 		},
+		JobContextDirFlag,
 
 		// Other shared flags
 		RedactedVars,
@@ -1141,6 +1143,7 @@ var AgentStartCommand = cli.Command{
 			ArtifactUploadConcurrency:       cfg.ArtifactUploadConcurrency,
 			KubernetesExec:                  cfg.KubernetesExec,
 			KubernetesContainerStartTimeout: cfg.KubernetesContainerStartTimeout,
+			JobContextDir:                   cfg.JobContextDir,
 			PingMode:                        cfg.PingMode,
 
 			SigningJWKSFile:  cfg.SigningJWKSFile,

clicommand/global.go

@@ -106,6 +106,17 @@ var (
 		EnvVar: "BUILDKITE_CONTAINER_ID",
 	}
 
+	JobContextDirFlag = cli.StringFlag{
+		Name: "job-context-dir",
+		Usage: "The directory for files the agent uses to coordinate with the " +
+			"processes running the job: the job env files, the job timeout marker " +
+			"file, and, in Kubernetes mode, the coordination socket " +
+			"(buildkite.sock). In Kubernetes mode this defaults to /workspace and " +
+			"must be on a volume mounted by all containers in the pod; otherwise " +
+			"it defaults to the system temporary directory",
+		EnvVar: "BUILDKITE_JOB_CONTEXT_DIR",
+	}
+
 	KubernetesLogCollectionGracePeriodFlag = cli.DurationFlag{
 		Name:   "kubernetes-log-collection-grace-period",
 		Usage:  "Deprecated, do not use",

clicommand/kubernetes_bootstrap.go

@@ -29,6 +29,7 @@ intended to be used directly.`
 type KubernetesBootstrapConfig struct {
 	KubernetesContainerID                int           `cli:"kubernetes-container-id"`
 	KubernetesBootstrapConnectionTimeout time.Duration `cli:"kubernetes-bootstrap-connection-timeout"`
+	JobContextDir                        string        `cli:"job-context-dir" normalize:"filepath"`
 
 	// Global flags for debugging, etc
 	LogLevel    string   `cli:"log-level"`
@@ -44,6 +45,7 @@ var KubernetesBootstrapCommand = cli.Command{
 	Description: kubernetesBootstrapHelpDescription,
 	Flags: []cli.Flag{
 		KubernetesContainerIDFlag,
+		JobContextDirFlag,
 		cli.DurationFlag{
 			Name: "kubernetes-bootstrap-connection-timeout",
 			Usage: "This is intended to be used only by the Buildkite k8s stack " +
@@ -70,7 +72,10 @@ var KubernetesBootstrapCommand = cli.Command{
 		defer cancel()
 
 		// Connect the socket.
-		socket := &kubernetes.Client{ID: cfg.KubernetesContainerID}
+		socket := &kubernetes.Client{
+			ID:         cfg.KubernetesContainerID,
+			SocketPath: kubernetes.SocketPath(cfg.JobContextDir),
+		}
 
 		// Registration passes down the env vars the agent normally sets on the
 		// subprocess, but in this case the bootstrap is in a separate
@@ -293,6 +298,11 @@ var existingEnvPriority = map[string]struct{}{
 	"BUILDKITE_COMMAND": {},
 	// BUILDKITE_CONTAINER_ID is preserved in case of Hyrum's Law.
 	"BUILDKITE_CONTAINER_ID": {},
+	// BUILDKITE_JOB_CONTEXT_DIR is set by the k8s stack on each container.
+	// The shared volume holding the socket and job context files may be
+	// mounted at a different path in each container, so the local value must
+	// win over the agent container's value.
+	"BUILDKITE_JOB_CONTEXT_DIR": {},
 	// BUILDKITE_SOCKETS_PATH is set by agent-stack-k8s and varies by container
 	// name.
 	"BUILDKITE_SOCKETS_PATH": {},

env/protected.go

@@ -74,6 +74,7 @@ var protectedEnv = map[string]protection{
 	"BUILDKITE_GIT_SUBMODULE_CLONE_CONFIG":      {mutableFromWithinJob: true},
 	"BUILDKITE_HOOKS_PATH":                      {},
 	"BUILDKITE_HOOKS_SHELL":                     {},
+	"BUILDKITE_JOB_CONTEXT_DIR":                 {},
 	"BUILDKITE_KUBERNETES_EXEC":                 {},
 	"BUILDKITE_LOCAL_HOOKS_ENABLED":             {},
 	"BUILDKITE_PLUGINS_ALWAYS_CLONE_FRESH":      {mutableFromWithinJob: true},

kubernetes/client.go

@@ -29,7 +29,7 @@ var errNotConnected = errors.New("client not connected")
 // Callers should use context.WithTimeout to control the connection timeout.
 func (c *Client) Connect(ctx context.Context) (*RegisterResponse, error) {
 	if c.SocketPath == "" {
-		c.SocketPath = defaultSocketPath
+		c.SocketPath = SocketPath("")
 	}
 
 	// Retry until the context is cancelled. The high maxAttempts is a safety net

kubernetes/runner.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"net/rpc"
 	"os"
+	"path/filepath"
 	"sync"
 	"syscall"
 	"time"
@@ -23,7 +24,25 @@ func init() {
 	gob.Register(new(syscall.WaitStatus))
 }
 
-const defaultSocketPath = "/workspace/buildkite.sock"
+const (
+	// socketName is the name of the coordination socket file within the job
+	// context directory.
+	socketName = "buildkite.sock"
+
+	// DefaultContextDir is the job context directory used when no other
+	// directory is configured. It must be on a volume shared by all
+	// containers in the pod.
+	DefaultContextDir = "/workspace"
+)
+
+// SocketPath returns the path of the coordination socket within the given job
+// context directory. An empty dir means DefaultContextDir.
+func SocketPath(dir string) string {
+	if dir == "" {
+		dir = DefaultContextDir
+	}
+	return filepath.Join(dir, socketName)
+}
 
 type RunnerConfig struct {
 	SocketPath         string
@@ -37,7 +56,7 @@ type RunnerConfig struct {
 // NewRunner returns a runner, implementing the agent's jobRunner interface.
 func NewRunner(l logger.Logger, c RunnerConfig) *Runner {
 	if c.SocketPath == "" {
-		c.SocketPath = defaultSocketPath
+		c.SocketPath = SocketPath("")
 	}
 	clients := make([]*clientResult, c.ClientCount)
 	for i := range c.ClientCount {