werkdock/internal/api/exec.go

package api

import (
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"

	"werkdock/internal/engine"
	"werkdock/internal/instance"
)

// stdout and stderr as docker's stream multiplexing numbers them.
const (
	streamStdout byte = 1
	streamStderr byte = 2
)

// frameWriter wraps a stream in docker's frame format: an 8-byte
// header per write, the stream number first and the payload length in
// the last four bytes, big-endian. Every frame is flushed, so a client
// following a log sees output as it happens.
type frameWriter struct {
	w      io.Writer
	stream byte
	flush  func()
}

func (f *frameWriter) Write(p []byte) (int, error) {
	if len(p) == 0 {
		return 0, nil
	}
	var header [8]byte
	header[0] = f.stream
	binary.BigEndian.PutUint32(header[4:], uint32(len(p)))
	if _, err := f.w.Write(header[:]); err != nil {
		return 0, err
	}
	n, err := f.w.Write(p)
	if f.flush != nil {
		f.flush()
	}
	return n, err
}

func framed(w http.ResponseWriter) (stdout, stderr io.Writer) {
	flush := func() {
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
	}
	return &frameWriter{w: w, stream: streamStdout, flush: flush},
		&frameWriter{w: w, stream: streamStderr, flush: flush}
}

// containerLogs streams the recorded output in docker's frame format.
func (s *Server) containerLogs(w http.ResponseWriter, r *http.Request) {
	inst, ok := s.instanceOf(w, r)
	if !ok {
		return
	}
	wantOut, wantErr := boolParam(r, "stdout"), boolParam(r, "stderr")
	if !wantOut && !wantErr {
		writeError(w, http.StatusBadRequest, "logs needs stdout or stderr")
		return
	}
	// Parameters that would change what the caller gets are refused
	// rather than ignored: a caller who asked for the last ten lines
	// and silently received ten thousand is worse off than one who was
	// told no (RFC 0002).
	if err := checkLogParams(r); err != nil {
		writeError(w, http.StatusNotImplemented, "%v", err)
		return
	}
	w.Header().Set("Content-Type", "application/octet-stream")
	w.WriteHeader(http.StatusOK)
	stdout, stderr := framed(w)
	if !wantOut {
		stdout = io.Discard
	}
	if !wantErr {
		stderr = io.Discard
	}
	stop := make(chan struct{})
	done := make(chan struct{})
	go func() {
		select {
		case <-r.Context().Done():
			close(stop)
		case <-done:
		}
	}()
	defer close(done)
	if err := s.Manager.Logs(inst, stdout, stderr, boolParam(r, "follow"), stop); err != nil {
		s.logf("logs %s: %v", inst.ShortID(), err)
	}
}

// checkLogParams refuses the log parameters werkdock cannot honour.
// Werkdock records the output as the command wrote it, without
// timestamps, so anything that selects by time or by line count has
// nothing to select on.
func checkLogParams(r *http.Request) error {
	q := r.URL.Query()
	if tail := q.Get("tail"); tail != "" && tail != "all" && tail != "-1" {
		return fmt.Errorf("logs tail=%s is not available: werkdock streams the recorded output whole; ask for all of it and cut it yourself", tail)
	}
	if boolParam(r, "timestamps") {
		return errors.New("logs timestamps are not available: werkdock records the output as written, without a clock beside it")
	}
	for _, name := range []string{"since", "until"} {
		if v := q.Get(name); v != "" && v != "0" {
			return fmt.Errorf("logs %s=%s is not available: werkdock records no timestamps to select by", name, v)
		}
	}
	return nil
}

// execCreateRequest is docker's exec create body.
type execCreateRequest struct {
	Cmd          []string `json:"Cmd"`
	Env          []string `json:"Env"`
	User         string   `json:"User"`
	WorkingDir   string   `json:"WorkingDir"`
	AttachStdout bool     `json:"AttachStdout"`
	AttachStderr bool     `json:"AttachStderr"`
	Tty          bool     `json:"Tty"`
	Privileged   bool     `json:"Privileged"`
}

func (s *Server) execCreate(w http.ResponseWriter, r *http.Request) {
	inst, ok := s.instanceOf(w, r)
	if !ok {
		return
	}
	var req execCreateRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		writeError(w, http.StatusBadRequest, "invalid exec body: %v", err)
		return
	}
	if len(req.Cmd) == 0 {
		writeError(w, http.StatusBadRequest, "exec needs a Cmd")
		return
	}
	if !inst.Running() {
		writeError(w, http.StatusConflict, "container %s is not running", inst.ShortID())
		return
	}
	var env []engine.EnvVar
	for _, e := range req.Env {
		key, value, _ := strings.Cut(e, "=")
		env = append(env, engine.EnvVar{Key: key, Value: value})
	}
	rec, err := s.Manager.CreateExec(inst, instance.Exec{
		Cmd: req.Cmd, Env: env, Workdir: req.WorkingDir, User: req.User,
	})
	if err != nil {
		writeError(w, http.StatusInternalServerError, "%v", err)
		return
	}
	writeJSON(w, http.StatusCreated, map[string]string{"Id": rec.ID})
}

// execStart runs the recorded command and streams its output framed.
// Docker hijacks the connection here only for an attached stdin; with
// AttachStdout/Stderr alone it is a plain response body, which is what
// a client sends (docs/testcontainers-api.md).
func (s *Server) execStart(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	rec, inst, err := s.Manager.GetExec(id)
	if err != nil {
		if errors.Is(err, instance.ErrNotFound) {
			writeError(w, http.StatusNotFound, "no such exec: %s", id)
			return
		}
		writeError(w, http.StatusInternalServerError, "%v", err)
		return
	}
	var body struct {
		Detach bool `json:"Detach"`
		Tty    bool `json:"Tty"`
	}
	json.NewDecoder(r.Body).Decode(&body)
	if body.Detach {
		writeError(w, http.StatusNotImplemented,
			"a detached exec has no instance to record it; werkdock runs an exec attached (werkdock run -d starts a second service)")
		return
	}
	w.Header().Set("Content-Type", "application/vnd.docker.raw-stream")
	w.WriteHeader(http.StatusOK)
	stdout, stderr := framed(w)
	code, err := s.Manager.RunExec(rec, inst, &engine.Bwrap{}, stdout, stderr)
	if err != nil {
		s.logf("exec %s: %v", id[:12], err)
		return
	}
	s.logf("exec %s in %s exited %d", id[:12], inst.ShortID(), code)
}

func (s *Server) execInspect(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	rec, inst, err := s.Manager.GetExec(id)
	if err != nil {
		if errors.Is(err, instance.ErrNotFound) {
			writeError(w, http.StatusNotFound, "no such exec: %s", id)
			return
		}
		writeError(w, http.StatusInternalServerError, "%v", err)
		return
	}
	entrypoint := ""
	var args []string
	if len(rec.Cmd) > 0 {
		entrypoint, args = rec.Cmd[0], rec.Cmd[1:]
	}
	writeJSON(w, http.StatusOK, map[string]interface{}{
		"ID":          rec.ID,
		"ContainerID": inst.Config.ID,
		"Running":     rec.Running,
		"ExitCode":    rec.ExitCode,
		"OpenStdin":   false,
		"OpenStdout":  true,
		"OpenStderr":  true,
		"CanRemove":   true,
		"DetachKeys":  "",
		"Pid":         0,
		"ProcessConfig": map[string]interface{}{
			"entrypoint": entrypoint,
			"arguments":  args,
			"privileged": false,
			"tty":        false,
			"user":       rec.User,
		},
	})
}