Beach/internal/beach/docker.go

package beach

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"time"
)

const defaultDockerHost = "/var/run/docker.sock"

// Container is one row in the containers table.
type Container struct {
	ID      string   `json:"id"`
	Name    string   `json:"name"`
	Image   string   `json:"image"`
	State   string   `json:"state"`  // running, exited, ...
	Status  string   `json:"status"` // "Up 3 hours (healthy)"
	Ports   []string `json:"ports"`
	Created int64    `json:"created"`
}

// Docker talks to the Engine API over a unix socket or plain TCP.
type Docker struct {
	base   string
	client *http.Client
}

// NewDocker accepts "/var/run/docker.sock", "unix:///path", "tcp://host:2375" or "host:2375".
func NewDocker(host string) *Docker {
	host = strings.TrimSpace(host)
	d := &Docker{base: "http://docker"}
	tr := &http.Transport{}
	switch {
	case strings.HasPrefix(host, "tcp://"), strings.HasPrefix(host, "http://"):
		d.base = "http://" + strings.TrimPrefix(strings.TrimPrefix(host, "tcp://"), "http://")
	case strings.HasPrefix(host, "/"), strings.HasPrefix(host, "unix://"):
		path := strings.TrimPrefix(host, "unix://")
		tr.DialContext = func(ctx context.Context, _, _ string) (net.Conn, error) {
			return (&net.Dialer{Timeout: 3 * time.Second}).DialContext(ctx, "unix", path)
		}
	default:
		d.base = "http://" + host
	}
	d.client = &http.Client{Transport: tr, Timeout: 30 * time.Second}
	return d
}

func (d *Docker) do(ctx context.Context, method, path string) ([]byte, error) {
	req, err := http.NewRequestWithContext(ctx, method, d.base+path, nil)
	if err != nil {
		return nil, err
	}
	resp, err := d.client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(io.LimitReader(resp.Body, 16<<20))
	if resp.StatusCode >= 300 && resp.StatusCode != http.StatusNotModified {
		var e struct{ Message string }
		json.Unmarshal(body, &e)
		if e.Message == "" {
			e.Message = resp.Status
		}
		return nil, fmt.Errorf("docker: %s", e.Message)
	}
	return body, nil
}

func (d *Docker) List(ctx context.Context) ([]Container, error) {
	body, err := d.do(ctx, "GET", "/containers/json?all=1")
	if err != nil {
		return nil, err
	}
	return parseContainers(body)
}

func parseContainers(body []byte) ([]Container, error) {
	var raw []struct {
		ID      string
		Names   []string
		Image   string
		State   string
		Status  string
		Created int64
		Ports   []struct {
			PrivatePort int
			PublicPort  int
			Type        string
		}
	}
	if err := json.Unmarshal(body, &raw); err != nil {
		return nil, err
	}
	out := make([]Container, 0, len(raw))
	for _, r := range raw {
		c := Container{ID: r.ID, Image: r.Image, State: r.State, Status: r.Status, Created: r.Created, Ports: []string{}}
		if len(r.Names) > 0 {
			c.Name = strings.TrimPrefix(r.Names[0], "/")
		}
		seen := map[string]bool{}
		for _, p := range r.Ports {
			if p.PublicPort == 0 {
				continue
			}
			s := fmt.Sprintf("%d:%d/%s", p.PublicPort, p.PrivatePort, p.Type)
			if !seen[s] { // IPv4 and IPv6 bindings repeat
				seen[s] = true
				c.Ports = append(c.Ports, s)
			}
		}
		out = append(out, c)
	}
	sort.Slice(out, func(i, j int) bool {
		if (out[i].State == "running") != (out[j].State == "running") {
			return out[i].State == "running"
		}
		return out[i].Name < out[j].Name
	})
	return out, nil
}

// Action is one of start, stop, restart.
func (d *Docker) Action(ctx context.Context, id, action string) error {
	switch action {
	case "start", "stop", "restart":
	default:
		return fmt.Errorf("unknown action %q", action)
	}
	_, err := d.do(ctx, "POST", "/containers/"+url.PathEscape(id)+"/"+action)
	return err
}

func (d *Docker) Logs(ctx context.Context, id string, tail int) (string, error) {
	if tail < 1 || tail > 5000 {
		tail = 200
	}
	// the multiplexed format is only used for containers without a TTY
	var info struct{ Config struct{ Tty bool } }
	if b, err := d.do(ctx, "GET", "/containers/"+url.PathEscape(id)+"/json"); err == nil {
		json.Unmarshal(b, &info)
	}
	body, err := d.do(ctx, "GET", fmt.Sprintf("/containers/%s/logs?stdout=1&stderr=1&timestamps=1&tail=%d", url.PathEscape(id), tail))
	if err != nil {
		return "", err
	}
	if info.Config.Tty {
		return string(body), nil
	}
	return demux(body), nil
}

// demux strips the 8-byte stream headers Docker puts in front of each log frame.
func demux(b []byte) string {
	var out bytes.Buffer
	for len(b) >= 8 {
		if b[0] > 2 || b[1] != 0 || b[2] != 0 || b[3] != 0 { // not framed; treat as plain text
			out.Write(b)
			return out.String()
		}
		n := int(binary.BigEndian.Uint32(b[4:8]))
		b = b[8:]
		if n > len(b) {
			n = len(b)
		}
		out.Write(b[:n])
		b = b[n:]
	}
	return out.String()
}