Designing a Docker Image Build API: Job Model and Streamed Logs

// Broken. Do not copy this.
func handleBuild(w http.ResponseWriter, r *http.Request) {
	resp, err := dockerClient.ImageBuild(r.Context(), r.Body, types.ImageBuildOptions{
		Tags: []string{"demo:latest"},
	})
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	io.Copy(w, resp.Body)
}

// main.go
package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"sync"

	"github.com/docker/docker/api/types"
	"github.com/docker/docker/client"
)

// --- Job model -------------------------------------------------------------

type Job struct {
	ID string

	mu     sync.Mutex
	log    []byte
	status string // "running", "success", "failed"
	notify chan struct{}
}

func newJob(id string) *Job {
	return &Job{ID: id, status: "running", notify: make(chan struct{})}
}

func (j *Job) append(p []byte) {
	j.mu.Lock()
	j.log = append(j.log, p...)
	old := j.notify
	j.notify = make(chan struct{})
	j.mu.Unlock()
	close(old)
}

func (j *Job) finish(status string) {
	j.mu.Lock()
	j.status = status
	old := j.notify
	j.notify = make(chan struct{})
	j.mu.Unlock()
	close(old)
}

// snapshot returns any log bytes after `from`, whether the job is finished, and a
// channel that closes the next time either changes — so a caller with nothing new to
// read can wait on it instead of polling.
func (j *Job) snapshot(from int) (chunk []byte, status string, notify chan struct{}) {
	j.mu.Lock()
	defer j.mu.Unlock()
	if from < len(j.log) {
		chunk = j.log[from:]
	}
	return chunk, j.status, j.notify
}

// --- Job store ---------------------------------------------------------------

type Store struct {
	mu   sync.Mutex
	jobs map[string]*Job
}

func newStore() *Store { return &Store{jobs: map[string]*Job{}} }

func (s *Store) create() *Job {
	id := randomID()
	job := newJob(id)
	s.mu.Lock()
	s.jobs[id] = job
	s.mu.Unlock()
	return job
}

func (s *Store) get(id string) (*Job, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	job, ok := s.jobs[id]
	return job, ok
}

func randomID() string {
	b := make([]byte, 8)
	_, _ = rand.Read(b)
	return hex.EncodeToString(b)
}

// --- Build execution ---------------------------------------------------------

type buildMessage struct {
	Stream string `json:"stream"`
	Error  string `json:"error"`
}

func runBuild(cli *client.Client, job *Job, buildContext io.Reader, tag string) {
	resp, err := cli.ImageBuild(context.Background(), buildContext, types.ImageBuildOptions{
		Tags:   []string{tag},
		Remove: true,
	})
	if err != nil {
		job.append([]byte(fmt.Sprintf("could not start build: %v\n", err)))
		job.finish("failed")
		return
	}
	defer resp.Body.Close()

	failed := false
	dec := json.NewDecoder(resp.Body)
	for {
		var msg buildMessage
		if err := dec.Decode(&msg); err != nil {
			if err != io.EOF {
				job.append([]byte(fmt.Sprintf("stream error: %v\n", err)))
				failed = true
			}
			break
		}
		if msg.Stream != "" {
			job.append([]byte(msg.Stream))
		}
		if msg.Error != "" {
			job.append([]byte("ERROR: " + msg.Error + "\n"))
			failed = true
		}
	}

	if failed {
		job.finish("failed")
	} else {
		job.finish("success")
	}
}

// --- HTTP handlers -------------------------------------------------------------

type server struct {
	store  *Store
	docker *client.Client
}

func (s *server) createBuild(w http.ResponseWriter, r *http.Request) {
	tmp, err := os.CreateTemp("", "build-context-*.tar")
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	if _, err := io.Copy(tmp, r.Body); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	_ = tmp.Close()

	job := s.store.create()

	go func() {
		defer os.Remove(tmp.Name())
		f, err := os.Open(tmp.Name())
		if err != nil {
			job.append([]byte(fmt.Sprintf("could not reopen build context: %v\n", err)))
			job.finish("failed")
			return
		}
		defer f.Close()
		runBuild(s.docker, job, f, "build-api-demo:"+job.ID)
	}()

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	json.NewEncoder(w).Encode(map[string]string{"id": job.ID})
}

func (s *server) getBuild(w http.ResponseWriter, r *http.Request) {
	job, ok := s.store.get(r.PathValue("id"))
	if !ok {
		http.NotFound(w, r)
		return
	}
	_, status, _ := job.snapshot(0)
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]string{"id": job.ID, "status": status})
}

func (s *server) streamLogs(w http.ResponseWriter, r *http.Request) {
	job, ok := s.store.get(r.PathValue("id"))
	if !ok {
		http.NotFound(w, r)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	w.WriteHeader(http.StatusOK)

	offset := 0
	for {
		chunk, status, notify := job.snapshot(offset)
		if len(chunk) > 0 {
			w.Write(chunk)
			flusher.Flush()
			offset += len(chunk)
			continue
		}
		if status != "running" {
			return
		}
		select {
		case <-notify:
		case <-r.Context().Done():
			return
		}
	}
}

func main() {
	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
	if err != nil {
		log.Fatal(err)
	}

	s := &server{store: newStore(), docker: cli}
	mux := http.NewServeMux()
	mux.HandleFunc("POST /builds", s.createBuild)
	mux.HandleFunc("GET /builds/{id}", s.getBuild)
	mux.HandleFunc("GET /builds/{id}/logs", s.streamLogs)

	log.Println("listening on :8081")
	log.Fatal(http.ListenAndServe(":8081", mux))
}