hamlab/internal/api/docker_events.go

package api

// A container's history.
//
// `docker events` is a live stream with only a short in-memory backlog — ask it for the
// last thirty days and you get the last few minutes, because the daemon keeps events in a
// small ring and loses them all when it restarts. That is why Dockhand kept its own copy,
// and why this does too: a watcher follows the stream and appends the lifecycle events to a
// ring kept beside the config, so "what happened to this container last Tuesday" has an
// answer.
//
// Only lifecycle events are kept. A health check produces three exec events every probe,
// which is thousands a day and tells you nothing that the health panel does not.

import (
	"bufio"
	"context"
	"encoding/json"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"hamlab/internal/priv"
)

const (
	eventsKept     = 4000
	eventsSaveEeem = 30 * time.Second
)

// ContainerEvent is one thing that happened to one container.
type ContainerEvent struct {
	T        int64  `json:"t"` // unix ms
	Name     string `json:"name"`
	Action   string `json:"action"`
	Image    string `json:"image,omitempty"`
	ExitCode string `json:"exit_code,omitempty"`
	Signal   string `json:"signal,omitempty"`
	Health   string `json:"health,omitempty"`
}

type eventLog struct {
	mu    sync.Mutex
	ring  []ContainerEvent
	dirty bool
	path  string
}

func (e *eventLog) load(path string) {
	e.mu.Lock()
	defer e.mu.Unlock()
	e.path = path
	b, err := os.ReadFile(path)
	if err != nil {
		return
	}
	var list []ContainerEvent
	if json.Unmarshal(b, &list) == nil {
		e.ring = list
	}
}

func (e *eventLog) add(ev ContainerEvent) {
	e.mu.Lock()
	defer e.mu.Unlock()
	e.ring = append(e.ring, ev)
	if len(e.ring) > eventsKept {
		e.ring = append([]ContainerEvent(nil), e.ring[len(e.ring)-eventsKept:]...)
	}
	e.dirty = true
}

// forName is one container's events, newest first.
func (e *eventLog) forName(name string, since time.Time, limit int) []ContainerEvent {
	e.mu.Lock()
	defer e.mu.Unlock()
	out := []ContainerEvent{}
	for _, ev := range e.ring {
		if ev.Name == name && ev.T >= since.UnixMilli() {
			out = append(out, ev)
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].T > out[j].T })
	if len(out) > limit {
		out = out[:limit]
	}
	return out
}

func (e *eventLog) save() {
	e.mu.Lock()
	if !e.dirty || e.path == "" {
		e.mu.Unlock()
		return
	}
	b, err := json.Marshal(e.ring)
	e.dirty = false
	path := e.path
	e.mu.Unlock()
	if err != nil {
		return
	}
	_ = os.MkdirAll(filepath.Dir(path), 0o700)
	tmp := path + ".tmp"
	if os.WriteFile(tmp, b, 0o600) == nil {
		_ = os.Rename(tmp, path)
	}
}

// watchDockerEvents follows the stream, restarting if Docker goes away. It is the only
// thing that writes the log, so nothing contends with it.
func (s *Server) watchDockerEvents(ctx context.Context) {
	s.events.load(filepath.Join(s.cfg.DataDir(), "docker-events.json"))
	go func() { // a periodic flush, so a crash loses at most half a minute of history
		t := time.NewTicker(eventsSaveEeem)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				s.events.save()
				return
			case <-t.C:
				s.events.save()
			}
		}
	}()
	for {
		if ctx.Err() != nil {
			return
		}
		if dockerTool() == "" {
			time.Sleep(time.Minute) // no Docker here; look again later in case it is installed
			continue
		}
		s.followEvents(ctx)
		select {
		case <-ctx.Done():
			return
		case <-time.After(10 * time.Second): // the daemon restarted, or the pipe broke
		}
	}
}

func (s *Server) followEvents(ctx context.Context) {
	args := []string{"events", "--format", "{{json .}}", "--filter", "type=container"}
	for _, e := range []string{"create", "start", "restart", "stop", "kill", "die", "destroy", "oom", "pause", "unpause", "rename", "update", "health_status"} {
		args = append(args, "--filter", "event="+e)
	}
	tool := dockerTool()
	full := append([]string{tool}, args...)
	if !dockerReachable() && priv.Available() {
		full = priv.Wrap(full...)
	}
	cmd := exec.CommandContext(ctx, full[0], full[1:]...)
	out, err := cmd.StdoutPipe()
	if err != nil {
		return
	}
	if err := cmd.Start(); err != nil {
		return
	}
	sc := bufio.NewScanner(out)
	sc.Buffer(make([]byte, 64<<10), 1<<20)
	n := 0
	for sc.Scan() {
		var e struct {
			Action   string `json:"Action"`
			Time     int64  `json:"time"`
			TimeNano int64  `json:"timeNano"`
			Actor    struct {
				Attributes map[string]string `json:"Attributes"`
			} `json:"Actor"`
		}
		if json.Unmarshal(sc.Bytes(), &e) != nil || e.Action == "" {
			continue
		}
		at := e.Actor.Attributes
		ev := ContainerEvent{
			T: e.Time * 1000, Name: at["name"], Image: at["image"],
			Action: e.Action, ExitCode: at["exitCode"], Signal: at["signal"],
		}
		if e.TimeNano > 0 {
			ev.T = e.TimeNano / 1e6
		}
		// health_status comes through as "health_status: healthy"
		if a, h, ok := strings.Cut(e.Action, ": "); ok {
			ev.Action, ev.Health = a, h
		}
		if ev.Name == "" {
			continue
		}
		s.events.add(ev)
		n++
	}
	_ = cmd.Wait()
	if n > 0 {
		log.Printf("docker events: stream ended after %d events; reconnecting", n)
	}
}

// dockerEvents answers from Ham Lab's own history, not Docker's short memory.
func (s *Server) dockerEvents(w http.ResponseWriter, r *http.Request) {
	id := r.PathValue("id")
	if !dockerRef.MatchString(id) {
		jsonErr(w, 400, "bad container")
		return
	}
	since := r.URL.Query().Get("since")
	d := 7 * 24 * time.Hour
	switch since {
	case "24h":
		d = 24 * time.Hour
	case "30d":
		d = 30 * 24 * time.Hour
	default:
		since = "7d"
	}
	ev := s.events.forName(id, time.Now().Add(-d), 500)
	s.events.mu.Lock()
	total := len(s.events.ring)
	s.events.mu.Unlock()
	jsonOK(w, 200, map[string]any{"events": ev, "since": since, "kept": total, "capacity": eventsKept})
}