techy/Media-Server-Setup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Logging framework: logfmt events in our scripts, container labels + rotated json-file logs, Caddy JSON access logs; LOGGING.md

Browse Source 
Store-agnostic: journald + Docker logs are the only taps any collector (Alloy/Loki, VictoriaLogs, Netdata) needs.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Brandyn / Techy
2026-11-01 19:53:05 -06:00
parent 88bfda4c9d
commit 704bc8c4d3
10 changed files with 186 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
bin/dumpd.py
View File

@@ -134,7 +134,7 @@ def quota_enforce_once():
try:
folders = st("GET", "/rest/config/folders")
except Exception as e:
print(f"{datetime.now(TZ):%F %T} quota: syncthing unreachable ({e})", flush=True); return
log("quota_unreachable", level="warn", error=e); return
for f in folders:
if not f["id"].startswith("clips-"): continue
user = f["id"][len("clips-"):]
@@ -144,12 +144,12 @@ def quota_enforce_once():
over = used > q; under = used < q * 0.9
if over and not f.get("paused"):
try: st("POST", f"/rest/db/pause?folder={f['id']}"); paused[user] = {"at": datetime.now(TZ).isoformat(), "used_gb": round(used / 1e9, 1), "quota_gb": q / 1e9}; changed = True
except Exception as e: print(f"{datetime.now(TZ):%F %T} quota: pause {user} failed: {e}", flush=True); continue
print(f"{datetime.now(TZ):%F %T} quota: PAUSED clips-{user} ({used/1e9:.1f} GB > {q/1e9:.0f} GB)", flush=True)
except Exception as e: log("quota_pause_failed", level="error", user=user, error=e); continue
log("quota_paused", level="warn", user=user, folder=f"clips-{user}", used_gb=f"{used/1e9:.1f}", quota_gb=f"{q/1e9:.0f}")
elif under and f.get("paused") and user in paused:
try: st("POST", f"/rest/db/resume?folder={f['id']}"); paused.pop(user, None); changed = True
except Exception as e: print(f"{datetime.now(TZ):%F %T} quota: resume {user} failed: {e}", flush=True); continue
print(f"{datetime.now(TZ):%F %T} quota: resumed clips-{user} ({used/1e9:.1f} GB)", flush=True)
except Exception as e: log("quota_resume_failed", level="error", user=user, error=e); continue
log("quota_resumed", user=user, folder=f"clips-{user}", used_gb=f"{used/1e9:.1f}")
if changed:
try: QUOTA_STATE.parent.mkdir(exist_ok=True); QUOTA_STATE.write_text(json.dumps(paused, indent=2))
except OSError: pass
@@ -157,7 +157,7 @@ def quota_enforce_once():
def quota_loop():
while True:
try: quota_enforce_once()
except Exception as e: print(f"{datetime.now(TZ):%F %T} quota loop error: {e}", flush=True)
except Exception as e: log("quota_loop_error", level="error", error=e)
_time.sleep(QUOTA_CHECK_S)
def st_apikey():
@@ -247,7 +247,7 @@ def clips_setup(user, device_id, device_name=None):
cur["devices"].append({"deviceID": device_id})
cur["type"] = "receiveonly"; cur["minDiskFree"] = spec["minDiskFree"]
st("PUT", f"/rest/config/folders/{fid}", cur)
print(f"{datetime.now(TZ):%F %T} clips setup user={user} device={device_id[:7]}… folder={fid}", flush=True)
log("clips_setup", user=user, device=device_id[:7], folder=fid)
return clips_status(user)
DEPLOY_ENV = Path(os.environ.get("DEPLOY_ENV_FILE", "/etc/vanguard/deploy.env"))
@@ -267,17 +267,25 @@ def run_deploy(reason):
_deploy_lock["running"] = True
def go():
try:
print(f"{datetime.now(TZ):%F %T} deploy: start ({reason})", flush=True)
log("deploy_start", reason=reason)
r = subprocess.run(["/usr/local/bin/vanguard-deploy", "pull"], capture_output=True, text=True, timeout=1800)
for line in (r.stdout + r.stderr).splitlines()[-40:]: print(f"deploy| {line}", flush=True)
print(f"{datetime.now(TZ):%F %T} deploy: exit {r.returncode}", flush=True)
log("deploy_exit", level=("info" if r.returncode == 0 else "error"), rc=r.returncode)
except Exception as e:
print(f"{datetime.now(TZ):%F %T} deploy: error {e}", flush=True)
log("deploy_error", level="error", error=e)
finally:
_deploy_lock["running"] = False
threading.Thread(target=go, daemon=True).start()
return True
def log(event, level="info", **kv):
"""One logfmt line per event (see LOGGING.md): ts= level= component=dumpd event= key=value ..."""
def q(v):
v = str(v)
return v if v and not re.search(r'[\s"=]', v) else '"' + v.replace("\\", "\\\\").replace('"', '\\"') + '"'
print(f"ts={datetime.now(TZ).isoformat(timespec='seconds')} level={level} component=dumpd event={event} "
+ " ".join(f"{k}={q(v)}" for k, v in kv.items() if v is not None), flush=True)
def token_secret():
try:
sec = TOKEN_SECRET_FILE.read_bytes().strip()
@@ -394,7 +402,7 @@ class H(BaseHTTPRequestHandler):
env = deploy_env(); secret = env.get("DEPLOY_WEBHOOK_SECRET", "")
sig = self.headers.get("X-Gitea-Signature") or self.headers.get("X-Hub-Signature-256", "").replace("sha256=", "")
if not secret or not sig or not hmac.compare_digest(hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest(), sig):
print(f"{datetime.now(TZ):%F %T} deploy webhook: bad signature from {self.client_address[0]}", flush=True)
log("deploy_webhook_bad_signature", level="warn", ip=self.client_address[0])
return self._json(401, {"error": "bad signature"})
try: payload = json.loads(raw or b"{}")
except json.JSONDecodeError: return self._json(400, {"error": "bad json"})
@@ -403,7 +411,7 @@ class H(BaseHTTPRequestHandler):
if ref != f"refs/heads/{branch}":
return self._json(200, {"ignored": f"ref {ref} is not {branch}"})
if env.get("DEPLOY_AUTO", "0") != "1":
print(f"{datetime.now(TZ):%F %T} deploy webhook: push {after} by {pusher} on {branch} — auto-deploy disabled (DEPLOY_AUTO=0)", flush=True)
log("deploy_webhook_push", rev=after, pusher=pusher, branch=branch, auto="0")
return self._json(202, {"queued": False, "note": "auto-deploy disabled (DEPLOY_AUTO=0); run vanguard-deploy pull by hand"})
started = run_deploy(f"push {after} by {pusher}")
return self._json(202, {"queued": started, "note": "deploy started" if started else "a deploy is already running"})
@@ -426,19 +434,19 @@ class H(BaseHTTPRequestHandler):
# audio mirrors (<user>/<Title>.webrtc) are published by the hook inside the mediamtx container only
if req.get("ip") in ("127.0.0.1", "::1") and re.fullmatch(r"[a-z0-9_.-]+/[a-z0-9_.-]+\.webrtc", path):
return self._json(200, {"ok": True})
print(f"{datetime.now(TZ):%F %T} mtx-auth DENY publish path={path} ip={req.get('ip')} (mirror paths are internal)", flush=True)
log("publish_deny", level="warn", stream=path, ip=req.get("ip"), reason="mirror paths are internal")
return self._json(401, {"error": ".webrtc paths are reserved"})
cred = req.get("password") or req.get("token") or ""
if cred.lower().startswith("bearer "): cred = cred[7:]
if ":" in cred and "." in cred.split(":", 1)[1]: cred = cred.split(":", 1)[1] # "user:token" form
user = verify_token(cred)
if not user:
print(f"{datetime.now(TZ):%F %T} mtx-auth DENY publish path={path} ip={req.get('ip')} (bad/expired token)", flush=True)
log("publish_deny", level="warn", stream=path, ip=req.get("ip"), reason="bad or expired token")
return self._json(401, {"error": "invalid or expired stream token"})
if not re.fullmatch(r"[a-z0-9_.-]+/[a-z0-9_.-]+", path) or not path.startswith(user + "/"):
print(f"{datetime.now(TZ):%F %T} mtx-auth DENY publish user={user} path={path} (not their prefix)", flush=True)
log("publish_deny", level="warn", user=user, stream=path, reason="not their prefix")
return self._json(401, {"error": "path must be <your-username>/<Title>"})
print(f"{datetime.now(TZ):%F %T} mtx-auth allow publish user={user} path={path} ip={req.get('ip')}", flush=True)
log("publish_allow", user=user, stream=path, ip=req.get("ip"))
return self._json(200, {"ok": True})
if self.path.startswith("/me/clips"):
who, _ = identity(self.headers)
@@ -465,7 +473,7 @@ class H(BaseHTTPRequestHandler):
moved, size = do_move(files, night)
MARK.mkdir(exist_ok=True)
(MARK / f"{night}-{tgt}.dumped").write_text(f"{datetime.now(TZ).isoformat()} by {u}\n")
print(f"{datetime.now(TZ):%F %T} dump night={night} user={tgt} by={u} files={moved} gb={size/1e9:.2f}", flush=True)
log("dump", night=night, user=tgt, by=u, files=moved, gb=f"{size/1e9:.2f}")
self._json(200, {"night": night, "user": tgt, "moved": moved, "gb": round(size / 1e9, 2),
"undo_until": (datetime.now(TZ) + timedelta(days=3)).strftime("%F")})
def log_message(self, *a): # quiet

2
bin/mtx-on-ready.sh
View File

@@ -25,7 +25,7 @@ case "$tracks" in
*"MPEG-4 Audio"*|*"MPEG-1 Audio"*|*G711*|*G722*|*LPCM*|*AC-3*) ;;
*) exit 0 ;; # no (recognised) audio: nothing to mirror
esac
echo "mirror: $MTX_PATH -> $MTX_PATH.webrtc (audio -> Opus)"
echo "event=mirror stream=$MTX_PATH target=$MTX_PATH.webrtc audio=opus"
n=0
while [ $n -lt 10 ]; do # retry: the source may not be readable for the first second
ffmpeg -nostdin -hide_banner -loglevel warning -fflags nobuffer -flags low_delay \

23
bin/normalize-sweep.sh
View File

@@ -36,7 +36,9 @@ MAX_W=2560; MAX_H=1440; MAX_FPS=60
exec 9>"${LOCK_FILE:-/run/lock/normalize-sweep.lock}"
flock -n 9 || exit 0
log() { echo "$(date '+%F %T') $*"; }
# logfmt (LOGGING.md): vglog <level> event=<verb> key=value ... values with spaces must arrive pre-quoted
vglog() { local lvl="$1"; shift; printf 'ts=%s level=%s component=sweep %s\n' "$(date -Is)" "$lvl" "$*"; }
log() { vglog info "msg=\"$*\""; }
dur() { ffprobe -v error -show_entries format=duration -of csv=p=0 "$1" 2>/dev/null | cut -d. -f1; }
# --- probe: prints "verdict reason" where verdict ∈ ok|remux|encode|skip ----------------------
@@ -87,10 +89,11 @@ find "$ROOT/streams" "$ROOT/clips" -type f \
! -path '*/.src/*' ! -path '*/.stfolder/*' ! -path '*/.stversions/*' ! -name '*.part' -mmin +"$MINS_IDLE" -print0 2>/dev/null |
while IFS= read -r -d '' src; do
base="${src%.*}"; ext="${src##*.}"
live=0
live=0; stream=""
if [[ $(basename "$(dirname "$src")") == .live ]]; then
live=1
titledir="$(dirname "$(dirname "$src")")" # streams/<user>/<Title>
stream="$(basename "$(dirname "$titledir")")/$(basename "$titledir")" # user/Title, the log correlation key
base="$(dirname "$titledir")/$(basename "$titledir")__$(basename "${src%.*}")" # streams/<user>/<Title>__<ts>
fi
synced=0; stroot=""
@@ -107,7 +110,7 @@ while IFS= read -r -d '' src; do
if (( live )); then
[[ $verdict == skip ]] && continue
[[ $verdict == ok ]] && verdict=remux # must leave .live/ either way
out="$base.mp4"; [[ -e "$out" ]] && { log "exists $out (leaving $src)"; continue; }
out="$base.mp4"; [[ -e "$out" ]] && { vglog info "event=skip reason=exists file=\"$out\" src=\"$src\""; continue; }
else
case "$verdict" in ok|skip) continue ;; esac
out="$base.mp4"
@@ -118,7 +121,7 @@ while IFS= read -r -d '' src; do
[[ -e "$out" ]] && continue
fi
tmp="$out.part"
log "$verdict $src ($reason)"
vglog info "event=verdict verdict=$verdict file=\"$src\" reason=\"$reason\" stream=\"$stream\""
tcargs=(); tcjson=""; band=0; deband=0
if (( live )); then
# timecode: decode the overscan strip (server clock) or fall back to the server arrival time in the filename
@@ -139,18 +142,18 @@ PYEOF
if [[ -n ${TC_TIMECODE:-} ]]; then
tcargs=(-timecode "$TC_TIMECODE" -metadata "creation_time=$TC_CREATION" -write_tmcd 1)
tcjson="${TC_JSON:-}"
log "tc $src source=$TC_SOURCE timecode=$TC_TIMECODE"
vglog info "event=timecode file=\"$src\" stream=\"$stream\" source=$TC_SOURCE timecode=\"$TC_TIMECODE\""
fi
# the band has done its job once decoded: remove it from the picture (re-encode with a crop)
if [[ ${TC_SOURCE:-} == strip && ${TC_BAND:-0} -gt 0 ]]; then
band=$TC_BAND; deband=1; verdict=encode; reason="remove ${band}px timecode band"
log "deband $src cropping ${band} rows (re-encode, timing preserved)"
vglog info "event=deband file=\"$src\" stream=\"$stream\" band_rows=$band"
fi
# opt-in gap filling: switch this stream from remux to a CFR re-encode when the holes add up to enough
if (( FILL_GAPS_MS > 0 )) && [[ $verdict == remux && -n $tcjson ]]; then
missing=$(python3 -c 'import json,sys; d=json.loads(sys.argv[1]); print(d.get("gaps",{}).get("total_missing_ms",0))' "$tcjson" 2>/dev/null || echo 0)
if (( missing >= FILL_GAPS_MS )); then
verdict=encode; reason="fill-gaps ${missing}ms"; log "gaps $src ${missing} ms missing -> CFR re-encode (FILL_GAPS_MS=$FILL_GAPS_MS)"
verdict=encode; reason="fill-gaps ${missing}ms"; vglog info "event=fill_gaps file=\"$src\" stream=\"$stream\" missing_ms=$missing"
fi
fi
fi
@@ -189,7 +192,7 @@ PYEOF
rm -f "$src" # band removal: the cropped file IS the recording; raw-with-band is not kept
fi
rmdir "$(dirname "$src")" "$titledir" 2>/dev/null || true # drop .live and the empty title folder
log "done $out"; continue
vglog info "event=published file=\"$out\" stream=\"$stream\" verdict=$verdict"; continue
elif (( synced )); then
rel_src="${src#"$stroot"/}"; rel_out="${out#"$stroot"/}"
{ echo "/$rel_out"; echo "/$rel_src"; } >> "$stroot/.stlocal" # ignore our output and the raw
@@ -205,10 +208,10 @@ PYEOF
mv -f "$tmp" "$out"
fi
chown 1000:1000 "$out" 2>/dev/null
log "done $out"
vglog info "event=done file=\"$out\" verdict=$verdict"
else
rm -f "$tmp"
{ echo "$(date '+%F %T') $verdict failed rc=$rc src_dur=${d1:-?} out_dur=${d2:-?} ($reason)"; } > "$base.normalize-failed.log"
log "FAILED $src (see $base.normalize-failed.log)" >&2
vglog error "event=failed file=\"$src\" stream=\"$stream\" verdict=$verdict rc=$rc log=\"$base.normalize-failed.log\"" >&2
fi
done

17
bin/vgdeband.sh
View File

@@ -8,23 +8,24 @@
# -p IOSchedulingClass=idle /usr/local/bin/vgdeband.sh ; follow with journalctl -fu vgdeband
set -uo pipefail
ROOT="${MEDIA_ROOT:-/srv/media}"
log() { echo "$(date '+%F %T') $*"; }
vglog() { local lvl="$1"; shift; printf 'ts=%s level=%s component=vgdeband %s\n' "$(date -Is)" "$lvl" "$*"; }
log() { vglog info "msg=\"$*\""; }
dur() { ffprobe -v error -show_entries format=duration -of csv=p=0 "$1" 2>/dev/null | cut -d. -f1; }
if (( $# )); then files=("$@"); else mapfile -t files < <(find "$ROOT/streams" -mindepth 2 -maxdepth 2 -type f -name '*.mp4' | sort); fi
for f in "${files[@]}"; do
side="${f%.*}.json"
[[ -f $side ]] || { log "skip $f (no sidecar)"; continue; }
[[ -f $side ]] || { vglog info "event=skip reason=no_sidecar file=\"$f\""; continue; }
read -r src removed tc ctime < <(python3 -c '
import json,sys; d=json.load(open(sys.argv[1]))
print(d.get("source","none"), 1 if d.get("band_removed") else 0, d.get("timecode",""), d.get("timestamp",""))' "$side")
[[ $src == strip ]] || { log "skip $f (source=$src)"; continue; }
(( removed )) && { log "skip $f (band already removed)"; continue; }
fuser -s "$f" 2>/dev/null && { log "skip $f (in use)"; continue; }
[[ $src == strip ]] || { vglog info "event=skip reason=source_$src file=\"$f\""; continue; }
(( removed )) && { vglog info "event=skip reason=already_removed file=\"$f\""; continue; }
fuser -s "$f" 2>/dev/null && { vglog info "event=skip reason=in_use file=\"$f\""; continue; }
w=$(ffprobe -v error -select_streams v:0 -show_entries stream=width -of csv=p=0 "$f")
band=$(python3 /usr/local/bin/vgstrip.py band "$w")
tmp="$f.part"
tcargs=(); [[ -n $tc ]] && tcargs=(-timecode "$tc" -write_tmcd 1); [[ -n $ctime ]] && tcargs+=(-metadata "creation_time=$ctime")
log "deband $f (${w}px wide -> crop ${band} rows; tc=$tc)"
vglog info "event=deband file=\"$f\" width=$w band_rows=$band timecode=\"$tc\""
nice -n 15 ionice -c 3 ffmpeg -nostdin -hide_banner -loglevel error -y -i "$f" \
-map 0:v:0 -map '0:a?' \
-c:v libx264 -preset veryfast -crf 18 -maxrate 30M -bufsize 60M -pix_fmt yuv420p -profile:v high -threads "${ENCODE_THREADS:-2}" \
@@ -36,8 +37,8 @@ print(d.get("source","none"), 1 if d.get("band_removed") else 0, d.get("timecode
python3 -c '
import json,sys; p=sys.argv[1]; d=json.load(open(p)); d["band_rows"]=int(sys.argv[2]); d["band_removed"]=True
json.dump(d, open(p,"w"))' "$side" "$band"; chown 1000:1000 "$side" 2>/dev/null
log "done $f"
vglog info "event=done file=\"$f\""
else
rm -f "$tmp"; log "FAILED $f rc=$rc src_dur=${d1:-?} out_dur=${d2:-?} (file left untouched)" >&2
rm -f "$tmp"; vglog error "event=failed file=\"$f\" rc=$rc src_dur=${d1:-?} out_dur=${d2:-?}" >&2
fi
done