Serve the rows on a read-only socket and web page, and print them with mesh-delivery-cli (issue 1134)

The command holds no credential: it asks the local socket, which answers only the rows ask.

modules/mesh-delivery-view/cmd/mesh-delivery-cli/main.go

@@ -0,0 +1,52 @@
+// Command mesh-delivery-cli is the operator's command for the delivery view (novox/hq ADR 0429 §3). It asks the
+// view's local socket for the rows and prints them; it holds no credential and changes nothing.
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+
+	"mesh-delivery-view/internal/view"
+)
+
+const defaultSocket = "/run/mesh-delivery-view/cli.sock"
+
+func main() {
+	socket := os.Getenv("MESH_DELIVERY_SOCKET")
+	if socket == "" {
+		socket = defaultSocket
+	}
+	os.Exit(run(os.Args[1:], socket, os.Stdout, os.Stderr))
+}
+
+func run(args []string, socket string, out, errOut io.Writer) int {
+	fs := flag.NewFlagSet("mesh-delivery-cli", flag.ContinueOnError)
+	fs.SetOutput(errOut)
+	asJSON := fs.Bool("json", false, "print the rows as JSON")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	if fs.NArg() > 0 {
+		fmt.Fprintln(errOut, "mesh-delivery-cli takes no argument: it only shows")
+		return 2
+	}
+	s, err := view.Ask(socket)
+	if err != nil {
+		fmt.Fprintln(errOut, "mesh-delivery-cli:", err)
+		return 1
+	}
+	if *asJSON {
+		enc := json.NewEncoder(out)
+		enc.SetIndent("", "  ")
+		if err := enc.Encode(s); err != nil {
+			fmt.Fprintln(errOut, "mesh-delivery-cli:", err)
+			return 1
+		}
+		return 0
+	}
+	view.Table(out, s)
+	return 0
+}

modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"mesh-delivery-view/internal/view"
+)
+
+func serve(t *testing.T) string {
+	t.Helper()
+	m := view.New()
+	m.Apply(view.Entry{Bucket: view.AssignmentsBucket, Key: "a.x", Value: []byte(`{"node":"a","module":"x","build":"b1"}`)})
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "c")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	l, err := view.ListenSocket(filepath.Join(dir, "cli.sock"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	go view.ServeSocket(ctx, l, m)
+	t.Cleanup(cancel)
+	return filepath.Join(dir, "cli.sock")
+}
+
+func TestPrintsATable(t *testing.T) { // 1134/AC6
+	var out, errOut bytes.Buffer
+	if code := run(nil, serve(t), &out, &errOut); code != 0 {
+		t.Fatalf("exit %d: %s", code, errOut.String())
+	}
+	if !strings.Contains(out.String(), "MODULE") || !strings.Contains(out.String(), "no have yet") {
+		t.Fatalf("table = %q", out.String())
+	}
+}
+
+func TestJSONFlag(t *testing.T) { // 1134/AC6
+	var out, errOut bytes.Buffer
+	if code := run([]string{"--json"}, serve(t), &out, &errOut); code != 0 {
+		t.Fatalf("exit %d: %s", code, errOut.String())
+	}
+	var s view.Snapshot
+	if err := json.Unmarshal(out.Bytes(), &s); err != nil || len(s.Rows) != 1 || s.Rows[0].Module != "x" {
+		t.Fatalf("json = %q, %v", out.String(), err)
+	}
+}
+
+func TestFailsLoudWhenTheViewIsNotThere(t *testing.T) { // 1134/AC6
+	var out, errOut bytes.Buffer
+	if code := run(nil, filepath.Join(os.Getenv("TMPDIR"), "none.sock"), &out, &errOut); code == 0 || errOut.Len() == 0 {
+		t.Fatalf("exit %d, stderr %q", code, errOut.String())
+	}
+	if code := run([]string{"--start"}, "", &out, &errOut); code != 2 {
+		t.Fatalf("an unknown flag must exit 2, got %d", code)
+	}
+}

modules/mesh-delivery-view/internal/view/dial_test.go

@@ -0,0 +1,5 @@
+package view
+
+import "net"
+
+func dial(path string) (net.Conn, error) { return net.Dial("unix", path) }

modules/mesh-delivery-view/internal/view/serve.go

@@ -0,0 +1,173 @@
+package view
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"html/template"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"text/tabwriter"
+	"time"
+)
+
+// ListenSocket listens on the unix socket at path, in a directory of mode 0755, with mode 0666 on the socket: the
+// status it carries is no secret and the socket can change nothing.
+func ListenSocket(path string) (net.Listener, error) {
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return nil, err
+	}
+	if err := os.Chmod(dir, 0o755); err != nil {
+		return nil, err
+	}
+	if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
+		return nil, err
+	}
+	l, err := net.Listen("unix", path)
+	if err != nil {
+		return nil, err
+	}
+	if err := os.Chmod(path, 0o666); err != nil {
+		l.Close()
+		return nil, err
+	}
+	return l, nil
+}
+
+type ask struct {
+	Ask string `json:"ask"`
+}
+
+// ServeSocket answers {"ask":"rows"} on l, line-delimited JSON, until ctx ends. Any other ask is refused.
+func ServeSocket(ctx context.Context, l net.Listener, m *Model) error {
+	go func() { <-ctx.Done(); l.Close() }()
+	for {
+		c, err := l.Accept()
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			return err
+		}
+		go answer(c, m)
+	}
+}
+
+func answer(c net.Conn, m *Model) {
+	defer c.Close()
+	c.SetDeadline(time.Now().Add(10 * time.Second))
+	line, err := bufio.NewReader(io.LimitReader(c, 4096)).ReadBytes('\n')
+	if err != nil && len(line) == 0 {
+		return
+	}
+	var a ask
+	if json.Unmarshal(line, &a) != nil || a.Ask != "rows" {
+		json.NewEncoder(c).Encode(map[string]string{"error": `refused: this socket answers {"ask":"rows"} and nothing else`})
+		return
+	}
+	json.NewEncoder(c).Encode(m.Snapshot())
+}
+
+// Ask asks the socket at path for the rows.
+func Ask(path string) (Snapshot, error) {
+	c, err := net.DialTimeout("unix", path, 5*time.Second)
+	if err != nil {
+		return Snapshot{}, fmt.Errorf("cannot reach the view at %s: %w", path, err)
+	}
+	defer c.Close()
+	c.SetDeadline(time.Now().Add(10 * time.Second))
+	if _, err := c.Write([]byte(`{"ask":"rows"}` + "\n")); err != nil {
+		return Snapshot{}, err
+	}
+	var reply struct {
+		Snapshot
+		Error string `json:"error"`
+	}
+	if err := json.NewDecoder(c).Decode(&reply); err != nil {
+		return Snapshot{}, fmt.Errorf("the view's answer is not rows: %w", err)
+	}
+	if reply.Error != "" {
+		return Snapshot{}, errors.New(reply.Error)
+	}
+	return reply.Snapshot, nil
+}
+
+func since(t time.Time) string {
+	if t.IsZero() {
+		return ""
+	}
+	return t.Local().Format("2006-01-02 15:04:05")
+}
+
+func rowText(r Row) []string {
+	what := r.Waits
+	if r.Error != "" {
+		what = "unreadable: " + r.Error
+	}
+	if r.Status == StatusLeftBehind && what == "" {
+		what = r.Status
+	} else if r.Status == StatusLeftBehind {
+		what = r.Status + "; " + what
+	}
+	return []string{r.Node, r.Module, r.Assigned, r.Have, r.Phase, r.Health, since(r.Since), r.Detail, what}
+}
+
+var head = []string{"NODE", "MODULE", "ASSIGNED", "HAVE", "PHASE", "HEALTH", "SINCE", "DETAIL", "WAITS"}
+
+// Table prints the snapshot as a table.
+func Table(w io.Writer, s Snapshot) {
+	tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
+	row := func(cells []string) {
+		for i, c := range cells {
+			if i > 0 {
+				fmt.Fprint(tw, "\t")
+			}
+			fmt.Fprint(tw, c)
+		}
+		fmt.Fprintln(tw)
+	}
+	row(head)
+	for _, r := range s.Rows {
+		row(rowText(r))
+	}
+	if len(s.Nodes) > 0 {
+		fmt.Fprintln(tw)
+		row([]string{"NODE", "WINDOWS", "BUSY", "STRAYS", "SINCE"})
+		for _, n := range s.Nodes {
+			row([]string{n.Node, fmt.Sprint(n.Windows), fmt.Sprint(n.Busy), fmt.Sprint(n.Strays), since(n.Since)})
+		}
+	}
+	tw.Flush()
+}
+
+var page = template.Must(template.New("p").Parse(`<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="5">
+<title>Delivery</title><style>body{font:14px sans-serif;margin:1em}table{border-collapse:collapse}td,th{padding:2px 10px;text-align:left}</style></head>
+<body><h1>What each node runs</h1><table><tr>{{range .Head}}<th>{{.}}</th>{{end}}</tr>
+{{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table></body></html>`))
+
+// Web is the read-only web view of the same rows, refreshed by the browser.
+func Web(m *Model) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet && r.Method != http.MethodHead {
+			w.Header().Set("Allow", "GET, HEAD")
+			http.Error(w, "this view is read-only", http.StatusMethodNotAllowed)
+			return
+		}
+		if r.URL.Path != "/" {
+			http.NotFound(w, r)
+			return
+		}
+		var rows [][]string
+		for _, row := range m.Snapshot().Rows {
+			rows = append(rows, rowText(row))
+		}
+		w.Header().Set("Content-Type", "text/html; charset=utf-8")
+		page.Execute(w, map[string]any{"Head": head, "Rows": rows})
+	})
+}

modules/mesh-delivery-view/internal/view/serve_test.go

@@ -0,0 +1,112 @@
+package view
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+func served(t *testing.T) (string, *Model) {
+	t.Helper()
+	m := New()
+	fill(m, base(t)...)
+	fill(m, haveE(t, "a", "w", have(t, "b9", delivery.PhaseRunning, delivery.HealthOK, ""), t0))
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "s")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	path := filepath.Join(dir, "run", "cli.sock")
+	l, err := ListenSocket(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() { ServeSocket(ctx, l, m); close(done) }()
+	t.Cleanup(func() { cancel(); l.Close(); <-done })
+	return path, m
+}
+
+func TestSocketAnswersRows(t *testing.T) { // 1134/AC6
+	path, m := served(t)
+	got, err := Ask(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := m.Snapshot()
+	b1, _ := json.Marshal(got)
+	b2, _ := json.Marshal(want)
+	if len(want.Rows) == 0 || string(b1) != string(b2) {
+		t.Fatalf("socket answered %s, want %s", b1, b2)
+	}
+	st, err := os.Stat(path)
+	if err != nil || st.Mode().Perm() != 0o666 {
+		t.Fatalf("socket mode = %v, %v; want 0666", st.Mode().Perm(), err)
+	}
+	dst, _ := os.Stat(filepath.Dir(path))
+	if dst.Mode().Perm() != 0o755 {
+		t.Fatalf("directory mode = %v, want 0755", dst.Mode().Perm())
+	}
+}
+
+func TestSocketRefusesAnyOtherAsk(t *testing.T) { // 1134/AC6
+	path, m := served(t)
+	before, _ := json.Marshal(m.Snapshot())
+	for _, ask := range []string{`{"ask":"start","module":"x"}`, `{"ask":"retry"}`, `{"ask":""}`, `not json`, `{}`} {
+		c, err := dial(path)
+		if err != nil {
+			t.Fatal(err)
+		}
+		c.Write([]byte(ask + "\n"))
+		line, err := bufio.NewReader(c).ReadString('\n')
+		c.Close()
+		if err != nil || !strings.Contains(line, `"error"`) || strings.Contains(line, `"rows"`) {
+			t.Errorf("ask %q answered %q, %v; want a refusal", ask, line, err)
+		}
+	}
+	after, _ := json.Marshal(m.Snapshot())
+	if string(before) != string(after) {
+		t.Fatal("a refused ask changed the rows")
+	}
+}
+
+func TestTableShowsEveryRow(t *testing.T) { // 1134/AC6
+	_, m := served(t)
+	var out strings.Builder
+	s := m.Snapshot()
+	Table(&out, s)
+	for _, r := range s.Rows {
+		if !strings.Contains(out.String(), r.Module) {
+			t.Fatalf("table lacks %s:\n%s", r.Module, out.String())
+		}
+	}
+	for _, w := range []string{"NODE", "MODULE", "ASSIGNED", "HAVE", "PHASE", "HEALTH", "SINCE", "WAITS", "no have yet", "left behind"} {
+		if !strings.Contains(out.String(), w) {
+			t.Errorf("table lacks %q:\n%s", w, out.String())
+		}
+	}
+}
+
+func TestWebIsTheSameRowsAndReadOnly(t *testing.T) { // 1134/AC6
+	_, m := served(t)
+	h := Web(m)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil))
+	if rec.Code != 200 || !strings.Contains(rec.Body.String(), "no have yet") || !strings.Contains(rec.Header().Get("Content-Type"), "text/html") {
+		t.Fatalf("web = %d %q", rec.Code, rec.Body.String())
+	}
+	rec = httptest.NewRecorder()
+	h.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/", strings.NewReader("x")))
+	if rec.Code != http.StatusMethodNotAllowed {
+		t.Fatalf("POST = %d, want 405", rec.Code)
+	}
+}