From 83f980f7f8ec962afd8a2cf854bd1fa4ace9bf65 Mon Sep 17 00:00:00 2001 From: Ryan Shpeherd Date: Tue, 8 Sep 2026 13:51:31 -0400 Subject: [PATCH] After updates --- README.md | 207 +++++++++++++++++++++++++++++++++++++++++++++++++ config.yaml | 8 +- db/schema.sql | 14 ++++ src/main.py | 25 +++++- src/scanner.py | 166 ++++++++++++++++++++++++++++++++++++--- 5 files changed, 409 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index ba71a14..cd48e8f 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,213 @@ open http://localhost:3000 (admin / your_grafana_password) open http://localhost:9090 ``` +## System Operation + +### How Processes Start + +**Service Initialization:** + +1. **MariaDB** starts first with health check +2. **Worker** initializes via `src/main.py`: + - Loads `config.yaml` + - Sets up JSON logging with rotation + - Connects to MariaDB (connection pooling) + - Initializes database schema + - Verifies GPU availability (CUDA/PyTorch) + - Starts the `DirectoryScanner` in a background thread (scans permanent storage in place) + - Creates `WorkerPool` with 1 worker thread + - Enters job processing loop +3. **UI** starts Flask review interface via Gunicorn (2 workers) +4. **Monitoring** starts Prometheus and Grafana independently + +### Processing Pipeline + +The worker follows this flow for each video: + +``` +Pending → Lock → Probe → Sample → Detect → Classify → Aggregate → Route → Persist → Export → Cleanup → Completed +``` + +**Detailed Steps:** + +1. **Job Queue** - Atomically lock `PENDING` jobs via `UPDATE status = 'PROCESSING'` + - Priority: newest files first (`last_scan_time DESC`) + - Max concurrent: 1 per GPU + +2. **Probe Video** - Extract metadata via FFprobe + - Duration, codec, resolution + - Validate against codec whitelist (H.264, H.265, VP8/9, AV1) + - Mark `UNSCANNABLE` if invalid + +3. **Sample Frames** - Extract frames at configured interval (default: 30s) + - Save as JPEG to `/scratch/{video_id}/frames/` + - Quality: 2 (lower=better) + +4. **Detect Faces** - YOLOv8n TensorRT inference (FP32) + - Batch size auto-tuned by GPU memory monitor + - NMS filtering (IoU: 0.45, confidence: 0.25) + - Cap: 10 faces/frame, 100 faces/video + +5. **Extract Crops** - Resize detected faces to 224×224 + - Save to `/scratch/{video_id}/crops/` + +6. **Classify Crops** - MobileNetV3-Small TensorRT inference + - Temperature-scaled softmax (T=1.0) + - Returns confidence per crop + +7. **Aggregate Confidence** - Combine crop confidences into video-level score + - Strategy: `max` (most conservative) + - Alternatives: `weighted_mean`, `top_k_mean` + +8. **Route Decision** - Threshold-based routing: + - `C ≥ 0.75` → **MATCH** + - `0.45 ≤ C < 0.75` → **REVIEW** (human annotation) + - `C < 0.45` → **SKIP** + - No faces → **SKIP** + +9. **Persist Results** - Atomic transaction: + - Update `videos` table (confidence, routing, status) + - Insert `processing_logs` row (audit trail) + - State guard: only update if `status='PROCESSING'` + +10. **Export** - Buffer and batch export (default: 100 videos) + - Format: Parquet with Snappy compression + - Path: `/data/output/{model_version}/` + - Fallback: JSONL if Parquet fails + +11. **Cleanup** - Delete `/scratch/{video_id}/` directory + - Only after successful persistence + - Prevents orphaned scratch files + +**Directory Scanner Service (runs alongside the worker):** +- Scans the permanent storage location **in place** (no staging/copy step) +- Walks `/data/input` every **2 hours** by default (configurable via `scanner.scan_interval_seconds`) +- Detects new, modified, and removed video files by comparing against the DB +- Filters to video files by extension +- Computes SHA256 hash, probes metadata, validates codec +- Queues any video that has not been scanned yet as `PENDING` for the worker pool +- **Single-instance guard:** an in-process lock plus a DB lock (with a lease) ensure only one scan runs at a time — a long-running scan never overlaps another, even across multiple worker replicas. The lock lease is refreshed via heartbeats during the scan and is taken over automatically if a scanner crashes. + +### Configuration Reference + +All configuration is in `config.yaml`. Environment variable override format: `VD_
_` (e.g., `VD_SAMPLING_INTERVAL_SECONDS=60`). + +#### Key Configuration Sections + +**Sampling & Thresholds:** +```yaml +sampling: + interval_seconds: 30 # Frame extraction frequency + quality: 2 # JPEG quality (1-31, lower=better) + format: jpeg + +thresholds: + T_high: 0.75 # MATCH threshold + T_low: 0.45 # REVIEW threshold +``` + +**GPU & Batching:** +```yaml +gpu: + max_memory_gb: 18 # Target VRAM usage + batch_size: auto # Auto-tune based on available VRAM + +batching: + max_batch_size: 16 # Maximum batch size + vram_target_gb: 16 # Target VRAM for batch tuning + vram_reduce_threshold_gb: 16 # Reduce batch if above + vram_increase_threshold_gb: 10 # Increase batch if below +``` + +**Storage Paths:** +```yaml +storage: + scratch_path: /scratch # Temporary processing (tmpfs) + input_path: /data/input # Source videos (NFS) + output_path: /data/output # Results (local/NAS) + models_path: /models # TensorRT models + training_path: /data/training # Training data +``` + +**Database:** +```yaml +database: + host: mariadb + port: 3306 + name: videodetect + user: videodetect + password: videodetect123 + pool_size: 20 # Connection pool size + pool_min: 5 + pool_recycle: 3600 # Recycle connections after 1h +``` + +**Face Detection:** +```yaml +face_detection: + model: yolo8n + model_path: /models/face_detector/face_detector.trt + input_size: 640 + confidence_threshold: 0.25 + iou_threshold: 0.45 + max_faces_per_frame: 10 + max_faces_per_video: 100 +``` + +**Classification & Aggregation:** +```yaml +classifier: + model: mobilenetv3-small + model_path: /models/classifier/classifier.trt + input_size: 224 + temperature: 1.0 # Calibration temperature + +aggregation: + strategy: max # max, weighted_mean, top_k_mean + alpha: 1.0 # weighted_mean weight for mean + beta: 0.1 # weighted_mean weight for variance + top_k: 3 # top_k_mean: average top 3 scores +``` + +**Export:** +```yaml +export: + format: parquet # parquet, jsonl, or both + compression: snappy + batch_size: 100 # Export after N videos + include_frame_confidences: true +``` + +**Review UI:** +```yaml +review_ui: + host: "0.0.0.0" + port: 5000 + per_page: 20 # Pagination + top_k_frames: 5 # Show top-k contributing frames + auth_enabled: false # No auth per TC-06 + ssl_enabled: false # Internal LAN only +``` + +#### Volume Mounts + +From `docker-compose.yml`: + +- **Input**: NFS mount → `/data/input` (read-only) +- **Output**: `./output` → `/data/output` +- **Models**: `./models` → `/models` +- **Training**: `./training` → `/data/training` +- **Scratch**: 100GB tmpfs at `/scratch` (RAM disk) + +### Key Design Principles + +- **Atomic state transitions** - Database locks prevent race conditions +- **Crash recovery** - `PROCESSING` jobs automatically requeued on restart +- **Idempotent** - Re-running same video produces same result +- **Stateless** - Scratch cleanup after each job +- **Fail-safe** - 3 retry attempts before marking `ERROR` +- **No auth/SSL** - Internal LAN deployment per TC-06 + ## Project Structure ``` diff --git a/config.yaml b/config.yaml index aef05d3..5e265e6 100644 --- a/config.yaml +++ b/config.yaml @@ -70,12 +70,18 @@ model: # ----------------------------------------------------- # Directory Scanner # ----------------------------------------------------- +# Scans the permanent storage location in place (no staging/copy step). +# The corpus is large (~163k files / ~41TB), so the default interval is 2 hours. +# A single-instance guard (in-process lock + DB lock with a lease) ensures a +# long-running scan never overlaps another scan, even across replicas. scanner: - scan_interval_seconds: 60 + scan_interval_seconds: 7200 # 2 hours (adaptive: raise for very large corpora) walker_threads: 8 ffprobe_timeout_seconds: 10 hash_algorithm: sha256 hash_chunk_size_mb: 1 + lock_lease_seconds: 21600 # 6 hours: max time a scan may hold the lock before it is considered stale + heartbeat_interval_files: 500 # refresh the lock lease every N files processed # ----------------------------------------------------- # Codec Validation diff --git a/db/schema.sql b/db/schema.sql index 331b286..1cef8cb 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -115,6 +115,20 @@ CREATE TABLE IF NOT EXISTS scan_history ( error_message TEXT DEFAULT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +-- ----------------------------------------------------- +-- Table: scanner_lock +-- Single-instance guard for the directory scanner. Ensures only one scan +-- runs at a time across all replicas. A lease (locked_at + lease_seconds) +-- lets a live scanner keep the lock via heartbeats, and lets a crashed +-- scanner's lock be taken over once it goes stale. +-- ----------------------------------------------------- +CREATE TABLE IF NOT EXISTS scanner_lock ( + lock_name VARCHAR(64) PRIMARY KEY, + owner VARCHAR(128) NOT NULL COMMENT 'Instance id (host-pid-uuid) holding the lock', + locked_at DATETIME NOT NULL COMMENT 'Last time the lock was acquired or heartbeated', + lease_seconds INT NOT NULL DEFAULT 21600 COMMENT 'Lock is stale if older than this (6 hours)' +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + -- ----------------------------------------------------- -- Insert default model entry -- ----------------------------------------------------- diff --git a/src/main.py b/src/main.py index ef7ee9f..4e1f8b1 100644 --- a/src/main.py +++ b/src/main.py @@ -8,6 +8,7 @@ Initializes all components and starts the processing pipeline. import logging import signal import sys +import threading import time from pathlib import Path @@ -18,6 +19,7 @@ from config_loader import get_config from db_connector import DBConnector from logging_config import setup_logging from orchestrator import WorkerPool +from scanner import DirectoryScanner logger = logging.getLogger(__name__) @@ -84,7 +86,27 @@ def main(): except ImportError: logger.warning("PyTorch not installed. GPU features disabled.") - logger.info("Worker initialization complete. Starting processing loop...") + # Initialize the directory scanner. + # It scans the permanent storage location in place (no staging/copy step), + # periodically discovering new/removed video files and queueing any that + # have not been scanned yet as PENDING for the worker pool to pick up. + storage_config = config.get_section("storage") + scanner_config = config.get_section("scanner") + scanner = DirectoryScanner( + db_connector=db, + config=config.data, + input_path=storage_config.get("input_path", "/data/input"), + scan_interval=int(scanner_config.get("scan_interval_seconds", 60)), + walker_threads=int(scanner_config.get("walker_threads", 8)), + ) + + logger.info("Worker initialization complete. Starting scanner and processing loop...") + + # Run the scanner in a background thread (its start() is a blocking loop). + scanner_thread = threading.Thread( + target=scanner.start, name="directory-scanner", daemon=True + ) + scanner_thread.start() pool = WorkerPool(db, config.data, max_workers=1) @@ -93,6 +115,7 @@ def main(): except KeyboardInterrupt: logger.info("Worker shutting down.") pool.stop() + scanner.stop() if __name__ == "__main__": diff --git a/src/scanner.py b/src/scanner.py index 6f0fc79..ebdd973 100644 --- a/src/scanner.py +++ b/src/scanner.py @@ -10,9 +10,12 @@ import hashlib import json import logging import os +import socket +import threading import time +import uuid from concurrent.futures import ThreadPoolExecutor, as_completed -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Optional, Tuple @@ -54,29 +57,164 @@ class DirectoryScanner: self._total_files_modified = 0 self._total_files_unscannable = 0 + # Single-instance guard: prevents overlapping scans both within this + # process (threading lock) and across replicas (DB lock with a lease). + self._instance_id = f"{socket.gethostname()}-{os.getpid()}-{uuid.uuid4().hex[:8]}" + self._lock_name = "directory_scanner" + self._lock_lease_seconds = int( + config.get("scanner", {}).get("lock_lease_seconds", 21600) + ) + self._heartbeat_every = int( + config.get("scanner", {}).get("heartbeat_interval_files", 500) + ) + self._scan_lock = threading.Lock() + self._lock_table_ensured = False + def start(self): - """Start the scanner loop.""" + """Start the scanner loop. + + Each cycle is guarded so that at most one scan runs at a time: + - an in-process threading lock prevents re-entrant scans, and + - a DB lock (with a lease) prevents overlapping scans across replicas. + A long-running scan keeps its lease alive via heartbeats, so the next + scheduled tick (or another replica) waits instead of starting a second + parallel scan. + """ self._running = True - logger.info("Scanner starting: input_path=%s interval=%ds threads=%d", - self.input_path, self.scan_interval, self.walker_threads) + logger.info( + "Scanner starting: input_path=%s interval=%ds threads=%d instance=%s", + self.input_path, self.scan_interval, self.walker_threads, self._instance_id, + ) while self._running: + # In-process re-entrancy guard: never run two scans at once. + if not self._scan_lock.acquire(blocking=False): + logger.warning("A scan is already in progress; skipping this cycle.") + self._sleep_interval() + continue + try: - self._run_scan() + if self.acquire_lock(): + try: + self._run_scan() + finally: + self.release_lock() + else: + logger.info( + "Scanner lock held by another instance; skipping this cycle." + ) except Exception as e: logger.error("Scanner error: %s", e, exc_info=True) + finally: + self._scan_lock.release() # Sleep until next scan - for _ in range(self.scan_interval): - if not self._running: - break - time.sleep(1) + self._sleep_interval() + + def _sleep_interval(self): + """Sleep for the scan interval, waking early if stopped.""" + for _ in range(self.scan_interval): + if not self._running: + break + time.sleep(1) def stop(self): """Stop the scanner.""" self._running = False logger.info("Scanner stopping. Total scans: %d", self._scan_count) + # ------------------------------------------------------------------ + # Single-instance lock (cross-process / cross-replica guard) + # ------------------------------------------------------------------ + def _ensure_lock_table(self): + """Create the scanner_lock table if it does not already exist.""" + if self._lock_table_ensured: + return + self.db.execute( + """CREATE TABLE IF NOT EXISTS scanner_lock ( + lock_name VARCHAR(64) PRIMARY KEY, + owner VARCHAR(128) NOT NULL, + locked_at DATETIME NOT NULL, + lease_seconds INT NOT NULL DEFAULT 21600 + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""" + ) + self._lock_table_ensured = True + + def acquire_lock(self) -> bool: + """Attempt to acquire the cross-process scanner lock. + + Returns True if this instance now owns the lock, False otherwise. + A stale lock (held longer than the lease) is taken over so a crashed + scanner does not block scanning forever. + """ + try: + self._ensure_lock_table() + except Exception as e: + logger.warning("Could not ensure scanner_lock table: %s", e) + return True # fail-open: keep scanning rather than block entirely + + now = datetime.now(timezone.utc).replace(tzinfo=None) + + # 1) Try to insert a fresh lock row. + try: + self.db.execute( + """INSERT INTO scanner_lock (lock_name, owner, locked_at, lease_seconds) + VALUES (%s, %s, %s, %s)""", + (self._lock_name, self._instance_id, now, self._lock_lease_seconds), + transaction=True, + ) + logger.info("Acquired scanner lock (fresh). owner=%s", self._instance_id) + return True + except Exception: + # Row already exists -> fall through to steal-if-stale. + pass + + # 2) Take over the lock if it is stale or already ours. + stale_before = now - timedelta(seconds=self._lock_lease_seconds) + try: + affected = self.db.execute( + """UPDATE scanner_lock + SET owner = %s, locked_at = %s + WHERE lock_name = %s + AND (owner = %s OR locked_at < %s)""", + (self._instance_id, now, self._lock_name, self._instance_id, stale_before), + transaction=True, + ) + if affected and affected > 0: + logger.info("Acquired scanner lock (stale takeover). owner=%s", + self._instance_id) + return True + except Exception as e: + logger.warning("Failed to check scanner lock: %s", e) + return True # fail-open + + logger.info("Scanner lock held by another instance; not acquiring.") + return False + + def release_lock(self): + """Release the scanner lock if we own it.""" + try: + self.db.execute( + """DELETE FROM scanner_lock WHERE lock_name = %s AND owner = %s""", + (self._lock_name, self._instance_id), + transaction=True, + ) + except Exception as e: + logger.warning("Failed to release scanner lock: %s", e) + + def _heartbeat(self): + """Refresh the lock lease so a long-running scan is not stolen.""" + try: + now = datetime.now(timezone.utc).replace(tzinfo=None) + self.db.execute( + """UPDATE scanner_lock SET locked_at = %s + WHERE lock_name = %s AND owner = %s""", + (now, self._lock_name, self._instance_id), + transaction=True, + ) + except Exception as e: + logger.debug("Scanner lock heartbeat failed: %s", e) + def _run_scan(self): """Execute a single scan cycle.""" scan_start = time.time() @@ -181,6 +319,7 @@ class DirectoryScanner: def _process_files_batch(self, files: List[Path]) -> List[dict]: """Process a batch of files in parallel.""" results = [] + processed = 0 with ThreadPoolExecutor(max_workers=self.walker_threads) as executor: future_to_file = { @@ -201,6 +340,11 @@ class DirectoryScanner: "error_message": str(e), }) + processed += 1 + # Keep the single-instance lock alive during long scans. + if self._heartbeat_every and processed % self._heartbeat_every == 0: + self._heartbeat() + return results def _process_single_file(self, file_path: Path) -> dict: @@ -384,6 +528,10 @@ class DirectoryScanner: "total_files_new": self._total_files_new, "total_files_modified": self._total_files_modified, "total_files_unscannable": self._total_files_unscannable, + "instance_id": self._instance_id, + "scan_interval_seconds": self.scan_interval, + "lock_lease_seconds": self._lock_lease_seconds, + "scan_in_progress": self._scan_lock.locked(), "input_path": str(self.input_path), "is_running": self._running, }