Jaypore CI

runner.go (9742B)

---

 package jci
 
 import (
 	"bytes"
 	"database/sql"
 	"encoding/json"
 	"fmt"
 	"io"
 	"log"
 	"math/rand"
 	"net/http"
 	"os"
 	"strings"
 	"time"
 
 	_ "modernc.org/sqlite"
 )
 
 // Runner polls the jci server for jobs and dispatches them into Docker containers.
 //
 // Configuration env vars:
 //
 //	JCI_SERVER         base URL or host:port of the jci server, e.g. https://jci.example.com or jci_server:8080
 //	                   If no scheme is present, http:// is prepended automatically.
 //	JCI_RUNNER_SECRET  shared secret matching RUNNER_SECRET on the server
 //	JCI_RUNNER_DB      path to runner's SQLite DB (default /var/lib/jci/runner.db)
 //	JCI_BINARY         path to the git-jci binary to inject into containers (default /usr/local/bin/git-jci)
 func Runner(_ []string) error {
 	cfg, err := loadRunnerConfig()
 	if err != nil {
 		return err
 	}
 
 	db, err := openRunnerDB(cfg.dbPath)
 	if err != nil {
 		return fmt.Errorf("open runner db: %w", err)
 	}
 	defer db.Close()
 
 	runnerID, err := getOrCreateRunnerID(db)
 	if err != nil {
 		return fmt.Errorf("runner identity: %w", err)
 	}
 	log.Printf("jci runner starting (id=%s)", runnerID)
 	log.Printf("connecting to server %s", cfg.serverURL)
 
 	r := &runner{cfg: cfg, db: db, runnerID: runnerID, rt: newDockerRuntime()}
 	r.pollLoop()
 	return nil // pollLoop runs forever; return only on fatal errors
 }
 
 // ---------------------------------------------------------------------------
 // Config
 // ---------------------------------------------------------------------------
 
 type runnerConfig struct {
 	serverURL    string
 	runnerSecret string
 	dbPath       string
 	binaryPath   string
 }
 
 func loadRunnerConfig() (runnerConfig, error) {
 	cfg := runnerConfig{
 		dbPath:     "/var/lib/jci/runner.db",
 		binaryPath: "/usr/local/bin/git-jci",
 	}
 	cfg.serverURL = normalizeServerURL(os.Getenv("JCI_SERVER"))
 	cfg.runnerSecret = os.Getenv("JCI_RUNNER_SECRET")
 
 	if cfg.serverURL == "" || cfg.runnerSecret == "" {
 		return cfg, fmt.Errorf("missing required env vars: JCI_SERVER, JCI_RUNNER_SECRET")
 	}
 	if v := os.Getenv("JCI_RUNNER_DB"); v != "" {
 		cfg.dbPath = v
 	}
 	if v := os.Getenv("JCI_BINARY"); v != "" {
 		cfg.binaryPath = v
 	}
 	return cfg, nil
 }
 
 // ---------------------------------------------------------------------------
 // Runner SQLite identity store
 // ---------------------------------------------------------------------------
 
 func openRunnerDB(path string) (*sql.DB, error) {
 	if err := os.MkdirAll(parentDir(path), 0755); err != nil {
 		return nil, err
 	}
 	db, err := sql.Open("sqlite", path)
 	if err != nil {
 		return nil, err
 	}
 	_, err = db.Exec(`CREATE TABLE IF NOT EXISTS identity (runner_id TEXT PRIMARY KEY)`)
 	return db, err
 }
 
 func getOrCreateRunnerID(db *sql.DB) (string, error) {
 	var id string
 	err := db.QueryRow(`SELECT runner_id FROM identity LIMIT 1`).Scan(&id)
 	if err == nil {
 		return id, nil
 	}
 	if err != sql.ErrNoRows {
 		return "", err
 	}
 	id = randomID()
 	_, err = db.Exec(`INSERT INTO identity (runner_id) VALUES (?)`, id)
 	return id, err
 }
 
 // ---------------------------------------------------------------------------
 // Runner
 // ---------------------------------------------------------------------------
 
 type runner struct {
 	cfg      runnerConfig
 	db       *sql.DB
 	runnerID string
 	rt       ContainerRuntime
 }
 
 // pollLoop runs forever: poll the server every 5s + 0–2s jitter.
 // Container reaping runs in a separate goroutine on a 60s interval.
 func (r *runner) pollLoop() {
 	const reapInterval = 60 * time.Second
 	go func() {
 		for {
 			time.Sleep(reapInterval)
 			r.reapStaleContainers()
 		}
 	}()
 
 	for {
 		jitter := time.Duration(rand.Intn(2000)) * time.Millisecond
 		time.Sleep(5*time.Second + jitter)
 
 		retryAfter, job, err := r.poll()
 		if err != nil {
 			log.Printf("poll error: %v", err)
 			continue
 		}
 		if retryAfter > 0 {
 			log.Printf("at capacity; server asked to back off %s", retryAfter)
 			time.Sleep(retryAfter)
 			continue
 		}
 		if job == nil {
 			log.Printf("poll: no jobs available")
 			continue
 		}
 		go r.dispatch(job) // non-blocking; runner returns to poll immediately
 	}
 }
 
 type serverJob struct {
 	JobID             string `json:"job_id"`
 	CloneURL          string `json:"clone_url"`
 	CommitSHA         string `json:"commit_sha"`
 	RepoOwner         string `json:"repo_owner"`
 	RepoName          string `json:"repo_name"`
 	TimeoutSeconds    int    `json:"timeout_seconds"`
 	Image             string `json:"image"`
 	Script            string `json:"script"`              // base64-encoded shell script
 	AllowDockerSocket bool   `json:"allow_docker_socket"` // mount /var/run/docker.sock into job container
 }
 
 // poll contacts the server. Returns (retryAfter, job, err).
 // retryAfter > 0 means 429 was received; job may be nil when there is no work.
 func (r *runner) poll() (time.Duration, *serverJob, error) {
 	log.Printf("polling %s/poll", r.cfg.serverURL)
 	body, _ := json.Marshal(map[string]string{
 		"runner_id": r.runnerID,
 		"secret":    r.cfg.runnerSecret,
 	})
 	resp, err := http.Post(r.cfg.serverURL+"/poll", "application/json", bytes.NewReader(body))
 	if err != nil {
 		return 0, nil, err
 	}
 	defer resp.Body.Close()
 
 	log.Printf("poll response: %d", resp.StatusCode)
 	if resp.StatusCode == http.StatusTooManyRequests {
 		secs := 5
 		if v := resp.Header.Get("Retry-After"); v != "" {
 			fmt.Sscanf(v, "%d", &secs)
 		}
 		return time.Duration(secs) * time.Second, nil, nil
 	}
 	if resp.StatusCode != http.StatusOK {
 		data, _ := io.ReadAll(resp.Body)
 		log.Printf("poll: unexpected status %d: %s", resp.StatusCode, data)
 		return 0, nil, fmt.Errorf("server returned %d: %s", resp.StatusCode, data)
 	}
 
 	var result struct {
 		Job *serverJob `json:"job"`
 	}
 	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
 		return 0, nil, fmt.Errorf("decode poll response: %w", err)
 	}
 	if result.Job != nil {
 		log.Printf("poll: received job %s (%s/%s@%.12s)", result.Job.JobID, result.Job.RepoOwner, result.Job.RepoName, result.Job.CommitSHA)
 	}
 	return 0, result.Job, nil
 }
 
 // dispatch starts a detached Docker container to run the job.
 func (r *runner) dispatch(job *serverJob) {
 	log.Printf("received job %s: %s/%s@%.12s", job.JobID, job.RepoOwner, job.RepoName, job.CommitSHA)
 	log.Printf("job %s: launching container", job.JobID)
 
 	containerID, err := r.startJobContainer(job)
 	if err != nil {
 		log.Printf("job %s: failed to start container: %v", job.JobID, err)
 		return
 	}
 	log.Printf("job %s: container %s started — waiting for completion (will be reaped after timeout)", job.JobID, containerID[:12])
 
 	// Container reaping is handled by the periodic reaper; nothing more to do here.
 }
 
 // startJobContainer launches a detached container and returns its ID.
 func (r *runner) startJobContainer(job *serverJob) (string, error) {
 	// The script is delivered as a base64-encoded env var (JCI_SCRIPT).
 	// The container decodes and pipes it into sh — no script content ever
 	// touches a shell on the runner host.
 	timeoutLabel := fmt.Sprintf("%dm", job.TimeoutSeconds/60)
 
 	binds := []string{r.cfg.binaryPath + ":/usr/local/bin/git-jci:ro"}
 	if job.AllowDockerSocket {
 		binds = append(binds, "/var/run/docker.sock:/var/run/docker.sock")
 	}
 
 	spec := ContainerSpec{
 		Image:   job.Image,
 		Command: []string{"/bin/sh", "-c", `echo "$JCI_SCRIPT" | base64 -d | /bin/sh`},
 		Env: []string{
 			"JCI_CLONE_URL=" + job.CloneURL,
 			"JCI_COMMIT=" + job.CommitSHA,
 			"JCI_SCRIPT=" + job.Script,
 		},
 		Binds: binds,
 		Labels: map[string]string{
 			"jci-job":     "y",
 			"jci-job-id":  job.JobID,
 			"jci-timeout": timeoutLabel,
 		},
 	}
 
 	// Log without credentials (clone URL contains token).
 	logSpec := spec
 	logSpec.Env = []string{
 		"JCI_CLONE_URL=" + maskURL(job.CloneURL),
 		"JCI_COMMIT=" + job.CommitSHA,
 		"JCI_SCRIPT=(base64, " + fmt.Sprintf("%d bytes", len(job.Script)) + ")",
 	}
 	log.Printf("job %s: %s", job.JobID, logContainerCmd(logSpec))
 
 	return r.rt.StartContainer(spec)
 }
 
 // reapStaleContainers kills containers labelled jci-job=y whose running time
 // exceeds the duration stored in their jci-timeout label.
 func (r *runner) reapStaleContainers() {
 	containers, err := r.rt.ListContainers("jci-job=y")
 	if err != nil {
 		log.Printf("reaper: list containers: %v", err)
 		return
 	}
 	log.Printf("reaper: found %d jci container(s)", len(containers))
 	for _, c := range containers {
 		label := c.Labels["jci-timeout"]
 		if label == "" {
 			continue
 		}
 		timeout, err := time.ParseDuration(label)
 		if err != nil {
 			log.Printf("reaper: container %s has invalid jci-timeout label %q: %v", c.ID[:12], label, err)
 			continue
 		}
 		startedAt, err := r.rt.InspectStartedAt(c.ID)
 		if err != nil {
 			log.Printf("reaper: inspect %s: %v", c.ID[:12], err)
 			continue
 		}
 		if time.Since(startedAt) > timeout {
 			log.Printf("reaper: killing stale container %s (timeout=%s, running since %s)", c.ID[:12], timeout, startedAt)
 			if err := r.rt.RemoveContainer(c.ID); err != nil {
 				log.Printf("reaper: remove %s: %v", c.ID[:12], err)
 			}
 		}
 	}
 }
 
 // normalizeServerURL ensures the URL has an http:// or https:// scheme.
 // This allows JCI_SERVER to be set as just a host:port (e.g. "jci_server:8080")
 // without triggering the "first path segment cannot contain colon" parse error.
 func normalizeServerURL(s string) string {
 	if s == "" {
 		return s
 	}
 	if !strings.HasPrefix(s, "http://") && !strings.HasPrefix(s, "https://") {
 		return "http://" + s
 	}
 	return s
 }
 
 // maskURL replaces the userinfo (credentials) in a URL with ***.
 // e.g. https://token:x@github.com/org/repo → https://***@github.com/org/repo
 func maskURL(raw string) string {
 	if i := strings.Index(raw, "@"); i != -1 {
 		if j := strings.Index(raw, "://"); j != -1 && j < i {
 			return raw[:j+3] + "***" + raw[i:]
 		}
 	}
 	return raw
 }