package builder import ( "bytes" "context" "encoding/json" "errors" "fmt" "net" "os" "os/exec" "path/filepath" "regexp" "strings" "time" "github.com/novox/mesh-controller/internal/artifacts" "github.com/novox/mesh-controller/internal/facts" ) // A pull request's merge check, run on the build seat (novox/hq to-be 45 §9). // // **The build machine already has what a check needs**: the repositories, a container runtime, the // artifact store where the controller keeps the facts snapshot, and a Go toolchain. So a check is one // more kind of work on the build seat's queue rather than a CI the mesh would have to run beside itself. // // One check: // // 1. clones the repository at the pull request's head, and beside it the repositories its check // reads — the controller the mesh runs, the catalogue, the host — each at the ref asked; // 2. reads the facts snapshot the controller keeps, and with it **the versions the mesh runs**: the // store a check's tests stand on is the store's own release, and the bus the bus's; // 3. raises a throwaway PostgreSQL and a throwaway bus of those versions, labelled with the ask so a // kill or a crash leaves nothing behind; // 4. builds the judge — the controller the mesh runs, or its main while the running one predates the // merge gate — and runs the repository's own merge-check.sh, or the merge gate alone for a // repository that declares none. **In the mesh's Go toolchain, in a container of its own**, never in // the build machine's: a pull request is code nobody has approved yet, and the build machine holds // the container runtime's socket; the check's container holds none, and reaches only the // throwaway store and bus on loopback; // 5. answers pass, warning or fail from what ran, and error — never pass — when it could not run. // CheckSpec is one check, as the controller asks it. type CheckSpec struct { ID string Repository string Ref string Owner string Repo string Number int Paths []string // Beside are the repositories cloned next to it, by the directory they are found under. Beside map[string]Beside // Toolchain is the image a check's Go runs in: the mesh's own Go toolchain, as it holds it. Toolchain string } // ToolchainOf is the Go toolchain image among what the mesh holds, empty when it holds none. func ToolchainOf(held map[string]string) string { for _, chain := range toolchains { if chain.Language == "go" { return held[chain.Base+"/"+chain.Artifact] } } return "" } // Beside is one repository cloned next to the one checked. type Beside struct { Repository string Ref string } // CheckVerdict is what came of one check. type CheckVerdict struct { Verdict string Summary string Report string Took time.Duration } // CheckScript is what a repository declares its merge check as: run from its root, with the // environment below. const CheckScript = "merge-check.sh" // Where a check finds what the builder raised and read for it. const ( EnvFacts = "MESH_FACTS" EnvGate = "MESH_GATE" EnvGateStore = "MESH_GATE_POSTGRES" EnvTestStore = "MESH_TEST_POSTGRES" EnvTestBus = "MESH_TEST_NATS" EnvRepository = "MESH_CHECK_REPOSITORY" EnvChanged = "MESH_CHECK_CHANGED" EnvVerdict = "MESH_CHECK_VERDICT" EnvBeside = "MESH_CHECK_BESIDE" ) // CheckTimeout bounds one check; a check that runs past it is an error, not a pass. var CheckTimeout = 45 * time.Minute // reportLines is how much of what a check printed travels in its verdict. const reportLines = 200 // Check runs one merge check. An error is that it could not run; the verdict is then "error". func Check(ctx context.Context, run Runner, spec CheckSpec, workspace, registry string, forge GitCredential, log Log) (CheckVerdict, error) { say := logging(log) began := time.Now() ctx, stop := context.WithTimeout(ctx, CheckTimeout) defer stop() root := filepath.Join(workspace, "check") if err := os.RemoveAll(root); err != nil { return CheckVerdict{}, err } if err := os.MkdirAll(root, 0o755); err != nil { return CheckVerdict{}, err } defer os.RemoveAll(root) credentials := "" if forge.URL != "" { credentials = filepath.Join(workspace, "git-credentials") if err := os.WriteFile(credentials, []byte(forge.URL+"\n"), 0o600); err != nil { return CheckVerdict{}, err } } clone := func(repository, ref, dir string) error { if _, err := run(ctx, root, "git", cloneWith(credentials, "clone", "--quiet", repository, dir)...); err != nil { return fmt.Errorf("cannot clone %s: %w", repository, err) } if ref != "" { if _, err := run(ctx, filepath.Join(root, dir), "git", "checkout", "--quiet", ref); err != nil { return fmt.Errorf("%s has no %s: %w", repository, ref, err) } } return nil } name := spec.Repo if name == "" { name = "checked" } say("check", "%s/%s#%d at %s", spec.Owner, spec.Repo, spec.Number, short(spec.Ref)) if err := clone(spec.Repository, spec.Ref, name); err != nil { return CheckVerdict{}, err } for dir, b := range spec.Beside { if dir == name || !safeName.MatchString(dir) { continue } if err := clone(b.Repository, b.Ref, dir); err != nil { return CheckVerdict{}, fmt.Errorf("beside it, %w", err) } say("check", "beside it %s at %s", dir, short(b.Ref)) } // The facts, and the versions they say the mesh runs. if registry == "" { return CheckVerdict{}, errors.New("no artifact store to read the facts snapshot from") } body, digest, err := (artifacts.Store{Address: registry}).GetTagged(ctx, facts.Repository, facts.Tag) if err != nil { return CheckVerdict{}, fmt.Errorf("the facts snapshot cannot be read, and a check without it judges nothing: %w", err) } f, err := facts.Decode(body) if err != nil { return CheckVerdict{}, err } factsFile := filepath.Join(root, "facts.json") if err := os.WriteFile(factsFile, body, 0o644); err != nil { return CheckVerdict{}, err } say("check", "the facts of %s (%s): %d machine(s), the bus at %s, the store at %s", f.Taken.Format(time.RFC3339), short(strings.TrimPrefix(digest, "sha256:")), len(f.Machines), f.Versions.Bus, f.Versions.Store) // The throwaway store and bus, of the versions the mesh runs, removed whatever happens. defer func() { removing, done := context.WithTimeout(context.Background(), time.Minute) defer done() if n, err := RemoveContainersOf(removing, run, spec.ID); err == nil && n > 0 { say("check", "removed %d throwaway container(s)", n) } }() labelled := Labelled(run, spec.ID) store, err := throwaway(ctx, labelled, run, spec.ID+"-store", StoreImage(f.Versions.Store), 5432, []string{"-e", "POSTGRES_PASSWORD=check"}, nil) if err != nil { return CheckVerdict{}, err } storeURL := "postgres://postgres:check@" + store + "/postgres?sslmode=disable" if err := waitFor(ctx, run, spec.ID+"-store", []string{"pg_isready", "-U", "postgres"}); err != nil { return CheckVerdict{}, err } bus, err := throwaway(ctx, labelled, run, spec.ID+"-bus", BusImage(f.Versions.Bus), 4222, nil, []string{"-js"}) if err != nil { return CheckVerdict{}, err } if err := dialable(ctx, bus); err != nil { return CheckVerdict{}, err } say("check", "a throwaway store (%s) and bus (%s) of the versions the mesh runs", StoreImage(f.Versions.Store), BusImage(f.Versions.Bus)) if spec.Toolchain == "" { return CheckVerdict{}, errors.New("the mesh holds no Go toolchain to run a check in") } inToolchain := func(dir string, env []string, command ...string) []string { // As the builder itself: what a check writes into the workspace is the builder's to remove. args := []string{"run", "--rm", "--network", "host", "--volume", workspace + ":" + workspace, "--workdir", dir, "--user", fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid()), "--env", "HOME=" + workspace} for _, e := range env { args = append(args, "--env", e) } return append(append(args, spec.Toolchain), command...) } // The judge: the controller the mesh runs, or its main while the running one has no merge gate. gate := "" if name != "mesh-controller" { gate, err = judge(ctx, labelled, run, root, inToolchain, say) if err != nil { return CheckVerdict{}, err } } verdictFile := filepath.Join(root, "verdict.json") env := append([]string{}, EnvFacts+"="+factsFile, EnvGate+"="+gate, EnvGateStore+"="+storeURL, EnvTestStore+"="+storeURL, EnvTestBus+"=nats://"+bus, EnvRepository+"="+spec.Owner+"/"+spec.Repo, EnvChanged+"="+strings.Join(spec.Paths, ","), EnvVerdict+"="+verdictFile, EnvBeside+"="+root, "GOCACHE="+filepath.Join(workspace, "go-cache"), "GOMODCACHE="+filepath.Join(workspace, "go-modules")) tree := filepath.Join(root, name) var command []string if _, err := os.Stat(filepath.Join(tree, CheckScript)); err == nil { say("check", "running its %s in the mesh's Go toolchain", CheckScript) command = []string{"sh", CheckScript} } else { if gate == "" { return CheckVerdict{}, fmt.Errorf("%s declares no %s and there is no judge to run", name, CheckScript) } say("check", "it declares no %s: the merge gate alone", CheckScript) command = []string{"sh", "-c", `"$MESH_GATE" merge-gate --facts "$MESH_FACTS" --store "$MESH_GATE_POSTGRES" ` + `--repository "$MESH_CHECK_REPOSITORY" --tree . --changed "$MESH_CHECK_CHANGED" --json > "$MESH_CHECK_VERDICT"`} } cmd := exec.CommandContext(ctx, "docker", LabelledArgs("docker", inToolchain(tree, env, command...), spec.ID)...) inItsOwnGroup(cmd) var out tail cmd.Stdout, cmd.Stderr = &out, &out runErr := cmd.Run() // **And the replays of what the mesh runs** (to-be 45 §9, M9): mesh-lab's, from its main — reviewed // code, so given the container runtime the resolver replay raises containers with — against the bus // of the release the mesh runs and the change's own catalogue when the change is to the catalogue. var replayErr error if lab := filepath.Join(root, "mesh-lab", "replays"); runErr == nil && ctx.Err() == nil { if _, err := os.Stat(lab); err == nil { catalogue := filepath.Join(root, "mesh-catalog") if name == "mesh-catalog" { catalogue = tree } say("check", "the replays of what the mesh runs, from mesh-lab") fmt.Fprintln(&out, "--- the replays (mesh-lab replays/)") args := inToolchain(lab, []string{EnvTestBus + "=nats://" + bus, "MESH_REPLAY_CATALOGUE=" + catalogue, "GOCACHE=" + filepath.Join(workspace, "go-cache"), "GOMODCACHE=" + filepath.Join(workspace, "go-modules")}, "go", "test", "-count=1", "./...") // The socket goes to the replays alone, never to the change's own script above. args = append([]string{args[0], "--volume", "/var/run/docker.sock:/var/run/docker.sock"}, args[1:]...) replays := exec.CommandContext(ctx, "docker", LabelledArgs("docker", args, spec.ID)...) inItsOwnGroup(replays) replays.Stdout, replays.Stderr = &out, &out replayErr = replays.Run() } } v := CheckVerdict{Report: out.String(), Took: time.Since(began)} var gateSaid struct { Verdict string `json:"verdict"` Summary string `json:"summary"` } if raw, err := os.ReadFile(verdictFile); err == nil { _ = json.Unmarshal(raw, &gateSaid) } switch { case errors.Is(ctx.Err(), context.DeadlineExceeded): v.Verdict, v.Summary = "error", fmt.Sprintf("the check ran past %s and was ended", CheckTimeout) case ctx.Err() != nil: return v, ctx.Err() case runErr != nil: v.Verdict = "fail" v.Summary = gateSaid.Summary if v.Summary == "" || gateSaid.Verdict != "fail" { v.Summary = "the merge check failed: " + lastLine(out.String()) } case replayErr != nil: v.Verdict, v.Summary = "fail", "a replay of a core incident fails with this change: "+lastLine(out.String()) default: v.Verdict, v.Summary = "pass", "the merge check passed" if gateSaid.Verdict == "warning" || gateSaid.Verdict == "pass" { v.Verdict, v.Summary = gateSaid.Verdict, gateSaid.Summary } } say("check", "%s — %s (%s)", strings.ToUpper(v.Verdict), v.Summary, v.Took.Round(time.Second)) return v, nil } // safeName is a directory a repository beside a check may be cloned under. var safeName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`) // StoreImage is the store a check stands on: the major release the mesh's store runs (`17.11` → 17), // on Alpine as the store module runs it. func StoreImage(version string) string { major, _, _ := strings.Cut(strings.TrimSpace(version), ".") if major == "" || strings.ContainsAny(major, " (") { major = "17" } return "postgres:" + major + "-alpine" } // BusImage is the bus a check stands on: the release the mesh's bus server runs. func BusImage(version string) string { v := strings.TrimPrefix(strings.TrimSpace(version), "v") if v == "" { v = "2.11" } return "nats:" + v + "-alpine" } // throwaway starts a container publishing one port on loopback, and answers where it is reached. func throwaway(ctx context.Context, run, plain Runner, name, image string, port int, opts, args []string) (string, error) { invocation := []string{"run", "-d", "--rm", "--name", name, "-p", fmt.Sprintf("127.0.0.1::%d", port)} invocation = append(invocation, opts...) invocation = append(invocation, image) invocation = append(invocation, args...) if _, err := run(ctx, "", "docker", invocation...); err != nil { return "", fmt.Errorf("a throwaway %s could not be raised: %w", image, err) } out, err := plain(ctx, "", "docker", "port", name, fmt.Sprintf("%d/tcp", port)) if err != nil { return "", err } for _, line := range strings.Split(strings.TrimSpace(out), "\n") { if strings.HasPrefix(line, "127.0.0.1:") { return strings.TrimSpace(line), nil } } return "", fmt.Errorf("%s published %d nowhere on loopback: %q", name, port, out) } // waitFor runs a readiness command in a container until it answers, for a minute. func waitFor(ctx context.Context, run Runner, name string, ready []string) error { for i := 0; i < 60; i++ { if _, err := run(ctx, "", "docker", append([]string{"exec", name}, ready...)...); err == nil { return nil } select { case <-ctx.Done(): return ctx.Err() case <-time.After(time.Second): } } return fmt.Errorf("%s never became ready", name) } // dialable waits for an address to take a connection, for a minute. func dialable(ctx context.Context, address string) error { for i := 0; i < 60; i++ { if c, err := net.DialTimeout("tcp", address, time.Second); err == nil { c.Close() return nil } select { case <-ctx.Done(): return ctx.Err() case <-time.After(time.Second): } } return fmt.Errorf("nothing took a connection at %s", address) } // judge builds the controller that judges a change: the one the mesh runs, beside the check as // mesh-controller — or, when that one predates the merge gate, the controller's main, said. func judge(ctx context.Context, run, plain Runner, root string, inToolchain func(string, []string, ...string) []string, say func(step, format string, args ...any)) (string, error) { bin := filepath.Join(root, "bin", "mesh-controller") build := func(dir string) error { _, err := run(ctx, root, "docker", inToolchain(filepath.Join(root, dir), []string{"CGO_ENABLED=0", "GOFLAGS=-mod=vendor", "GOPROXY=off", "GOCACHE=" + filepath.Join(filepath.Dir(root), "go-cache")}, "go", "build", "-o", bin, "./cmd/mesh-controller")...) return err } // Static, so it runs where the builder does. hasGate := func() bool { out, _ := plain(ctx, root, bin, "merge-gate") return strings.Contains(out, "merge-gate --facts") } if _, err := os.Stat(filepath.Join(root, "mesh-controller")); err == nil { if err := build("mesh-controller"); err != nil { return "", fmt.Errorf("the controller the mesh runs does not build: %w", err) } if hasGate() { say("check", "judged by the controller the mesh runs") return bin, nil } } if _, err := os.Stat(filepath.Join(root, "mesh-controller-main")); err != nil { return "", errors.New("no controller beside the check to judge it with") } if err := build("mesh-controller-main"); err != nil { return "", fmt.Errorf("the controller's main does not build: %w", err) } say("check", "judged by the controller's main: the one the mesh runs predates the merge gate") return bin, nil } // tail keeps the last lines written to it. type tail struct{ buf bytes.Buffer } func (t *tail) Write(p []byte) (int, error) { t.buf.Write(p) if t.buf.Len() > 1<<20 { keep := t.buf.Bytes()[t.buf.Len()-(512<<10):] t.buf = *bytes.NewBuffer(append([]byte(nil), keep...)) } return len(p), nil } func (t *tail) String() string { lines := strings.Split(strings.TrimRight(t.buf.String(), "\n"), "\n") if len(lines) > reportLines { lines = lines[len(lines)-reportLines:] } return strings.Join(lines, "\n") } func lastLine(s string) string { lines := strings.Split(strings.TrimSpace(s), "\n") return strings.TrimSpace(lines[len(lines)-1]) }