mqtt-security-test/docker/viewer/main.go

// Live log viewer for the enrollment demo. Tails /logs/*.log (written by the backend and the two
// boards) and streams the parsed lines to the web page over Server-Sent Events. It can also restart
// the whole demo (the "Run again" button) by talking to the Docker Engine API. Stdlib only.
package main

import (
	"context"
	"embed"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"
)

//go:embed index.html
var assets embed.FS

// line is one parsed log line: "HH:MM:SS.mmm [actor] message". Seq is its arrival order.
type line struct {
	Seq  int    `json:"seq"`
	Time string `json:"time"`
	Tag  string `json:"tag"`
	Msg  string `json:"msg"`
}

// status is what the page shows next to the "Run again" button.
type status struct {
	Text string `json:"text"`
	Busy bool   `json:"busy"`
}

var (
	mu     sync.Mutex
	lines  []line
	gen    int // incremented whenever the history is cleared (a new run), so clients can reset too
	state  = status{Text: "ready"}
	logDir = "/logs"
	// The restart button needs the Docker socket; it is only enabled when the deployment says so.
	restartEnabled = os.Getenv("RESTART_ENABLED") == "1"
	re             = regexp.MustCompile(`^(\d\d:\d\d:\d\d\.\d{3}) \[([a-z-]+)\] (.*)$`)
)

// tail remembers how far a file has been read and any incomplete last line.
type tail struct {
	offset  int64
	partial string
}

// tailLoop polls every *.log file in dir every 250 ms, parses the new complete lines and appends
// them to the shared slice. If a file shrinks (a new run truncated it) it is read from the start,
// and after a reset all read positions are forgotten.
func tailLoop(dir string) {
	files := map[string]*tail{}
	seenGen := 0
	for {
		mu.Lock()
		if gen != seenGen {
			files, seenGen = map[string]*tail{}, gen
		}
		mu.Unlock()

		paths, _ := filepath.Glob(filepath.Join(dir, "*.log"))
		for _, p := range paths {
			t := files[p]
			if t == nil {
				t = &tail{}
				files[p] = t
			}
			st, err := os.Stat(p)
			if err != nil {
				continue
			}
			if st.Size() < t.offset { // file was truncated by a new run
				t.offset, t.partial = 0, ""
			}
			if st.Size() == t.offset {
				continue
			}
			f, err := os.Open(p)
			if err != nil {
				continue
			}
			buf := make([]byte, st.Size()-t.offset)
			n, _ := f.ReadAt(buf, t.offset)
			f.Close()
			t.offset += int64(n)
			data := t.partial + string(buf[:n])
			parts := strings.Split(data, "\n")
			t.partial = parts[len(parts)-1]
			for _, l := range parts[:len(parts)-1] {
				if m := re.FindStringSubmatch(l); m != nil {
					mu.Lock()
					lines = append(lines, line{Seq: len(lines), Time: m[1], Tag: m[2], Msg: m[3]})
					mu.Unlock()
				}
			}
		}
		time.Sleep(250 * time.Millisecond)
	}
}

func setStatus(text string, busy bool) {
	mu.Lock()
	state = status{Text: text, Busy: busy}
	mu.Unlock()
}

// clearHistory forgets every line and deletes the log files, so the next run starts clean.
func clearHistory() {
	mu.Lock()
	lines = nil
	gen++
	mu.Unlock()
	old, _ := filepath.Glob(filepath.Join(logDir, "*.log"))
	for _, p := range old {
		os.Remove(p)
	}
}

// ---- Docker Engine API (over the mounted unix socket) ----

type docker struct{ c *http.Client }

func newDocker() docker {
	return docker{c: &http.Client{
		Timeout: 60 * time.Second,
		Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
			return (&net.Dialer{}).DialContext(ctx, "unix", "/var/run/docker.sock")
		}},
	}}
}

func (d docker) call(method, path string) ([]byte, int, error) {
	req, err := http.NewRequest(method, "http://docker"+path, nil)
	if err != nil {
		return nil, 0, err
	}
	resp, err := d.c.Do(req)
	if err != nil {
		return nil, 0, err
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return b, resp.StatusCode, nil
}

// containers returns compose service name -> container id for this demo's project.
func (d docker) containers(project string) (map[string]string, error) {
	filter := url.QueryEscape(fmt.Sprintf(`{"label":["com.docker.compose.project=%s"]}`, project))
	b, code, err := d.call("GET", "/containers/json?all=1&filters="+filter)
	if err != nil {
		return nil, err
	}
	if code != 200 {
		return nil, fmt.Errorf("docker returned %d: %s", code, strings.TrimSpace(string(b)))
	}
	var list []struct {
		ID     string            `json:"Id"`
		Labels map[string]string `json:"Labels"`
	}
	if err := json.Unmarshal(b, &list); err != nil {
		return nil, err
	}
	out := map[string]string{}
	for _, c := range list {
		out[c.Labels["com.docker.compose.service"]] = c.ID
	}
	return out, nil
}

func (d docker) action(id, verb string) error {
	_, code, err := d.call("POST", "/containers/"+id+"/"+verb+"?t=2")
	if err != nil {
		return err
	}
	if code >= 300 && code != 304 { // 304 = already in that state
		return fmt.Errorf("docker %s returned %d", verb, code)
	}
	return nil
}

func (d docker) healthy(id string) bool {
	b, code, err := d.call("GET", "/containers/"+id+"/json")
	if err != nil || code != 200 {
		return false
	}
	var info struct {
		State struct {
			Running bool
			Health  *struct{ Status string }
		}
	}
	if json.Unmarshal(b, &info) != nil {
		return false
	}
	if info.State.Health != nil {
		return info.State.Health.Status == "healthy"
	}
	return info.State.Running
}

// restartDemo performs a clean re-run: stop the backend and boards, restart the broker (which
// recreates its Dynamic Security state), clear the old logs, then start the backend and boards
// again. Their own start delays then play out exactly like the first run.
func restartDemo() {
	project := os.Getenv("COMPOSE_PROJECT")
	if project == "" {
		project = "crypto-interop-spike"
	}
	d := newDocker()
	ids, err := d.containers(project)
	if err != nil {
		setStatus("cannot reach Docker: "+err.Error(), false)
		return
	}
	for _, s := range []string{"backend", "board-c", "board-cpp", "broker"} {
		if ids[s] == "" {
			setStatus("container for service "+s+" not found", false)
			return
		}
	}

	setStatus("stopping the backend and boards...", true)
	for _, s := range []string{"board-c", "board-cpp", "backend"} {
		if err := d.action(ids[s], "stop"); err != nil {
			setStatus("stop "+s+": "+err.Error(), false)
			return
		}
	}

	setStatus("restarting the broker (fresh Dynamic Security state)...", true)
	if err := d.action(ids["broker"], "restart"); err != nil {
		setStatus("restart broker: "+err.Error(), false)
		return
	}
	ok := false
	for i := 0; i < 60 && !ok; i++ {
		time.Sleep(500 * time.Millisecond)
		ok = d.healthy(ids["broker"])
	}
	if !ok {
		setStatus("the broker did not become healthy", false)
		return
	}

	setStatus("clearing the old logs...", true)
	clearHistory()

	setStatus("starting the backend and boards...", true)
	for _, s := range []string{"backend", "board-c", "board-cpp"} {
		if err := d.action(ids[s], "start"); err != nil {
			setStatus("start "+s+": "+err.Error(), false)
			return
		}
	}
	setStatus("new run started", false)
}

// main: clear stale logs from the previous run, start the tailer, then serve:
//
//	/             the web page (embedded index.html)
//	/api/lines    all lines so far as JSON (handy for debugging)
//	/api/restart  POST: restart the whole demo (needs the X-Demo-Action header, see below)
//	/events       Server-Sent Events: replays every line, then pushes new ones live, plus
//	              "reset" (history cleared) and "status" (restart progress) events
func main() {
	if len(os.Args) > 1 {
		logDir = os.Args[1]
	}
	os.MkdirAll(logDir, 0o755)
	clearHistory() // stale logs from a previous run
	go tailLoop(logDir)

	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/" {
			http.NotFound(w, r)
			return
		}
		b, _ := assets.ReadFile("index.html")
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		w.Write(b)
	})
	http.HandleFunc("/api/lines", func(w http.ResponseWriter, r *http.Request) {
		mu.Lock()
		defer mu.Unlock()
		json.NewEncoder(w).Encode(lines)
	})
	http.HandleFunc("/api/config", func(w http.ResponseWriter, r *http.Request) {
		json.NewEncoder(w).Encode(map[string]bool{"restart": restartEnabled})
	})
	http.HandleFunc("/api/restart", func(w http.ResponseWriter, r *http.Request) {
		if !restartEnabled {
			http.NotFound(w, r)
			return
		}
		// The custom header cannot be sent by a cross-site form or fetch without a CORS preflight
		// (which this server never grants), so other web pages cannot trigger a restart.
		if r.Method != http.MethodPost || r.Header.Get("X-Demo-Action") != "restart" {
			http.Error(w, "forbidden", http.StatusForbidden)
			return
		}
		mu.Lock()
		if state.Busy {
			mu.Unlock()
			http.Error(w, "a restart is already running", http.StatusConflict)
			return
		}
		state = status{Text: "starting restart...", Busy: true}
		mu.Unlock()
		go restartDemo()
		w.WriteHeader(http.StatusAccepted)
	})
	http.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
		fl, ok := w.(http.Flusher)
		if !ok {
			http.Error(w, "streaming unsupported", http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "text/event-stream")
		w.Header().Set("Cache-Control", "no-cache")
		idx, myGen, sent := 0, -1, status{Text: "\x00"}
		for {
			mu.Lock()
			if myGen == -1 {
				myGen = gen
			}
			resetNow := gen != myGen
			if resetNow {
				myGen, idx = gen, 0
			}
			batch := append([]line(nil), lines[min(idx, len(lines)):]...)
			cur := state
			mu.Unlock()

			if resetNow {
				fmt.Fprint(w, "event: reset\ndata: {}\n\n")
			}
			for _, l := range batch {
				b, _ := json.Marshal(l)
				fmt.Fprintf(w, "data: %s\n\n", b)
			}
			idx += len(batch)
			if cur != sent {
				b, _ := json.Marshal(cur)
				fmt.Fprintf(w, "event: status\ndata: %s\n\n", b)
				sent = cur
			}
			fl.Flush()
			select {
			case <-r.Context().Done():
				return
			case <-time.After(250 * time.Millisecond):
			}
		}
	})
	fmt.Println("viewer listening on :8080")
	http.ListenAndServe(":8080", nil)
}