package api // Previews for the Files tab: GET /api/files/thumb hands back one small JPEG of an image or // a video, made by internal/thumbs and kept in /thumbs.db. The work is bounded // two ways — a handful of ffmpegs at once, and one per file however many rows ask for it — // because a folder of a thousand photos asks for every row the moment previews go on. // // Root mode (?sudo=1) makes the preview through the same root helper the file reads use, // so a file only root can read gets a preview too, and the probe runs as root as well: root // opens every render node, whatever groups the daemon's user has. import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "log" "net/http" "net/http/httptest" "net/url" "os" "os/exec" "path/filepath" "runtime" "strconv" "strings" "sync" "time" "hamlab/internal/media" "hamlab/internal/priv" "hamlab/internal/store" ) const thumbSize = 256 const thumbTimeout = 30 * time.Second const capsTTL = 10 * time.Minute // how long a probe is believed; a refresh after an install or a group change does not wait type thumbSvc struct { cache *media.Cache self string sem chan struct{} mu sync.Mutex caps map[bool]*capsEntry // keyed by root calls map[string]*thumbCall } type capsEntry struct { mu sync.Mutex caps media.Caps err error at time.Time } type thumbCall struct { done chan struct{} res media.Result err error } func newThumbSvc(dataDir, self string) *thumbSvc { n := runtime.NumCPU() / 2 // each ffmpeg is itself threaded; previews must not take the whole machine if n < 2 { n = 2 } if n > 4 { n = 4 } return &thumbSvc{cache: media.NewCache(filepath.Join(dataDir, "thumbs.db")), self: self, sem: make(chan struct{}, n), caps: map[bool]*capsEntry{}, calls: map[string]*thumbCall{}} } // capsFor is what previews can be made with, probed once and kept for a while. func (t *thumbSvc) capsFor(ctx context.Context, root, refresh bool) (media.Caps, error) { t.mu.Lock() e := t.caps[root] if e == nil { e = &capsEntry{} t.caps[root] = e } t.mu.Unlock() e.mu.Lock() defer e.mu.Unlock() if !refresh && !e.at.IsZero() && time.Since(e.at) < capsTTL { return e.caps, e.err } if root { e.caps, e.err = media.ProbeRoot(ctx, t.self) } else { e.caps, e.err = media.Probe(ctx), nil } e.at = time.Now() mode := "as " + map[bool]string{false: "the daemon", true: "root"}[root] switch { case e.err != nil: log.Printf("thumbs: probe %s failed: %v", mode, e.err) case e.caps.Error != "": log.Printf("thumbs: %s: %s", mode, e.caps.Error) default: log.Printf("thumbs: %s: ffmpeg %s, %s", mode, e.caps.Version, describeCaps(e.caps)) } return e.caps, e.err } // describeCaps is the one line that says how previews get made here. func describeCaps(c media.Caps) string { if len(c.GPUs) == 0 { return "CPU only" } parts := make([]string, 0, len(c.GPUs)) for _, g := range c.GPUs { name := g.Name if name == "" { name = g.Device } p := name + " (" + g.API if len(g.Codecs) > 0 { p += ": " + strings.Join(g.Codecs, ", ") } parts = append(parts, p+")") } return "GPU " + strings.Join(parts, ", ") } // make produces a preview, sharing the work when the same file is asked for twice at once. // in is a path, or a URL the hub serves another machine's file at (kind then says what it // is); device is the one to decode on ("" any, "cpu" none). func (t *thumbSvc) make(ctx context.Context, root bool, key, in, kind, device string) (media.Result, error) { t.mu.Lock() if c, ok := t.calls[key]; ok { t.mu.Unlock() select { case <-c.done: return c.res, c.err case <-ctx.Done(): return media.Result{}, ctx.Err() } } c := &thumbCall{done: make(chan struct{})} t.calls[key] = c t.mu.Unlock() defer func() { t.mu.Lock() delete(t.calls, key) t.mu.Unlock() close(c.done) }() // The work itself runs on a context of its own: the first requester going away must not // cancel a preview three other rows are waiting for. wctx, cancel := context.WithTimeout(context.Background(), thumbTimeout+5*time.Second) defer cancel() select { case t.sem <- struct{}{}: defer func() { <-t.sem }() case <-ctx.Done(): c.err = ctx.Err() return c.res, c.err } caps, err := t.capsFor(wctx, root, false) if err != nil { c.err = err return c.res, c.err } o := media.Opts{Size: thumbSize, Timeout: thumbTimeout, Kind: kind} caps = caps.Only(device) if root { c.res, c.err = media.MakeRoot(wctx, t.self, caps, in, o) } else { c.res, c.err = media.Make(wctx, caps, in, o) } return c.res, c.err } // ---- handlers ---- // thumbDeviceHeader carries the device a hub wants a video preview made on: the Transcode // device setting, when it is on the machine the file is on. const thumbDeviceHeader = "X-Hamlab-Device" // filesThumb: GET /api/files/thumb?path=…&v=-[&sudo=1][&cached=1]. v is the // client's idea of the file, from its listing; it makes a cache hit free of a stat, and a // changed file a new URL. cached=1 answers from the cache or 404, and makes nothing: the hub // asks that before having the preview made on another machine. // // A video preview is made on the Transcode device: on the hub, by the setting (another // machine's device reads the file through the hub); on a node, by the hub's header. func (s *Server) filesThumb(w http.ResponseWriter, r *http.Request) { fsys, ok := s.fsFor(w, r) if !ok { return } p, ok := cleanQuery(w, r, "path") if !ok { return } if media.Kind(p) == "" { jsonErr(w, 415, "no preview for this kind of file") return } root := r.URL.Query().Get("sudo") == "1" if mt, sz, ok := parseV(r.URL.Query().Get("v")); ok { key := media.Key(p, sz, mt) if r.Header.Get("If-None-Match") == `"`+key+`"` { w.WriteHeader(304) return } if s.serveCachedThumb(w, key) { return } } in, err := fsys.Stat(r.Context(), p) if err != nil { fsErr(w, err) return } if in.Type == "dir" || in.Type == "other" || in.Broken { jsonErr(w, 400, "not a regular file") return } key := media.Key(p, in.Size, in.Mtime) if s.serveCachedThumb(w, key) { return } if r.URL.Query().Get("cached") == "1" { jsonErr(w, 404, "not made yet") return } device := r.Header.Get(thumbDeviceHeader) if c := s.transcodeChoice(); c != nil && device == "" && media.Kind(p) == "video" { if c.Node != "" { // The hub's own file, previewed on another machine. if res, ok := s.thumbElsewhere(r.Context(), c, "", p, root, key); ok { if res.Failure != "" { _ = s.thumbs.cache.Put(key, media.Entry{Path: p, Failure: res.Failure}) jsonErr(w, 422, res.Failure) return } s.keepThumb(w, key, p, res) return } } else { device = c.Device } } res, err := s.thumbs.make(r.Context(), root, key, p, "", device) if err != nil { var de *media.DecodeError switch { case errors.As(err, &de): if err := s.thumbs.cache.Put(key, media.Entry{Path: p, Failure: de.Msg}); err != nil { log.Printf("thumbs: cache: %v", err) } log.Printf("thumbs: %s: %s", p, de.Msg) jsonErr(w, 422, de.Msg) case errors.Is(err, context.DeadlineExceeded): log.Printf("thumbs: %s: timed out", p) jsonErr(w, 504, "preview timed out") case errors.Is(err, context.Canceled): jsonErr(w, 499, "cancelled") default: jsonErr(w, 503, err.Error()) } return } s.keepThumb(w, key, p, res) } // keepThumb puts a preview in the cache and answers with it. func (s *Server) keepThumb(w http.ResponseWriter, key, p string, res media.Result) { if err := s.thumbs.cache.Put(key, media.Entry{Path: p, Engine: res.Engine, Codec: res.Codec, Width: res.Width, Height: res.Height, JPEG: res.JPEG}); err != nil { log.Printf("thumbs: cache: %v", err) } writeThumb(w, key, res.JPEG, res.Engine) } // ---- previews made on another machine (the Transcode device) ---- // thumbFrom: POST /api/files/thumb-from {url, kind, key, device} — a preview of a file on // another machine, read through the hub at url, made here on device. Answers the JPEG (the // engine in X-Hamlab-Thumb), or 422 with why the file can't be previewed. func (s *Server) thumbFrom(w http.ResponseWriter, r *http.Request) { var body struct{ URL, Kind, Key, Device string } if err := json.NewDecoder(io.LimitReader(r.Body, 8192)).Decode(&body); err != nil || !strings.HasPrefix(body.URL, "http") || body.Key == "" { jsonErr(w, 400, "expected {url, kind, key, device}") return } res, err := s.thumbs.make(r.Context(), false, "from:"+body.Key, body.URL, body.Kind, body.Device) if err != nil { var de *media.DecodeError if errors.As(err, &de) { jsonErr(w, 422, de.Msg) } else { jsonErr(w, 503, err.Error()) } return } w.Header().Set("X-Hamlab-Thumb-Codec", res.Codec) w.Header().Set("X-Hamlab-Thumb-Size", fmt.Sprintf("%dx%d", res.Width, res.Height)) writeThumb(w, body.Key, res.JPEG, res.Engine) } // thumbPut: PUT /api/files/thumb?path=…&v=-[&sudo=1] — a preview another machine // made of this file, kept here as if made here. The body is the JPEG, or nothing with // X-Hamlab-Thumb-Failure saying why the file can't be previewed. Refused if the file has // changed since v. func (s *Server) thumbPut(w http.ResponseWriter, r *http.Request) { fsys, ok := s.fsFor(w, r) if !ok { return } p, ok := cleanQuery(w, r, "path") if !ok { return } mt, sz, ok := parseV(r.URL.Query().Get("v")) if !ok { jsonErr(w, 400, "v=- is required") return } in, err := fsys.Stat(r.Context(), p) if err != nil { fsErr(w, err) return } if in.Size != sz || in.Mtime != mt { jsonErr(w, 409, "the file has changed") return } key := media.Key(p, sz, mt) e := media.Entry{Path: p, Engine: r.Header.Get("X-Hamlab-Thumb"), Codec: r.Header.Get("X-Hamlab-Thumb-Codec"), Failure: r.Header.Get("X-Hamlab-Thumb-Failure")} fmt.Sscanf(r.Header.Get("X-Hamlab-Thumb-Size"), "%dx%d", &e.Width, &e.Height) if e.Failure == "" { if e.JPEG, err = io.ReadAll(io.LimitReader(r.Body, 4<<20)); err != nil || len(e.JPEG) == 0 { jsonErr(w, 400, "no picture") return } } if err := s.thumbs.cache.Put(key, e); err != nil { jsonErr(w, 500, err.Error()) return } w.WriteHeader(204) } // thumbElsewhere has a video preview made on the Transcode device's machine c.Node, of a // file on machine node ("" the hub), and answers it. ok is false when that machine can't // (unreachable, an older Ham Lab): the caller makes it where the file is instead. A file // that can't be previewed is a result too, with the failure in it. func (s *Server) thumbElsewhere(ctx context.Context, c *store.Transcode, node, p string, root bool, key string) (media.Result, bool) { src, err := s.srcURL(c.Node, node, p, root) if err != nil { return media.Result{}, false } body, _ := json.Marshal(map[string]string{"url": src, "kind": "video", "key": key, "device": c.Device}) ctx, cancel := context.WithTimeout(ctx, thumbTimeout+10*time.Second) defer cancel() var code int var b []byte var hdr http.Header if c.Node == "" { rec := httptest.NewRecorder() req := httptest.NewRequest("POST", "/api/files/thumb-from", bytes.NewReader(body)).WithContext(ctx) s.thumbFrom(rec, req) code, b, hdr = rec.Code, rec.Body.Bytes(), rec.Header() } else { code, b, hdr, err = s.nodes.DoHeaders(ctx, c.Node, "POST", "/api/files/thumb-from", body, http.Header{"Content-Type": []string{"application/json"}}) if err != nil { log.Printf("thumbs: %s on %s: %v; made where it is", p, c.Node, err) return media.Result{}, false } } switch code { case 200: res := media.Result{JPEG: b, Engine: c.Node + " " + hdr.Get("X-Hamlab-Thumb"), Codec: hdr.Get("X-Hamlab-Thumb-Codec")} fmt.Sscanf(hdr.Get("X-Hamlab-Thumb-Size"), "%dx%d", &res.Width, &res.Height) if c.Node == "" { res.Engine = s.hostName() + " " + hdr.Get("X-Hamlab-Thumb") } return res, true case 422: var e struct{ Error string } _ = json.Unmarshal(b, &e) return media.Result{Failure: e.Error}, true } log.Printf("thumbs: %s on %s answered %d; made where it is", p, nodeLabel(c.Node), code) return media.Result{}, false } // hubThumb: GET /api/nodes/{name}/files/thumb — a node's preview, through the hub. Without a // Transcode device setting, or for an image or a sound file's cover, it goes to the node as // any request does. A video is made on the setting's device: the node's own (told by header), // or another machine's, which reads the file through the hub; the node keeps the result in // its own cache either way. func (s *Server) hubThumb(w http.ResponseWriter, r *http.Request) { node := r.PathValue("name") proxy, ok := s.nodes.Proxy(node) if !ok { jsonErr(w, 404, "no such machine") return } c := s.transcodeChoice() q := r.URL.Query() p := q.Get("path") if c == nil || media.Kind(p) != "video" { proxy.ServeHTTP(w, r) return } if c.Node == node { r.Header.Set(thumbDeviceHeader, c.Device) proxy.ServeHTTP(w, r) return } mt, sz, okV := parseV(q.Get("v")) if !okV { proxy.ServeHTTP(w, r) return } // The node's own cache first. cq := url.Values{"path": {p}, "v": {q.Get("v")}, "cached": {"1"}} if q.Get("sudo") == "1" { cq.Set("sudo", "1") } hdr := http.Header{} if inm := r.Header.Get("If-None-Match"); inm != "" { hdr.Set("If-None-Match", inm) } code, b, rh, err := s.nodes.DoHeaders(r.Context(), node, "GET", "/api/files/thumb?"+cq.Encode(), nil, hdr) if err == nil && code != 404 { for _, k := range []string{"Content-Type", "Cache-Control", "ETag", "X-Hamlab-Thumb"} { if v := rh.Get(k); v != "" { w.Header().Set(k, v) } } w.WriteHeader(code) _, _ = w.Write(b) return } key := media.Key(p, sz, mt) res, made := s.thumbElsewhere(r.Context(), c, node, p, q.Get("sudo") == "1", key) if !made { proxy.ServeHTTP(w, r) // the node makes it itself return } // Kept where the file is, so it is made once. ph := http.Header{"X-Hamlab-Thumb": {res.Engine}, "X-Hamlab-Thumb-Codec": {res.Codec}, "X-Hamlab-Thumb-Size": {fmt.Sprintf("%dx%d", res.Width, res.Height)}} if res.Failure != "" { ph.Set("X-Hamlab-Thumb-Failure", res.Failure) } pq := url.Values{"path": {p}, "v": {q.Get("v")}} if q.Get("sudo") == "1" { pq.Set("sudo", "1") } if code, _, _, err := s.nodes.DoHeaders(r.Context(), node, "PUT", "/api/files/thumb?"+pq.Encode(), res.JPEG, ph); err != nil || code != 204 { log.Printf("thumbs: keeping %s's preview of %s: %d %v", node, p, code, err) } if res.Failure != "" { jsonErr(w, 422, res.Failure) return } writeThumb(w, key, res.JPEG, res.Engine) } // serveCachedThumb answers from the cache when it can. A cache that cannot be read is a // miss: the preview is made again rather than refused. func (s *Server) serveCachedThumb(w http.ResponseWriter, key string) bool { e, found, err := s.thumbs.cache.Get(key) if err != nil { log.Printf("thumbs: cache: %v", err) } switch { case !found: return false case e.Failure != "": jsonErr(w, 422, e.Failure) default: writeThumb(w, key, e.JPEG, "cache ("+e.Engine+")") } return true } func writeThumb(w http.ResponseWriter, key string, jpeg []byte, engine string) { w.Header().Set("Content-Type", "image/jpeg") w.Header().Set("Content-Length", strconv.Itoa(len(jpeg))) w.Header().Set("Cache-Control", "private, max-age=604800") w.Header().Set("ETag", `"`+key+`"`) w.Header().Set("X-Hamlab-Thumb", engine) w.Header().Set("X-Content-Type-Options", "nosniff") _, _ = w.Write(jpeg) } func parseV(v string) (mtime, size int64, ok bool) { a, b, found := strings.Cut(v, "-") if !found { return 0, 0, false } mtime, err1 := strconv.ParseInt(a, 10, 64) size, err2 := strconv.ParseInt(b, 10, 64) return mtime, size, err1 == nil && err2 == nil } type thumbStatus struct { OK bool `json:"ok"` Engine string `json:"engine"` // gpu | cpu | "" Summary string `json:"summary"` FFmpeg string `json:"ffmpeg,omitempty"` GPUs []media.GPU `json:"gpus"` Unusable []media.GPU `json:"unusable,omitempty"` Error string `json:"error,omitempty"` Install *thumbInstall `json:"install,omitempty"` // ffmpeg is missing and this can put it on GPUFix bool `json:"gpu_fix,omitempty"` // a render node is there but closed to the daemon's user, and this can open it User string `json:"user"` Cache thumbCache `json:"cache"` } type thumbCache struct { Path string `json:"path"` Entries int64 `json:"entries"` Bytes int64 `json:"bytes"` } type thumbInstall struct { PM string `json:"pm"` Package string `json:"package"` } // thumbsStatus: GET /api/files/thumbs[?sudo=1][&refresh=1] — how previews get made on this machine. func (s *Server) thumbsStatus(w http.ResponseWriter, r *http.Request) { root := r.URL.Query().Get("sudo") == "1" if root && s.fsSudo == nil { jsonErr(w, 403, "root access needs passwordless sudo on the host") return } caps, err := s.thumbs.capsFor(r.Context(), root, r.URL.Query().Get("refresh") == "1") st := thumbStatus{GPUs: []media.GPU{}, User: s.user, Cache: thumbCache{Path: s.thumbs.cache.Path}} st.Cache.Entries, st.Cache.Bytes = s.thumbs.cache.Stats() if root { st.User = "root" } if err != nil { st.Error = err.Error() jsonOK(w, 200, st) return } st.FFmpeg, st.GPUs, st.Unusable, st.Error = caps.Version, caps.GPUs, caps.Unusable, caps.Error if st.GPUs == nil { st.GPUs = []media.GPU{} } st.OK = caps.Error == "" switch { case !st.OK: if pkg := ffmpegPackage(s.updates.PM()); pkg != "" && priv.Available() { st.Install = &thumbInstall{PM: s.updates.PM(), Package: pkg} } st.Summary = st.Error case len(caps.GPUs) > 0: st.Engine = "gpu" st.Summary = describeCaps(caps) + " · images on the CPU" default: st.Engine = "cpu" st.Summary = "CPU only" } if !root { for _, g := range caps.Unusable { if g.Permission && gpuFixPossible() { st.GPUFix = true } } } jsonOK(w, 200, st) } // ffmpegPackage is what to install for ffmpeg on a package manager. Fedora ships it as // ffmpeg-free; the others under its own name. func ffmpegPackage(pm string) string { switch pm { case "": return "" case "dnf": if b, _ := os.ReadFile("/etc/os-release"); strings.Contains(string(b), "ID=fedora") { return "ffmpeg-free" } } return "ffmpeg" } const hamlabUnit = "/etc/systemd/system/hamlab.service" // gpuFixPossible: the daemon is an ordinary user under systemd, with root to hand. func gpuFixPossible() bool { if os.Geteuid() == 0 || !priv.Available() { return false } _, err := os.Stat(hamlabUnit) return err == nil } // thumbsInstall: POST /api/files/thumbs/install — put ffmpeg on with the machine's package manager. func (s *Server) thumbsInstall(w http.ResponseWriter, r *http.Request) { pkg := ffmpegPackage(s.updates.PM()) if pkg == "" { jsonErr(w, 400, "no supported package manager on this machine") return } t, err := s.updates.Install([]string{pkg}) if err != nil { jsonErr(w, 400, err.Error()) return } log.Printf("thumbs: installing %s as task %s (from %s)", pkg, t.ID, clientIP(r)) jsonOK(w, 202, t.Snapshot(false)) } // gpuAccessScript gives hamlab.service the groups that own the render nodes, by a drop-in, // then restarts it: a running process cannot gain a group. const gpuAccessScript = `set -e unit=` + hamlabUnit + ` [ -f "$unit" ] || { echo "no $unit: this machine does not run Ham Lab under systemd"; exit 1; } groups="" for g in render video; do if getent group "$g" >/dev/null 2>&1; then groups="$groups $g"; fi done [ -n "$groups" ] || { echo "neither a render nor a video group exists here"; exit 1; } mkdir -p "$unit.d" printf '[Service]\nSupplementaryGroups=%s\n' "${groups# }" > "$unit.d/gpu.conf" echo "hamlab.service: supplementary groups${groups} ($unit.d/gpu.conf)" systemctl daemon-reload echo 'Ham Lab restarts in 3 seconds to pick up its new groups; the page reconnects by itself.' systemd-run --quiet --on-active=3 --unit=hamlab-gpu-restart systemctl restart hamlab.service ` // thumbsGPUAccess: POST /api/files/thumbs/gpu-access — open the render nodes to the daemon. func (s *Server) thumbsGPUAccess(w http.ResponseWriter, r *http.Request) { if !gpuFixPossible() { jsonErr(w, 400, "nothing to do: Ham Lab is root, has no root, or is not under systemd here") return } if t, busy := s.tasks.RunningOf("thumbs:gpu"); busy { // a second run would only collide with the first's restart timer jsonErr(w, 409, "already being done (task "+t.ID+")") return } full := priv.Wrap("sh", "-c", gpuAccessScript) t, err := s.tasks.Start("thumbs:gpu", "give Ham Lab access to the GPU", exec.Command(full[0], full[1:]...)) if err != nil { jsonErr(w, 400, err.Error()) return } log.Printf("thumbs: GPU access drop-in as task %s, restart follows (from %s)", t.ID, clientIP(r)) jsonOK(w, 202, t.Snapshot(false)) }