feat(master): htmx UI with live service overview and agent management
Vendored static assets (sha256): e209dda5c8235479f3166defc7750e1dbcd5a5c1808b7792fc2e6733768fb447 htmx.min.js (htmx.org@2.0.4) 83eca6fa0611fe2b0bf1700b424b88b5eced38ef448ef9760a2ea08fbc875611 htmx-ext-sse.js (htmx-ext-sse@2.2.2)
This commit is contained in:
@@ -0,0 +1,284 @@
|
||||
// Package web serves the master's htmx UI.
|
||||
package web
|
||||
|
||||
import (
|
||||
"context"
|
||||
"embed"
|
||||
"errors"
|
||||
"fmt"
|
||||
"html/template"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/agenttoken"
|
||||
"github.com/shcizo/package-updater/internal/master/events"
|
||||
"github.com/shcizo/package-updater/internal/master/model"
|
||||
"github.com/shcizo/package-updater/internal/master/webauth"
|
||||
)
|
||||
|
||||
//go:embed templates/*.html static/*
|
||||
var assets embed.FS
|
||||
|
||||
var agentNamePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,62}$`)
|
||||
|
||||
// Store is the storage the UI reads and writes. *store.Store satisfies it.
|
||||
type Store interface {
|
||||
ListServices(ctx context.Context) ([]model.Service, error)
|
||||
ListAgents(ctx context.Context) ([]model.Agent, error)
|
||||
CreateAgent(ctx context.Context, name, tokenHash string) (model.Agent, error)
|
||||
RevokeAgent(ctx context.Context, id int64, at time.Time) error
|
||||
}
|
||||
|
||||
// Hub reports and controls live agent connections. *hub.Hub satisfies it.
|
||||
type Hub interface {
|
||||
Online(agentID int64) bool
|
||||
Disconnect(agentID int64)
|
||||
}
|
||||
|
||||
// Subscriber delivers change signals. *events.Broker satisfies it.
|
||||
type Subscriber interface {
|
||||
Subscribe() (<-chan events.Topic, func())
|
||||
}
|
||||
|
||||
// Config holds values shown in the "add agent" snippet.
|
||||
type Config struct {
|
||||
AgentURL string // what agents dial, e.g. wss://10.0.0.5:8090/agent
|
||||
MasterKey string // the master's fingerprint
|
||||
AgentImage string
|
||||
}
|
||||
|
||||
// Server renders the UI.
|
||||
type Server struct {
|
||||
store Store
|
||||
hub Hub
|
||||
sub Subscriber
|
||||
cfg Config
|
||||
log *slog.Logger
|
||||
now func() time.Time
|
||||
pages map[string]*template.Template
|
||||
}
|
||||
|
||||
// New parses the templates. Each page is its own template set so every page
|
||||
// can define its own "content" block.
|
||||
func New(store Store, hub Hub, sub Subscriber, cfg Config, log *slog.Logger) (*Server, error) {
|
||||
funcs := template.FuncMap{"shortRef": shortRef}
|
||||
pages := map[string]*template.Template{}
|
||||
for _, page := range []string{"overview.html", "agents.html", "agent_created.html"} {
|
||||
t, err := template.New(page).Funcs(funcs).ParseFS(assets, "templates/layout.html", "templates/"+page)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse %s: %w", page, err)
|
||||
}
|
||||
pages[page] = t
|
||||
}
|
||||
return &Server{store: store, hub: hub, sub: sub, cfg: cfg, log: log, now: time.Now, pages: pages}, nil
|
||||
}
|
||||
|
||||
// Register adds the UI routes to mux. protect wraps every route except
|
||||
// static assets, normally with webauth's Require.
|
||||
func (s *Server) Register(mux *http.ServeMux, protect func(http.Handler) http.Handler) {
|
||||
mux.Handle("GET /static/", http.FileServerFS(assets))
|
||||
route := func(pattern string, h http.HandlerFunc) { mux.Handle(pattern, protect(h)) }
|
||||
route("GET /{$}", s.overview)
|
||||
route("GET /fragments/services", s.servicesFragment)
|
||||
route("GET /agents", s.agents)
|
||||
route("GET /fragments/agents", s.agentsFragment)
|
||||
route("POST /agents", s.createAgent)
|
||||
route("POST /agents/{id}/revoke", s.revokeAgent)
|
||||
route("GET /events", s.events)
|
||||
}
|
||||
|
||||
type pageData struct {
|
||||
Title string
|
||||
CSRF string
|
||||
Groups []serviceGroup
|
||||
Agents []agentRow
|
||||
Error string
|
||||
Created model.Agent
|
||||
Snippet string
|
||||
}
|
||||
|
||||
func (s *Server) base(r *http.Request, title string) pageData {
|
||||
sess, _ := webauth.SessionFrom(r.Context())
|
||||
return pageData{Title: title, CSRF: sess.CSRFToken}
|
||||
}
|
||||
|
||||
func (s *Server) render(w http.ResponseWriter, status int, page, block string, data pageData) {
|
||||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||||
w.WriteHeader(status)
|
||||
if err := s.pages[page].ExecuteTemplate(w, block, data); err != nil {
|
||||
s.log.Error("render_failed", "page", page, "block", block, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) serviceData(r *http.Request) (pageData, error) {
|
||||
data := s.base(r, "Services")
|
||||
services, err := s.store.ListServices(r.Context())
|
||||
if err != nil {
|
||||
return data, err
|
||||
}
|
||||
data.Groups = groupServices(services)
|
||||
return data, nil
|
||||
}
|
||||
|
||||
func (s *Server) overview(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := s.serviceData(r)
|
||||
if err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
s.render(w, http.StatusOK, "overview.html", "layout", data)
|
||||
}
|
||||
|
||||
func (s *Server) servicesFragment(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := s.serviceData(r)
|
||||
if err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
s.render(w, http.StatusOK, "overview.html", "services", data)
|
||||
}
|
||||
|
||||
func (s *Server) agentData(r *http.Request) (pageData, error) {
|
||||
data := s.base(r, "Agents")
|
||||
agents, err := s.store.ListAgents(r.Context())
|
||||
if err != nil {
|
||||
return data, err
|
||||
}
|
||||
for _, a := range agents {
|
||||
data.Agents = append(data.Agents, agentRow{Agent: a, Online: s.hub.Online(a.ID)})
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
|
||||
func (s *Server) agents(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := s.agentData(r)
|
||||
if err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
s.render(w, http.StatusOK, "agents.html", "layout", data)
|
||||
}
|
||||
|
||||
func (s *Server) agentsFragment(w http.ResponseWriter, r *http.Request) {
|
||||
data, err := s.agentData(r)
|
||||
if err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
s.render(w, http.StatusOK, "agents.html", "agents", data)
|
||||
}
|
||||
|
||||
func (s *Server) createAgent(w http.ResponseWriter, r *http.Request) {
|
||||
name := strings.TrimSpace(r.FormValue("name"))
|
||||
if !agentNamePattern.MatchString(name) {
|
||||
s.agentsWithError(w, r, http.StatusBadRequest, "Agent names use lowercase letters, digits and dashes.")
|
||||
return
|
||||
}
|
||||
token, err := agenttoken.Generate()
|
||||
if err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
agent, err := s.store.CreateAgent(r.Context(), name, agenttoken.Hash(token))
|
||||
if errors.Is(err, model.ErrConflict) {
|
||||
s.agentsWithError(w, r, http.StatusConflict, fmt.Sprintf("An agent named %q already exists.", name))
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
data := s.base(r, "Agent created")
|
||||
data.Created = agent
|
||||
data.Snippet = agentSnippet(s.cfg, token)
|
||||
s.render(w, http.StatusOK, "agent_created.html", "layout", data)
|
||||
}
|
||||
|
||||
func (s *Server) agentsWithError(w http.ResponseWriter, r *http.Request, status int, msg string) {
|
||||
data, err := s.agentData(r)
|
||||
if err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
data.Error = msg
|
||||
s.render(w, status, "agents.html", "layout", data)
|
||||
}
|
||||
|
||||
func (s *Server) revokeAgent(w http.ResponseWriter, r *http.Request) {
|
||||
id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
|
||||
if err != nil {
|
||||
http.Error(w, "invalid agent id", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if err := s.store.RevokeAgent(r.Context(), id, s.now()); err != nil {
|
||||
s.fail(w, err)
|
||||
return
|
||||
}
|
||||
s.hub.Disconnect(id)
|
||||
http.Redirect(w, r, "/agents", http.StatusSeeOther)
|
||||
}
|
||||
|
||||
// events streams change signals as Server-Sent Events. htmx's SSE extension
|
||||
// turns "event: services" into a refetch of the services fragment. The event
|
||||
// carries no state, so a dropped signal is recovered by the next one.
|
||||
func (s *Server) events(w http.ResponseWriter, r *http.Request) {
|
||||
rc := http.NewResponseController(w)
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.Header().Set("Cache-Control", "no-cache")
|
||||
w.Header().Set("X-Accel-Buffering", "no") // stop nginx from buffering the stream
|
||||
|
||||
ch, cancel := s.sub.Subscribe()
|
||||
defer cancel()
|
||||
|
||||
if _, err := fmt.Fprint(w, ": connected\n\n"); err != nil {
|
||||
return
|
||||
}
|
||||
if err := rc.Flush(); err != nil {
|
||||
s.log.Error("sse_flush_unsupported", "error", err)
|
||||
return
|
||||
}
|
||||
keepalive := time.NewTicker(25 * time.Second)
|
||||
defer keepalive.Stop()
|
||||
for {
|
||||
var msg string
|
||||
select {
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
case t := <-ch:
|
||||
msg = fmt.Sprintf("event: %s\ndata: changed\n\n", t)
|
||||
case <-keepalive.C:
|
||||
msg = ": keepalive\n\n"
|
||||
}
|
||||
if _, err := fmt.Fprint(w, msg); err != nil {
|
||||
return
|
||||
}
|
||||
if err := rc.Flush(); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) fail(w http.ResponseWriter, err error) {
|
||||
s.log.Error("request_failed", "error", err)
|
||||
http.Error(w, "internal error", http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
// agentSnippet is the compose file shown once when an agent is added.
|
||||
func agentSnippet(cfg Config, token string) string {
|
||||
return fmt.Sprintf(`services:
|
||||
updater-agent:
|
||||
image: %s
|
||||
entrypoint: ["/usr/local/bin/updater-agent"]
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
MASTER_URL: %s
|
||||
MASTER_KEY: %s
|
||||
AGENT_TOKEN: %s
|
||||
volumes:
|
||||
- /var/run/docker.sock:/var/run/docker.sock
|
||||
`, cfg.AgentImage, cfg.AgentURL, cfg.MasterKey, token)
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/agenttoken"
|
||||
"github.com/shcizo/package-updater/internal/master/events"
|
||||
"github.com/shcizo/package-updater/internal/master/model"
|
||||
"github.com/shcizo/package-updater/internal/protocol"
|
||||
)
|
||||
|
||||
// fakeStore is written from the HTTP handler goroutine and read by the test
|
||||
// goroutine, so every access goes through mu.
|
||||
type fakeStore struct {
|
||||
mu sync.Mutex
|
||||
services []model.Service
|
||||
agents []model.Agent
|
||||
created map[string]string // name → token hash
|
||||
revokedID int64
|
||||
}
|
||||
|
||||
func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) { return f.services, nil }
|
||||
func (f *fakeStore) ListAgents(context.Context) ([]model.Agent, error) { return f.agents, nil }
|
||||
|
||||
func (f *fakeStore) CreateAgent(_ context.Context, name, hash string) (model.Agent, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if _, taken := f.created[name]; taken {
|
||||
return model.Agent{}, model.ErrConflict
|
||||
}
|
||||
f.created[name] = hash
|
||||
return model.Agent{ID: 9, Name: name}, nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) RevokeAgent(_ context.Context, id int64, _ time.Time) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.revokedID = id
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeStore) createdHash(name string) string {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.created[name]
|
||||
}
|
||||
|
||||
func (f *fakeStore) revoked() int64 {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.revokedID
|
||||
}
|
||||
|
||||
type fakeHub struct {
|
||||
mu sync.Mutex
|
||||
online map[int64]bool
|
||||
disconnected int64
|
||||
}
|
||||
|
||||
func (h *fakeHub) Online(id int64) bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return h.online[id]
|
||||
}
|
||||
|
||||
func (h *fakeHub) Disconnect(id int64) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.disconnected = id
|
||||
}
|
||||
|
||||
func (h *fakeHub) lastDisconnected() int64 {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return h.disconnected
|
||||
}
|
||||
|
||||
func newTestServer(t *testing.T) (*httptest.Server, *fakeStore, *fakeHub, *events.Broker) {
|
||||
t.Helper()
|
||||
agentID := int64(2)
|
||||
store := &fakeStore{
|
||||
created: map[string]string{"node-2": "x"},
|
||||
services: []model.Service{
|
||||
{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", AgentID: &agentID, AgentName: "node-2",
|
||||
ActualRef: "gitea.shcizo.se/shcizo/slusk:1.4.2", Status: "running", UpdateMode: "manual"},
|
||||
{Kind: protocol.KindSwarm, SwarmService: "jellyfin_jellyfin",
|
||||
ActualRef: "lscr.io/linuxserver/jellyfin@sha256:0f42497a69fa0441bfd5f9d6bba8694f2a656ec984e571d0ac04c6dd91250039", Status: "degraded", UpdateMode: "manual"},
|
||||
{Kind: protocol.KindContainer, ContainerName: "updater-agent", AgentID: &agentID, AgentName: "node-2",
|
||||
ActualRef: "pu:1", Status: "running", UpdateMode: "manual", Self: true},
|
||||
},
|
||||
agents: []model.Agent{
|
||||
{ID: 2, Name: "node-2", Hostname: "node-2"},
|
||||
{ID: 3, Name: "node-3"},
|
||||
{ID: 4, Name: "old", RevokedAt: ptr(time.Now())},
|
||||
},
|
||||
}
|
||||
hub := &fakeHub{online: map[int64]bool{2: true}}
|
||||
broker := events.NewBroker()
|
||||
s, err := New(store, hub, broker, Config{
|
||||
AgentURL: "wss://10.0.0.5:8090/agent", MasterKey: "sha256:abc", AgentImage: "gitea.shcizo.se/shcizo/package-updater:latest",
|
||||
}, slog.New(slog.NewTextHandler(io.Discard, nil)))
|
||||
require.NoError(t, err)
|
||||
mux := http.NewServeMux()
|
||||
s.Register(mux, func(h http.Handler) http.Handler { return h })
|
||||
srv := httptest.NewServer(mux)
|
||||
t.Cleanup(srv.Close)
|
||||
return srv, store, hub, broker
|
||||
}
|
||||
|
||||
func ptr[T any](v T) *T { return &v }
|
||||
|
||||
func get(t *testing.T, srv *httptest.Server, path string) (int, string) {
|
||||
t.Helper()
|
||||
resp, err := http.Get(srv.URL + path)
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
require.NoError(t, err)
|
||||
return resp.StatusCode, string(body)
|
||||
}
|
||||
|
||||
func noRedirectClient() *http.Client {
|
||||
return &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
|
||||
}
|
||||
|
||||
func TestOverview_GroupsServicesPerMachineAndSwarm(t *testing.T) {
|
||||
srv, _, _, _ := newTestServer(t)
|
||||
code, body := get(t, srv, "/")
|
||||
assert.Equal(t, http.StatusOK, code)
|
||||
assert.Contains(t, body, "<html")
|
||||
assert.Contains(t, body, "node-2")
|
||||
assert.Contains(t, body, "Swarm")
|
||||
assert.Contains(t, body, "slusk/web")
|
||||
assert.Contains(t, body, "jellyfin_jellyfin")
|
||||
assert.Contains(t, body, "sha256:0f42497a69fa…", "digests are shortened")
|
||||
assert.Contains(t, body, "managed outside")
|
||||
assert.Less(t, strings.Index(body, "node-2"), strings.Index(body, "Swarm"), "machines before Swarm")
|
||||
}
|
||||
|
||||
func TestServicesFragment_IsNotAFullPage(t *testing.T) {
|
||||
srv, _, _, _ := newTestServer(t)
|
||||
code, body := get(t, srv, "/fragments/services")
|
||||
assert.Equal(t, http.StatusOK, code)
|
||||
assert.NotContains(t, body, "<html")
|
||||
assert.Contains(t, body, "slusk/web")
|
||||
}
|
||||
|
||||
func TestAgentsPage_ShowsConnectionState(t *testing.T) {
|
||||
srv, _, _, _ := newTestServer(t)
|
||||
_, body := get(t, srv, "/agents")
|
||||
assert.Regexp(t, regexp.MustCompile(`node-2[\s\S]*online`), body)
|
||||
assert.Regexp(t, regexp.MustCompile(`node-3[\s\S]*offline`), body)
|
||||
assert.Regexp(t, regexp.MustCompile(`old[\s\S]*revoked`), body)
|
||||
}
|
||||
|
||||
func TestCreateAgent_ShowsSnippetOnceAndStoresHash(t *testing.T) {
|
||||
srv, store, _, _ := newTestServer(t)
|
||||
resp, err := http.PostForm(srv.URL+"/agents", url.Values{"name": {"node-4"}})
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
|
||||
require.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
token := regexp.MustCompile(`pua_[A-Za-z0-9_-]{43}`).FindString(string(body))
|
||||
require.NotEmpty(t, token)
|
||||
assert.Equal(t, agenttoken.Hash(token), store.createdHash("node-4"), "only the hash is stored")
|
||||
assert.Contains(t, string(body), "MASTER_KEY: sha256:abc")
|
||||
assert.Contains(t, string(body), "MASTER_URL: wss://10.0.0.5:8090/agent")
|
||||
assert.Contains(t, string(body), "shown only once")
|
||||
}
|
||||
|
||||
func TestCreateAgent_Validation(t *testing.T) {
|
||||
srv, _, _, _ := newTestServer(t)
|
||||
cases := []struct {
|
||||
name, input string
|
||||
wantCode int
|
||||
wantText string
|
||||
}{
|
||||
{"duplicate", "node-2", http.StatusConflict, "already exists"},
|
||||
{"invalid characters", "Node 2!", http.StatusBadRequest, "lowercase letters, digits and dashes"},
|
||||
{"empty", " ", http.StatusBadRequest, "lowercase letters, digits and dashes"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
resp, err := http.PostForm(srv.URL+"/agents", url.Values{"name": {c.input}})
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
assert.Equal(t, c.wantCode, resp.StatusCode)
|
||||
assert.Contains(t, string(body), c.wantText)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRevokeAgent_RevokesAndDisconnects(t *testing.T) {
|
||||
srv, store, hub, _ := newTestServer(t)
|
||||
resp, err := noRedirectClient().Post(srv.URL+"/agents/2/revoke", "application/x-www-form-urlencoded", nil)
|
||||
require.NoError(t, err)
|
||||
_ = resp.Body.Close()
|
||||
|
||||
assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
|
||||
assert.Equal(t, "/agents", resp.Header.Get("Location"))
|
||||
assert.Equal(t, int64(2), store.revoked())
|
||||
assert.Equal(t, int64(2), hub.lastDisconnected(), "a revoked agent must be cut off immediately")
|
||||
}
|
||||
|
||||
func TestRevokeAgent_BadID(t *testing.T) {
|
||||
srv, _, _, _ := newTestServer(t)
|
||||
resp, err := noRedirectClient().Post(srv.URL+"/agents/abc/revoke", "application/x-www-form-urlencoded", nil)
|
||||
require.NoError(t, err)
|
||||
_ = resp.Body.Close()
|
||||
assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
|
||||
}
|
||||
|
||||
func TestEvents_StreamsTopics(t *testing.T) {
|
||||
srv, _, _, broker := newTestServer(t)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/events", nil)
|
||||
require.NoError(t, err)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
require.NoError(t, err)
|
||||
defer resp.Body.Close()
|
||||
|
||||
assert.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
|
||||
assert.Equal(t, "no", resp.Header.Get("X-Accel-Buffering"), "nginx must not buffer the stream")
|
||||
|
||||
reader := bufio.NewReader(resp.Body)
|
||||
line, err := reader.ReadString('\n') // ": connected"
|
||||
require.NoError(t, err)
|
||||
require.True(t, strings.HasPrefix(line, ":"))
|
||||
|
||||
broker.Publish(events.Services)
|
||||
for {
|
||||
line, err = reader.ReadString('\n')
|
||||
require.NoError(t, err)
|
||||
if strings.HasPrefix(line, "event:") {
|
||||
break
|
||||
}
|
||||
}
|
||||
assert.Equal(t, "event: services\n", line)
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
:root { --bg: #111418; --fg: #e6e8eb; --muted: #8a929c; --line: #262b31; --ok: #3fb950; --warn: #d29922; --bad: #f85149; --accent: #58a6ff; }
|
||||
@media (prefers-color-scheme: light) {
|
||||
:root { --bg: #ffffff; --fg: #1f2328; --muted: #59636e; --line: #d1d9e0; --ok: #1a7f37; --warn: #9a6700; --bad: #cf222e; --accent: #0969da; }
|
||||
}
|
||||
* { box-sizing: border-box; }
|
||||
body { margin: 0; background: var(--bg); color: var(--fg); font: 14px/1.5 system-ui, sans-serif; }
|
||||
nav { display: flex; gap: 16px; align-items: center; padding: 12px 16px; border-bottom: 1px solid var(--line); }
|
||||
nav form { margin-left: auto; }
|
||||
main { padding: 16px; max-width: 1200px; margin: 0 auto; }
|
||||
a { color: var(--accent); }
|
||||
table { width: 100%; border-collapse: collapse; margin-bottom: 24px; }
|
||||
th, td { text-align: left; padding: 6px 8px; border-bottom: 1px solid var(--line); }
|
||||
th { color: var(--muted); font-weight: 500; }
|
||||
code, pre { font: 13px ui-monospace, monospace; }
|
||||
pre { padding: 12px; border: 1px solid var(--line); overflow-x: auto; }
|
||||
.badge { font-size: 12px; color: var(--muted); border: 1px solid var(--line); padding: 0 6px; border-radius: 8px; }
|
||||
.status-running { color: var(--ok); }
|
||||
.status-degraded, .status-unknown { color: var(--warn); }
|
||||
.status-stopped { color: var(--bad); }
|
||||
.error { color: var(--bad); }
|
||||
.warning { color: var(--warn); }
|
||||
form.inline { display: flex; gap: 8px; margin-bottom: 16px; }
|
||||
@media (max-width: 600px) { table { display: block; overflow-x: auto; } }
|
||||
@@ -0,0 +1,290 @@
|
||||
/*
|
||||
Server Sent Events Extension
|
||||
============================
|
||||
This extension adds support for Server Sent Events to htmx. See /www/extensions/sse.md for usage instructions.
|
||||
|
||||
*/
|
||||
|
||||
(function() {
|
||||
/** @type {import("../htmx").HtmxInternalApi} */
|
||||
var api
|
||||
|
||||
htmx.defineExtension('sse', {
|
||||
|
||||
/**
|
||||
* Init saves the provided reference to the internal HTMX API.
|
||||
*
|
||||
* @param {import("../htmx").HtmxInternalApi} api
|
||||
* @returns void
|
||||
*/
|
||||
init: function(apiRef) {
|
||||
// store a reference to the internal API.
|
||||
api = apiRef
|
||||
|
||||
// set a function in the public API for creating new EventSource objects
|
||||
if (htmx.createEventSource == undefined) {
|
||||
htmx.createEventSource = createEventSource
|
||||
}
|
||||
},
|
||||
|
||||
getSelectors: function() {
|
||||
return ['[sse-connect]', '[data-sse-connect]', '[sse-swap]', '[data-sse-swap]']
|
||||
},
|
||||
|
||||
/**
|
||||
* onEvent handles all events passed to this extension.
|
||||
*
|
||||
* @param {string} name
|
||||
* @param {Event} evt
|
||||
* @returns void
|
||||
*/
|
||||
onEvent: function(name, evt) {
|
||||
var parent = evt.target || evt.detail.elt
|
||||
switch (name) {
|
||||
case 'htmx:beforeCleanupElement':
|
||||
var internalData = api.getInternalData(parent)
|
||||
// Try to remove remove an EventSource when elements are removed
|
||||
var source = internalData.sseEventSource
|
||||
if (source) {
|
||||
api.triggerEvent(parent, 'htmx:sseClose', {
|
||||
source,
|
||||
type: 'nodeReplaced',
|
||||
})
|
||||
internalData.sseEventSource.close()
|
||||
}
|
||||
|
||||
return
|
||||
|
||||
// Try to create EventSources when elements are processed
|
||||
case 'htmx:afterProcessNode':
|
||||
ensureEventSourceOnElement(parent)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
/// ////////////////////////////////////////////
|
||||
// HELPER FUNCTIONS
|
||||
/// ////////////////////////////////////////////
|
||||
|
||||
/**
|
||||
* createEventSource is the default method for creating new EventSource objects.
|
||||
* it is hoisted into htmx.config.createEventSource to be overridden by the user, if needed.
|
||||
*
|
||||
* @param {string} url
|
||||
* @returns EventSource
|
||||
*/
|
||||
function createEventSource(url) {
|
||||
return new EventSource(url, { withCredentials: true })
|
||||
}
|
||||
|
||||
/**
|
||||
* registerSSE looks for attributes that can contain sse events, right
|
||||
* now hx-trigger and sse-swap and adds listeners based on these attributes too
|
||||
* the closest event source
|
||||
*
|
||||
* @param {HTMLElement} elt
|
||||
*/
|
||||
function registerSSE(elt) {
|
||||
// Add message handlers for every `sse-swap` attribute
|
||||
if (api.getAttributeValue(elt, 'sse-swap')) {
|
||||
// Find closest existing event source
|
||||
var sourceElement = api.getClosestMatch(elt, hasEventSource)
|
||||
if (sourceElement == null) {
|
||||
// api.triggerErrorEvent(elt, "htmx:noSSESourceError")
|
||||
return null // no eventsource in parentage, orphaned element
|
||||
}
|
||||
|
||||
// Set internalData and source
|
||||
var internalData = api.getInternalData(sourceElement)
|
||||
var source = internalData.sseEventSource
|
||||
|
||||
var sseSwapAttr = api.getAttributeValue(elt, 'sse-swap')
|
||||
var sseEventNames = sseSwapAttr.split(',')
|
||||
|
||||
for (var i = 0; i < sseEventNames.length; i++) {
|
||||
const sseEventName = sseEventNames[i].trim()
|
||||
const listener = function(event) {
|
||||
// If the source is missing then close SSE
|
||||
if (maybeCloseSSESource(sourceElement)) {
|
||||
return
|
||||
}
|
||||
|
||||
// If the body no longer contains the element, remove the listener
|
||||
if (!api.bodyContains(elt)) {
|
||||
source.removeEventListener(sseEventName, listener)
|
||||
return
|
||||
}
|
||||
|
||||
// swap the response into the DOM and trigger a notification
|
||||
if (!api.triggerEvent(elt, 'htmx:sseBeforeMessage', event)) {
|
||||
return
|
||||
}
|
||||
swap(elt, event.data)
|
||||
api.triggerEvent(elt, 'htmx:sseMessage', event)
|
||||
}
|
||||
|
||||
// Register the new listener
|
||||
api.getInternalData(elt).sseEventListener = listener
|
||||
source.addEventListener(sseEventName, listener)
|
||||
}
|
||||
}
|
||||
|
||||
// Add message handlers for every `hx-trigger="sse:*"` attribute
|
||||
if (api.getAttributeValue(elt, 'hx-trigger')) {
|
||||
// Find closest existing event source
|
||||
var sourceElement = api.getClosestMatch(elt, hasEventSource)
|
||||
if (sourceElement == null) {
|
||||
// api.triggerErrorEvent(elt, "htmx:noSSESourceError")
|
||||
return null // no eventsource in parentage, orphaned element
|
||||
}
|
||||
|
||||
// Set internalData and source
|
||||
var internalData = api.getInternalData(sourceElement)
|
||||
var source = internalData.sseEventSource
|
||||
|
||||
var triggerSpecs = api.getTriggerSpecs(elt)
|
||||
triggerSpecs.forEach(function(ts) {
|
||||
if (ts.trigger.slice(0, 4) !== 'sse:') {
|
||||
return
|
||||
}
|
||||
|
||||
var listener = function (event) {
|
||||
if (maybeCloseSSESource(sourceElement)) {
|
||||
return
|
||||
}
|
||||
if (!api.bodyContains(elt)) {
|
||||
source.removeEventListener(ts.trigger.slice(4), listener)
|
||||
}
|
||||
// Trigger events to be handled by the rest of htmx
|
||||
htmx.trigger(elt, ts.trigger, event)
|
||||
htmx.trigger(elt, 'htmx:sseMessage', event)
|
||||
}
|
||||
|
||||
// Register the new listener
|
||||
api.getInternalData(elt).sseEventListener = listener
|
||||
source.addEventListener(ts.trigger.slice(4), listener)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* ensureEventSourceOnElement creates a new EventSource connection on the provided element.
|
||||
* If a usable EventSource already exists, then it is returned. If not, then a new EventSource
|
||||
* is created and stored in the element's internalData.
|
||||
* @param {HTMLElement} elt
|
||||
* @param {number} retryCount
|
||||
* @returns {EventSource | null}
|
||||
*/
|
||||
function ensureEventSourceOnElement(elt, retryCount) {
|
||||
if (elt == null) {
|
||||
return null
|
||||
}
|
||||
|
||||
// handle extension source creation attribute
|
||||
if (api.getAttributeValue(elt, 'sse-connect')) {
|
||||
var sseURL = api.getAttributeValue(elt, 'sse-connect')
|
||||
if (sseURL == null) {
|
||||
return
|
||||
}
|
||||
|
||||
ensureEventSource(elt, sseURL, retryCount)
|
||||
}
|
||||
|
||||
registerSSE(elt)
|
||||
}
|
||||
|
||||
function ensureEventSource(elt, url, retryCount) {
|
||||
var source = htmx.createEventSource(url)
|
||||
|
||||
source.onerror = function(err) {
|
||||
// Log an error event
|
||||
api.triggerErrorEvent(elt, 'htmx:sseError', { error: err, source })
|
||||
|
||||
// If parent no longer exists in the document, then clean up this EventSource
|
||||
if (maybeCloseSSESource(elt)) {
|
||||
return
|
||||
}
|
||||
|
||||
// Otherwise, try to reconnect the EventSource
|
||||
if (source.readyState === EventSource.CLOSED) {
|
||||
retryCount = retryCount || 0
|
||||
retryCount = Math.max(Math.min(retryCount * 2, 128), 1)
|
||||
var timeout = retryCount * 500
|
||||
window.setTimeout(function() {
|
||||
ensureEventSourceOnElement(elt, retryCount)
|
||||
}, timeout)
|
||||
}
|
||||
}
|
||||
|
||||
source.onopen = function(evt) {
|
||||
api.triggerEvent(elt, 'htmx:sseOpen', { source })
|
||||
|
||||
if (retryCount && retryCount > 0) {
|
||||
const childrenToFix = elt.querySelectorAll("[sse-swap], [data-sse-swap], [hx-trigger], [data-hx-trigger]")
|
||||
for (let i = 0; i < childrenToFix.length; i++) {
|
||||
registerSSE(childrenToFix[i])
|
||||
}
|
||||
// We want to increase the reconnection delay for consecutive failed attempts only
|
||||
retryCount = 0
|
||||
}
|
||||
}
|
||||
|
||||
api.getInternalData(elt).sseEventSource = source
|
||||
|
||||
|
||||
var closeAttribute = api.getAttributeValue(elt, "sse-close");
|
||||
if (closeAttribute) {
|
||||
// close eventsource when this message is received
|
||||
source.addEventListener(closeAttribute, function() {
|
||||
api.triggerEvent(elt, 'htmx:sseClose', {
|
||||
source,
|
||||
type: 'message',
|
||||
})
|
||||
source.close()
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* maybeCloseSSESource confirms that the parent element still exists.
|
||||
* If not, then any associated SSE source is closed and the function returns true.
|
||||
*
|
||||
* @param {HTMLElement} elt
|
||||
* @returns boolean
|
||||
*/
|
||||
function maybeCloseSSESource(elt) {
|
||||
if (!api.bodyContains(elt)) {
|
||||
var source = api.getInternalData(elt).sseEventSource
|
||||
if (source != undefined) {
|
||||
api.triggerEvent(elt, 'htmx:sseClose', {
|
||||
source,
|
||||
type: 'nodeMissing',
|
||||
})
|
||||
source.close()
|
||||
// source = null
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @param {HTMLElement} elt
|
||||
* @param {string} content
|
||||
*/
|
||||
function swap(elt, content) {
|
||||
api.withExtensions(elt, function(extension) {
|
||||
content = extension.transformResponse(content, null, elt)
|
||||
})
|
||||
|
||||
var swapSpec = api.getSwapSpecification(elt)
|
||||
var target = api.getTarget(elt)
|
||||
api.swap(target, content, swapSpec)
|
||||
}
|
||||
|
||||
|
||||
function hasEventSource(node) {
|
||||
return api.getInternalData(node).sseEventSource != null
|
||||
}
|
||||
})()
|
||||
+1
File diff suppressed because one or more lines are too long
@@ -0,0 +1,6 @@
|
||||
{{define "content"}}
|
||||
<h1>Agent "{{.Created.Name}}" created</h1>
|
||||
<p class="warning">The token is shown only once. Save this as <code>docker-compose.yml</code> on the machine and run <code>docker compose up -d</code>.</p>
|
||||
<pre>{{.Snippet}}</pre>
|
||||
<p><a href="/agents">Back to agents</a></p>
|
||||
{{end}}
|
||||
@@ -0,0 +1,38 @@
|
||||
{{define "content"}}
|
||||
<h1>Agents</h1>
|
||||
<form method="post" action="/agents" class="inline">
|
||||
<input type="hidden" name="csrf_token" value="{{.CSRF}}">
|
||||
<label>Name <input name="name" required placeholder="node-2"></label>
|
||||
<button type="submit">Add agent</button>
|
||||
</form>
|
||||
{{if .Error}}<p class="error">{{.Error}}</p>{{end}}
|
||||
<div id="agents" hx-get="/fragments/agents" hx-trigger="sse:agents" hx-swap="innerHTML">
|
||||
{{template "agents" .}}
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
{{define "agents"}}
|
||||
<table>
|
||||
<thead><tr><th>Name</th><th>Hostname</th><th>State</th><th>Version</th><th>Swarm manager</th><th>Last seen</th><th></th></tr></thead>
|
||||
<tbody>
|
||||
{{range .Agents}}
|
||||
<tr>
|
||||
<td>{{.Name}}</td>
|
||||
<td>{{.Hostname}}</td>
|
||||
<td>{{if .RevokedAt}}<span class="status status-stopped">revoked</span>{{else if .Online}}<span class="status status-running">online</span>{{else}}<span class="status status-degraded">offline</span>{{end}}</td>
|
||||
<td>{{.AgentVersion}}</td>
|
||||
<td>{{if .SwarmManager}}yes{{end}}</td>
|
||||
<td>{{with .LastSeenAt}}{{.Format "2006-01-02 15:04"}}{{end}}</td>
|
||||
<td>{{if not .RevokedAt}}
|
||||
<form method="post" action="/agents/{{.ID}}/revoke" onsubmit="return confirm('Revoke {{.Name}}? It is disconnected immediately and must be added again to come back.')">
|
||||
<input type="hidden" name="csrf_token" value="{{$.CSRF}}">
|
||||
<button type="submit">Revoke</button>
|
||||
</form>
|
||||
{{end}}</td>
|
||||
</tr>
|
||||
{{else}}
|
||||
<tr><td colspan="7">No agents yet.</td></tr>
|
||||
{{end}}
|
||||
</tbody>
|
||||
</table>
|
||||
{{end}}
|
||||
@@ -0,0 +1,22 @@
|
||||
{{define "layout"}}<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>{{.Title}} · package-updater</title>
|
||||
<link rel="stylesheet" href="/static/app.css">
|
||||
<script src="/static/htmx.min.js"></script>
|
||||
<script src="/static/htmx-ext-sse.js"></script>
|
||||
</head>
|
||||
<body hx-headers='{"X-CSRF-Token": "{{.CSRF}}"}' hx-ext="sse" sse-connect="/events">
|
||||
<nav>
|
||||
<a href="/">Services</a>
|
||||
<a href="/agents">Agents</a>
|
||||
<form method="post" action="/auth/logout">
|
||||
<input type="hidden" name="csrf_token" value="{{.CSRF}}">
|
||||
<button type="submit">Log out</button>
|
||||
</form>
|
||||
</nav>
|
||||
<main>{{template "content" .}}</main>
|
||||
</body>
|
||||
</html>{{end}}
|
||||
@@ -0,0 +1,29 @@
|
||||
{{define "content"}}
|
||||
<h1>Services</h1>
|
||||
<div id="services" hx-get="/fragments/services" hx-trigger="sse:services" hx-swap="innerHTML">
|
||||
{{template "services" .}}
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
{{define "services"}}
|
||||
{{range .Groups}}
|
||||
<section>
|
||||
<h2>{{.Title}}</h2>
|
||||
<table>
|
||||
<thead><tr><th>Service</th><th>Image</th><th>Status</th><th>Mode</th></tr></thead>
|
||||
<tbody>
|
||||
{{range .Services}}
|
||||
<tr>
|
||||
<td>{{.DisplayName}}{{if .Self}} <span class="badge">managed outside</span>{{end}}</td>
|
||||
<td><code>{{shortRef .ActualRef}}</code></td>
|
||||
<td class="status status-{{.Status}}">{{.Status}}</td>
|
||||
<td>{{.UpdateMode}}</td>
|
||||
</tr>
|
||||
{{end}}
|
||||
</tbody>
|
||||
</table>
|
||||
</section>
|
||||
{{else}}
|
||||
<p>No services reported yet. <a href="/agents">Add an agent</a> to get started.</p>
|
||||
{{end}}
|
||||
{{end}}
|
||||
@@ -0,0 +1,51 @@
|
||||
package web
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/shcizo/package-updater/internal/imageref"
|
||||
"github.com/shcizo/package-updater/internal/master/model"
|
||||
"github.com/shcizo/package-updater/internal/protocol"
|
||||
)
|
||||
|
||||
type serviceGroup struct {
|
||||
Title string
|
||||
Services []model.Service
|
||||
}
|
||||
|
||||
// groupServices groups machine-bound services per agent and Swarm services
|
||||
// in one group, keeping the store's order (machines first, Swarm last).
|
||||
func groupServices(services []model.Service) []serviceGroup {
|
||||
var groups []serviceGroup
|
||||
index := map[string]int{}
|
||||
for _, s := range services {
|
||||
title := s.AgentName
|
||||
if s.Kind == protocol.KindSwarm {
|
||||
title = "Swarm"
|
||||
} else if title == "" {
|
||||
title = "Unknown agent"
|
||||
}
|
||||
i, ok := index[title]
|
||||
if !ok {
|
||||
i = len(groups)
|
||||
index[title] = i
|
||||
groups = append(groups, serviceGroup{Title: title})
|
||||
}
|
||||
groups[i].Services = append(groups[i].Services, s)
|
||||
}
|
||||
return groups
|
||||
}
|
||||
|
||||
// shortRef shortens a digest in an image reference to 12 hex characters.
|
||||
func shortRef(ref string) string {
|
||||
r := imageref.Parse(ref)
|
||||
if hex, ok := strings.CutPrefix(r.Digest, "sha256:"); ok && len(hex) > 12 {
|
||||
r.Digest = "sha256:" + hex[:12] + "…"
|
||||
}
|
||||
return r.String()
|
||||
}
|
||||
|
||||
type agentRow struct {
|
||||
model.Agent
|
||||
Online bool
|
||||
}
|
||||
Reference in New Issue
Block a user