gengxin
This commit is contained in:
Ankle committed
1 parent
c460ae8fbd
commit
71c4711282
6 files changed
+303
-95
No files matched your search
+269
-78
@@ -1,16 +1,20 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
TARGET_MOUNT="/userdata"
|
||||
DATA_ROOT="/userdata/data"
|
||||
TARGET_MOUNT="${TARGET_MOUNT:-/userdata}"
|
||||
DATA_ROOT="${DATA_ROOT:-/userdata/data}"
|
||||
CONTAINER="smart-cam"
|
||||
PROCESSED_FILE="/smart-cam/kafka_sender/processed_files.log"
|
||||
STATE_DIR="/var/lib/cleanup-smartcam-data"
|
||||
PAUSED_STATE="${STATE_DIR}/capture-paused"
|
||||
LOCK_FILE="/run/cleanup-smartcam-data.lock"
|
||||
SCAN_PROCESSED_FILE="${DATA_ROOT}/upload_processed_scans.log"
|
||||
STATE_DIR="${STATE_DIR:-/var/lib/cleanup-smartcam-data}"
|
||||
RECORDING_PAUSE_MARKER="${RECORDING_PAUSE_MARKER:-${DATA_ROOT}/.recording-paused-low-disk}"
|
||||
LEGACY_PAUSED_STATE="${STATE_DIR}/capture-paused"
|
||||
LEGACY_OLD_PAUSED_STATE="/var/lib/userdata-video-cleanup/capture-paused"
|
||||
LOCK_FILE="${LOCK_FILE:-/run/cleanup-smartcam-data.lock}"
|
||||
|
||||
# Delete confirmed bundles on every scheduled run. Pause capture and raise a
|
||||
# critical alert below 3 GiB, and resume after recovery to 5 GiB.
|
||||
# Delete confirmed bundles on every scheduled run. Disable only local video
|
||||
# recording below 3 GiB; detection, RTSP/live output and upload stay running.
|
||||
# Re-enable recording after recovery to 5 GiB.
|
||||
RESUME_TARGET_BYTES="${RESUME_TARGET_BYTES:-5368709120}"
|
||||
PAUSE_BELOW_BYTES="${PAUSE_BELOW_BYTES:-3221225472}"
|
||||
CRITICAL_BELOW_BYTES="${CRITICAL_BELOW_BYTES:-3221225472}"
|
||||
@@ -79,64 +83,43 @@ stop_fast_resume_checks() {
|
||||
systemctl --no-block stop cleanup-smartcam-data-resume.timer 2>/dev/null || true
|
||||
}
|
||||
|
||||
wait_for_safe_window() {
|
||||
local newest size1 size2
|
||||
newest="$(find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f -name '*.mp4' \
|
||||
-printf '%T@ %p\n' 2>/dev/null | sort -n | tail -1 | cut -d' ' -f2-)"
|
||||
[[ -n "$newest" && -f "$newest" ]] || return 0
|
||||
size1="$(stat -c %s "$newest" 2>/dev/null || printf 0)"
|
||||
sleep "$SAFE_WINDOW_SECONDS"
|
||||
[[ -f "$newest" ]] || return 0
|
||||
size2="$(stat -c %s "$newest" 2>/dev/null || printf 0)"
|
||||
[[ "$size1" -eq "$size2" && "$size2" -gt 0 ]]
|
||||
recover_legacy_capture_pause() {
|
||||
local state_file pid
|
||||
for state_file in "$LEGACY_PAUSED_STATE" "$LEGACY_OLD_PAUSED_STATE"; do
|
||||
[[ -f "$state_file" ]] || continue
|
||||
pid="$(sed -n 's/^pid=//p' "$state_file" | head -1)"
|
||||
if valid_camera_pid "$pid" && camera_is_stopped "$pid"; then
|
||||
kill -CONT "$pid"
|
||||
alert RECOVERED "released legacy whole-process pause; detection and live stream remain running"
|
||||
fi
|
||||
rm -f "$state_file"
|
||||
done
|
||||
}
|
||||
|
||||
pause_capture() {
|
||||
local free_bytes="$1" pid
|
||||
pid="$(camera_host_pid || true)"
|
||||
if ! valid_camera_pid "$pid"; then
|
||||
alert ERROR "cannot find smart_camera process; capture was not paused"
|
||||
return 1
|
||||
fi
|
||||
if camera_is_stopped "$pid"; then
|
||||
printf 'pid=%s\npaused_at=%s\n' "$pid" "$(date --iso-8601=seconds)" > "$PAUSED_STATE"
|
||||
start_fast_resume_checks
|
||||
return 0
|
||||
fi
|
||||
|
||||
if ! wait_for_safe_window; then
|
||||
alert WARNING "no stable recording gap found; pausing capture to protect disk"
|
||||
fi
|
||||
kill -STOP "$pid"
|
||||
printf 'pid=%s\npaused_at=%s\n' "$pid" "$(date --iso-8601=seconds)" > "$PAUSED_STATE"
|
||||
start_fast_resume_checks
|
||||
alert CRITICAL "capture paused at free=$(human_bytes "$free_bytes"); SFTP/Kafka uploader remains running; automatic resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
|
||||
}
|
||||
|
||||
resume_capture_if_ready() {
|
||||
local free_bytes="$1" pid
|
||||
[[ -f "$PAUSED_STATE" ]] || return 0
|
||||
pid="$(sed -n 's/^pid=//p' "$PAUSED_STATE" | head -1)"
|
||||
|
||||
if ! valid_camera_pid "$pid"; then
|
||||
log WARNING "discarding stale capture pause state (pid=${pid:-unknown})"
|
||||
rm -f "$PAUSED_STATE"
|
||||
stop_fast_resume_checks
|
||||
return 0
|
||||
fi
|
||||
if ! camera_is_stopped "$pid"; then
|
||||
log WARNING "capture is already running; clearing stale pause state"
|
||||
rm -f "$PAUSED_STATE"
|
||||
stop_fast_resume_checks
|
||||
return 0
|
||||
fi
|
||||
if [[ "$free_bytes" -ge "$RESUME_TARGET_BYTES" ]]; then
|
||||
kill -CONT "$pid"
|
||||
rm -f "$PAUSED_STATE"
|
||||
stop_fast_resume_checks
|
||||
alert RECOVERED "free space reached $(human_bytes "$free_bytes"); capture resumed automatically"
|
||||
disable_recording() {
|
||||
local free_bytes="$1"
|
||||
if [[ ! -f "$RECORDING_PAUSE_MARKER" ]]; then
|
||||
printf 'disabled_at=%s\nfree_bytes=%s\n' \
|
||||
"$(date --iso-8601=seconds)" "$free_bytes" > "$RECORDING_PAUSE_MARKER"
|
||||
alert CRITICAL "local video recording disabled at free=$(human_bytes "$free_bytes"); videos/videos_raw/annotations stop growing; detection, RTSP/live stream and uploader remain running; automatic recording resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
|
||||
else
|
||||
log WARNING "capture remains paused: free=$(human_bytes "$free_bytes"), resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
|
||||
log WARNING "local video recording remains disabled: free=$(human_bytes "$free_bytes"), resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
|
||||
fi
|
||||
start_fast_resume_checks
|
||||
}
|
||||
|
||||
enable_recording_if_ready() {
|
||||
local free_bytes="$1"
|
||||
[[ -f "$RECORDING_PAUSE_MARKER" ]] || {
|
||||
stop_fast_resume_checks
|
||||
return 0
|
||||
}
|
||||
if [[ "$free_bytes" -ge "$RESUME_TARGET_BYTES" ]]; then
|
||||
rm -f "$RECORDING_PAUSE_MARKER"
|
||||
stop_fast_resume_checks
|
||||
alert RECOVERED "free space reached $(human_bytes "$free_bytes"); local video recording enabled automatically; detection and live stream were never stopped"
|
||||
else
|
||||
log WARNING "local video recording remains disabled: free=$(human_bytes "$free_bytes"), resume target=$(human_bytes "$RESUME_TARGET_BYTES")"
|
||||
fi
|
||||
}
|
||||
|
||||
@@ -145,24 +128,66 @@ load_processed_files() {
|
||||
if ! docker inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null | grep -qx true; then
|
||||
return 1
|
||||
fi
|
||||
docker exec "$CONTAINER" sh -c \
|
||||
'test -r "$1" && cat "$1"' sh "$PROCESSED_FILE" 2>/dev/null |
|
||||
sed -nE '/^video_[0-9]{8}_[0-9]{6}_[0-9]{3}\.xml$/p' |
|
||||
docker exec -i "$CONTAINER" /opt/venv/bin/python - 2>/dev/null <<'PY' |
|
||||
import json
|
||||
import re
|
||||
|
||||
processed_path = "/smart-cam/kafka_sender/processed_files.log"
|
||||
config_path = "/smart-cam/config/config.json"
|
||||
receipts_path = "/smart-cam/data/upload_completed_types.json"
|
||||
legacy_types = {"videos_raw", "annotations"}
|
||||
pattern = re.compile(r"^video_[0-9]{8}_[0-9]{6}_[0-9]{3}\.xml$")
|
||||
|
||||
with open(config_path, encoding="utf-8") as stream:
|
||||
upload = json.load(stream).get("upload", {})
|
||||
required = {
|
||||
name for name, default in (
|
||||
("videos", False), ("videos_raw", True), ("annotations", True)
|
||||
) if bool(upload.get(name, default))
|
||||
}
|
||||
try:
|
||||
with open(receipts_path, encoding="utf-8") as stream:
|
||||
receipts = json.load(stream)
|
||||
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
||||
receipts = {}
|
||||
|
||||
with open(processed_path, encoding="utf-8") as stream:
|
||||
for line in stream:
|
||||
name = line.strip()
|
||||
if not pattern.fullmatch(name):
|
||||
continue
|
||||
confirmed = set(receipts.get(name, {}).get("types", legacy_types))
|
||||
if required.issubset(confirmed):
|
||||
print(name)
|
||||
PY
|
||||
sort -u > "$destination"
|
||||
[[ -s "$destination" ]]
|
||||
}
|
||||
|
||||
build_safe_bundles() {
|
||||
local processed="$1" output="$2" xml base raw annotated newest_ts
|
||||
local processed="$1" output="$2" xml base raw annotated newest_ts path complete
|
||||
: > "$output"
|
||||
while IFS= read -r xml; do
|
||||
base="${xml%.xml}"
|
||||
raw="$DATA_ROOT/videos_raw/${base}_raw.mp4"
|
||||
annotated="$DATA_ROOT/videos/${base}_annotated.mp4"
|
||||
# The raw file is the authoritative large asset. Only select completed,
|
||||
# sufficiently old bundles. Missing members are harmless and skipped later.
|
||||
[[ -f "$raw" ]] || continue
|
||||
find "$raw" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null | grep -q . || continue
|
||||
complete=1
|
||||
for path in "$raw" "$annotated" "$DATA_ROOT/annotations/${base}.xml"; do
|
||||
[[ -f "$path" ]] || {
|
||||
complete=0
|
||||
break
|
||||
}
|
||||
find "$path" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
|
||||
grep -q . || {
|
||||
complete=0
|
||||
break
|
||||
}
|
||||
if command -v fuser >/dev/null 2>&1 && fuser -s "$path" 2>/dev/null; then
|
||||
complete=0
|
||||
break
|
||||
fi
|
||||
done
|
||||
[[ "$complete" -eq 1 ]] || continue
|
||||
newest_ts="$(stat -c %Y "$raw" 2>/dev/null || printf 0)"
|
||||
printf '%s %s\n' "$newest_ts" "$base" >> "$output"
|
||||
done < "$processed"
|
||||
@@ -170,9 +195,23 @@ build_safe_bundles() {
|
||||
}
|
||||
|
||||
delete_safe_bundles() {
|
||||
local bundles="$1" deleted_count=0 deleted_bytes=0 record base path size
|
||||
local bundles="$1" deleted_count=0 deleted_bytes=0 record base path size complete
|
||||
while IFS= read -r record; do
|
||||
base="${record#* }"
|
||||
complete=1
|
||||
for path in \
|
||||
"$DATA_ROOT/videos_raw/${base}_raw.mp4" \
|
||||
"$DATA_ROOT/videos/${base}_annotated.mp4" \
|
||||
"$DATA_ROOT/annotations/${base}.xml"; do
|
||||
[[ -f "$path" ]] || {
|
||||
complete=0
|
||||
break
|
||||
}
|
||||
done
|
||||
if [[ "$complete" -ne 1 ]]; then
|
||||
log WARNING "confirmed bundle became incomplete before deletion; skipped as a unit: ${base}"
|
||||
continue
|
||||
fi
|
||||
for path in \
|
||||
"$DATA_ROOT/videos_raw/${base}_raw.mp4" \
|
||||
"$DATA_ROOT/videos/${base}_annotated.mp4" \
|
||||
@@ -191,6 +230,152 @@ delete_safe_bundles() {
|
||||
log INFO "safe cleanup result: deleted=${deleted_count} files bytes=${deleted_bytes} free=$(human_bytes "$(available_bytes)")"
|
||||
}
|
||||
|
||||
delete_confirmed_scans() {
|
||||
local scan_id scan_dir resolved size
|
||||
local deleted_scans=0 deleted_bytes=0 unsafe
|
||||
|
||||
[[ -r "$SCAN_PROCESSED_FILE" ]] || {
|
||||
log INFO "no confirmed scan upload list yet; scan cleanup skipped"
|
||||
return 0
|
||||
}
|
||||
|
||||
while IFS= read -r scan_id; do
|
||||
[[ "$scan_id" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}_[0-9]{2}-[0-9]{2}-[0-9]{2}$ ]] ||
|
||||
continue
|
||||
scan_dir="$DATA_ROOT/scan/$scan_id"
|
||||
[[ -d "$scan_dir" ]] || continue
|
||||
|
||||
resolved="$(readlink -f -- "$scan_dir" 2>/dev/null || true)"
|
||||
[[ "$resolved" == "$DATA_ROOT/scan/$scan_id" ]] || {
|
||||
log WARNING "scan path failed safety check; skipped: ${scan_dir}"
|
||||
continue
|
||||
}
|
||||
|
||||
# A processed marker is only written after every image/meta SFTP upload
|
||||
# and the Kafka delivery callback have succeeded. Re-check that this is a
|
||||
# completed scan and that it has aged out before deleting the directory.
|
||||
if ! python3 - "$scan_dir/meta.json" <<'PY'
|
||||
import json
|
||||
import sys
|
||||
try:
|
||||
with open(sys.argv[1], encoding="utf-8") as stream:
|
||||
meta = json.load(stream)
|
||||
state = meta.get("state", {})
|
||||
images = meta.get("images", [])
|
||||
total = int(state.get("total", 0))
|
||||
valid = (
|
||||
state.get("state") == "completed"
|
||||
and total > 0
|
||||
and int(state.get("completed", -1)) == total
|
||||
and len(images) == total
|
||||
)
|
||||
except (OSError, ValueError, TypeError, json.JSONDecodeError):
|
||||
valid = False
|
||||
raise SystemExit(0 if valid else 1)
|
||||
PY
|
||||
then
|
||||
log WARNING "confirmed scan has invalid/incomplete meta; skipped: ${scan_id}"
|
||||
continue
|
||||
fi
|
||||
|
||||
if find "$scan_dir" -xdev -type f \
|
||||
! -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
|
||||
grep -q .; then
|
||||
continue
|
||||
fi
|
||||
|
||||
unsafe=0
|
||||
if command -v fuser >/dev/null 2>&1; then
|
||||
while IFS= read -r -d '' path; do
|
||||
if fuser -s "$path" 2>/dev/null; then
|
||||
log WARNING "confirmed scan file is still open; skipped scan=${scan_id} file=${path}"
|
||||
unsafe=1
|
||||
break
|
||||
fi
|
||||
done < <(find "$scan_dir" -xdev -type f -print0 2>/dev/null)
|
||||
fi
|
||||
[[ "$unsafe" -eq 0 ]] || continue
|
||||
|
||||
size="$(du -sb "$scan_dir" 2>/dev/null | awk '{print $1}')"
|
||||
size="${size:-0}"
|
||||
if find "$scan_dir" -xdev -depth -delete 2>/dev/null &&
|
||||
[[ ! -e "$scan_dir" ]]; then
|
||||
deleted_scans=$((deleted_scans + 1))
|
||||
deleted_bytes=$((deleted_bytes + size))
|
||||
log INFO "deleted confirmed uploaded scan (${size} bytes): ${scan_dir}"
|
||||
else
|
||||
log ERROR "failed to delete confirmed uploaded scan: ${scan_dir}"
|
||||
fi
|
||||
done < <(sort -u "$SCAN_PROCESSED_FILE")
|
||||
|
||||
log INFO "scan cleanup result: deleted=${deleted_scans} scans bytes=${deleted_bytes}"
|
||||
}
|
||||
|
||||
delete_stale_incomplete_bundles() {
|
||||
local bases base filename raw annotated xml path
|
||||
local present safe size deleted_bundles=0 deleted_files=0 deleted_bytes=0
|
||||
bases="$(mktemp /tmp/userdata-incomplete-bases.XXXXXX)"
|
||||
|
||||
find "$DATA_ROOT/videos_raw" -maxdepth 1 -type f -name 'video_*_raw.mp4' \
|
||||
-printf '%f\n' 2>/dev/null |
|
||||
sed 's/_raw\.mp4$//' >> "$bases"
|
||||
find "$DATA_ROOT/videos" -maxdepth 1 -type f -name 'video_*_annotated.mp4' \
|
||||
-printf '%f\n' 2>/dev/null |
|
||||
sed 's/_annotated\.mp4$//' >> "$bases"
|
||||
find "$DATA_ROOT/annotations" -maxdepth 1 -type f -name 'video_*.xml' \
|
||||
-printf '%f\n' 2>/dev/null |
|
||||
sed 's/\.xml$//' >> "$bases"
|
||||
sort -u "$bases" -o "$bases"
|
||||
|
||||
while IFS= read -r base; do
|
||||
[[ "$base" =~ ^video_[0-9]{8}_[0-9]{6}_[0-9]{3}$ ]] || continue
|
||||
raw="$DATA_ROOT/videos_raw/${base}_raw.mp4"
|
||||
annotated="$DATA_ROOT/videos/${base}_annotated.mp4"
|
||||
xml="$DATA_ROOT/annotations/${base}.xml"
|
||||
present=0
|
||||
[[ -f "$raw" ]] && present=$((present + 1))
|
||||
[[ -f "$annotated" ]] && present=$((present + 1))
|
||||
[[ -f "$xml" ]] && present=$((present + 1))
|
||||
[[ "$present" -gt 0 && "$present" -lt 3 ]] || continue
|
||||
|
||||
# Recording creates members at slightly different times. Only reconcile
|
||||
# bundles when every existing member is old and no process has it open.
|
||||
safe=1
|
||||
for path in "$raw" "$annotated" "$xml"; do
|
||||
[[ -f "$path" ]] || continue
|
||||
find "$path" -mmin "+${MIN_FILE_AGE_MINUTES}" -print -quit 2>/dev/null |
|
||||
grep -q . || {
|
||||
safe=0
|
||||
break
|
||||
}
|
||||
if command -v fuser >/dev/null 2>&1 && fuser -s "$path" 2>/dev/null; then
|
||||
log WARNING "incomplete bundle member is still open; skipped bundle=${base} file=${path}"
|
||||
safe=0
|
||||
break
|
||||
fi
|
||||
done
|
||||
[[ "$safe" -eq 1 ]] || continue
|
||||
|
||||
for path in "$raw" "$annotated" "$xml"; do
|
||||
[[ -f "$path" ]] || continue
|
||||
size="$(stat -c %s "$path" 2>/dev/null || printf 0)"
|
||||
if rm -f -- "$path"; then
|
||||
deleted_files=$((deleted_files + 1))
|
||||
deleted_bytes=$((deleted_bytes + size))
|
||||
log INFO "deleted stale incomplete bundle member: ${path}"
|
||||
else
|
||||
log ERROR "failed to delete stale incomplete bundle member: ${path}"
|
||||
fi
|
||||
done
|
||||
deleted_bundles=$((deleted_bundles + 1))
|
||||
done < "$bases"
|
||||
rm -f -- "$bases"
|
||||
|
||||
if [[ "$deleted_bundles" -gt 0 ]]; then
|
||||
log INFO "incomplete bundle reconciliation result: bundles=${deleted_bundles} files=${deleted_files} bytes=${deleted_bytes}"
|
||||
fi
|
||||
}
|
||||
|
||||
delete_stale_zero_bundles() {
|
||||
local raw base path deleted_bundles=0 deleted_files=0
|
||||
while IFS= read -r -d '' raw; do
|
||||
@@ -313,19 +498,21 @@ delete_stale_zero_part_bundles() {
|
||||
|
||||
main() {
|
||||
local free_bytes processed bundles
|
||||
recover_legacy_capture_pause
|
||||
delete_stale_zero_bundles
|
||||
delete_stale_zero_part_bundles
|
||||
delete_stale_incomplete_bundles
|
||||
free_bytes="$(available_bytes)"
|
||||
|
||||
# Protect the disk before cleanup work: once free space is below 3 GiB,
|
||||
# stop continuous detection/recording/inference first while upload stays up.
|
||||
# Protect the disk before cleanup work. Only recording is disabled; continuous
|
||||
# detection, inference, RTSP/live output and upload remain available.
|
||||
if [[ "$free_bytes" -lt "$CRITICAL_BELOW_BYTES" ]]; then
|
||||
alert CRITICAL "free space below critical threshold: $(human_bytes "$free_bytes")"
|
||||
fi
|
||||
if [[ "$free_bytes" -lt "$PAUSE_BELOW_BYTES" ]]; then
|
||||
pause_capture "$free_bytes"
|
||||
disable_recording "$free_bytes"
|
||||
fi
|
||||
resume_capture_if_ready "$free_bytes"
|
||||
enable_recording_if_ready "$free_bytes"
|
||||
|
||||
processed="$(mktemp /tmp/userdata-processed.XXXXXX)"
|
||||
bundles="$(mktemp /tmp/userdata-bundles.XXXXXX)"
|
||||
@@ -342,14 +529,18 @@ main() {
|
||||
alert ERROR "cannot read confirmed-upload list; fail-safe mode prevents all deletion"
|
||||
fi
|
||||
|
||||
delete_confirmed_scans
|
||||
|
||||
free_bytes="$(available_bytes)"
|
||||
if [[ "$free_bytes" -lt "$CRITICAL_BELOW_BYTES" ]]; then
|
||||
alert CRITICAL "free space below critical threshold: $(human_bytes "$free_bytes")"
|
||||
fi
|
||||
if [[ "$free_bytes" -lt "$PAUSE_BELOW_BYTES" ]]; then
|
||||
pause_capture "$free_bytes"
|
||||
disable_recording "$free_bytes"
|
||||
fi
|
||||
resume_capture_if_ready "$(available_bytes)"
|
||||
enable_recording_if_ready "$(available_bytes)"
|
||||
}
|
||||
|
||||
main "$@"
|
||||
if [[ "${CLEANUP_SMARTCAM_LIB_ONLY:-0}" != "1" ]]; then
|
||||
main "$@"
|
||||
fi
|
||||
Reference in new issue
Block a user