werkdock/tools/apirecord/main.go

// Command apirecord is a recording proxy between a docker API client
// and an engine socket: it speaks HTTP/1.1 on a unix socket, forwards
// every request upstream unchanged, and writes one JSON line per
// exchange.
//
// It exists to answer "what does Testcontainers actually send?" from
// measurement rather than from the Engine API documentation, and to
// compare werkdock's own answers with podman's later
// (docs/testcontainers-api.md, RFC 0005 milestone 3). It is a
// development tool, not part of the werkdock binary.
//
//	go run ./tools/apirecord -listen /tmp/rec.sock \
//	    -upstream $XDG_RUNTIME_DIR/podman/podman.sock -trace trace.jsonl
//	DOCKER_HOST=unix:///tmp/rec.sock TESTCONTAINERS_RYUK_DISABLED=true ...
package main

import (
	"bufio"
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

var (
	listenPath = flag.String("listen", "", "unix socket to listen on")
	upstream   = flag.String("upstream", "", "unix socket to forward to")
	traceFile  = flag.String("trace", "trace.jsonl", "where to write the JSON lines")
	maxBody    = flag.Int("max-body", 4000, "bytes of a body to record")
)

type exchange struct {
	Seq        int64             `json:"seq"`
	At         string            `json:"at"`
	Method     string            `json:"method"`
	Path       string            `json:"path"`
	Query      string            `json:"query,omitempty"`
	ReqHeaders map[string]string `json:"reqHeaders,omitempty"`
	ReqBody    string            `json:"reqBody,omitempty"`
	ReqBytes   int               `json:"reqBytes,omitempty"`
	Status     int               `json:"status"`
	RespType   string            `json:"respType,omitempty"`
	RespBody   string            `json:"respBody,omitempty"`
	RespBytes  int64             `json:"respBytes"`
	Upgraded   bool              `json:"upgraded,omitempty"`
	DurationMS int64             `json:"durationMs"`
	Error      string            `json:"error,omitempty"`
}

var (
	seq   int64
	mu    sync.Mutex
	trace *os.File
)

func record(e exchange) {
	data, err := json.Marshal(e)
	if err != nil {
		log.Printf("marshal: %v", err)
		return
	}
	mu.Lock()
	defer mu.Unlock()
	fmt.Fprintln(trace, string(data))
	trace.Sync()
	fmt.Printf("%4d  %-6s %-58s -> %d  %s\n", e.Seq, e.Method, truncate(e.Path+e.Query, 58), e.Status, human(e.RespBytes))
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n-1] + "…"
}

func human(n int64) string {
	if n < 1024 {
		return fmt.Sprintf("%dB", n)
	}
	return fmt.Sprintf("%dkB", n/1024)
}

// interesting headers, the ones that shape the protocol
var keepHeaders = []string{"Content-Type", "Connection", "Upgrade", "User-Agent", "Api-Version"}

func selected(h http.Header) map[string]string {
	out := map[string]string{}
	for _, k := range keepHeaders {
		if v := h.Get(k); v != "" {
			out[k] = v
		}
	}
	return out
}

func main() {
	flag.Parse()
	if *listenPath == "" || *upstream == "" {
		log.Fatal("need -listen and -upstream")
	}
	var err error
	trace, err = os.Create(*traceFile)
	if err != nil {
		log.Fatal(err)
	}
	defer trace.Close()
	os.Remove(*listenPath)
	ln, err := net.Listen("unix", *listenPath)
	if err != nil {
		log.Fatal(err)
	}
	defer ln.Close()
	log.Printf("recording %s -> %s into %s", *listenPath, *upstream, *traceFile)
	for {
		conn, err := ln.Accept()
		if err != nil {
			log.Printf("accept: %v", err)
			return
		}
		go handle(conn)
	}
}

func handle(client net.Conn) {
	defer client.Close()
	up, err := net.Dial("unix", *upstream)
	if err != nil {
		log.Printf("dial upstream: %v", err)
		return
	}
	defer up.Close()
	cr := bufio.NewReader(client)
	ur := bufio.NewReader(up)
	for {
		req, err := http.ReadRequest(cr)
		if err != nil {
			if err != io.EOF && !strings.Contains(err.Error(), "closed") {
				log.Printf("read request: %v", err)
			}
			return
		}
		e := exchange{
			Seq:        atomic.AddInt64(&seq, 1),
			At:         time.Now().Format("15:04:05.000"),
			Method:     req.Method,
			Path:       req.URL.Path,
			ReqHeaders: selected(req.Header),
		}
		if q := req.URL.RawQuery; q != "" {
			e.Query = "?" + q
		}
		body, err := io.ReadAll(req.Body)
		req.Body.Close()
		if err != nil {
			e.Error = "read request body: " + err.Error()
		}
		e.ReqBytes = len(body)
		e.ReqBody = preview(body, req.Header.Get("Content-Type"), *maxBody)
		req.Body = io.NopCloser(bytes.NewReader(body))
		req.ContentLength = int64(len(body))
		start := time.Now()
		if err := req.Write(up); err != nil {
			e.Error = "forward: " + err.Error()
			record(e)
			return
		}
		resp, err := http.ReadResponse(ur, req)
		if err != nil {
			e.Error = "read response: " + err.Error()
			record(e)
			return
		}
		e.Status = resp.StatusCode
		e.RespType = resp.Header.Get("Content-Type")
		if resp.StatusCode == http.StatusSwitchingProtocols {
			e.Upgraded = true
			e.DurationMS = time.Since(start).Milliseconds()
			record(e)
			resp.Write(client)
			rawCopy(client, cr, up, ur)
			return
		}
		var captured bytes.Buffer
		counter := &countingWriter{limit: *maxBody, buf: &captured}
		resp.Body = io.NopCloser(io.TeeReader(resp.Body, counter))
		werr := resp.Write(client)
		e.RespBytes = counter.total
		e.RespBody = preview(captured.Bytes(), e.RespType, *maxBody)
		e.DurationMS = time.Since(start).Milliseconds()
		if werr != nil {
			e.Error = "write response: " + werr.Error()
		}
		record(e)
		if werr != nil || resp.Close || req.Close {
			return
		}
	}
}

type countingWriter struct {
	limit int
	total int64
	buf   *bytes.Buffer
}

func (w *countingWriter) Write(p []byte) (int, error) {
	w.total += int64(len(p))
	if w.buf.Len() < w.limit {
		n := w.limit - w.buf.Len()
		if n > len(p) {
			n = len(p)
		}
		w.buf.Write(p[:n])
	}
	return len(p), nil
}

// preview renders a body for the trace: text as text, binary as a note.
func preview(body []byte, contentType string, max int) string {
	if len(body) == 0 {
		return ""
	}
	if !isText(body) {
		n := len(body)
		if n > 512 {
			n = 512
		}
		return fmt.Sprintf("<binary %d bytes> hex:%x", len(body), body[:n])
	}
	s := string(body)
	if len(s) > max {
		s = s[:max] + fmt.Sprintf("…(%d bytes total)", len(body))
	}
	return s
}

func isText(b []byte) bool {
	n := len(b)
	if n > 512 {
		n = 512
	}
	for _, c := range b[:n] {
		if c == 0 {
			return false
		}
	}
	return true
}

// rawCopy streams both directions after an upgrade, buffered bytes first.
func rawCopy(client net.Conn, cr *bufio.Reader, up net.Conn, ur *bufio.Reader) {
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		io.Copy(up, cr)
		if c, ok := up.(*net.UnixConn); ok {
			c.CloseWrite()
		}
	}()
	go func() {
		defer wg.Done()
		io.Copy(client, ur)
		if c, ok := client.(*net.UnixConn); ok {
			c.CloseWrite()
		}
	}()
	wg.Wait()
}