"""FRAMERESULT-01: 저장과 독립된 최신 snapshot 송신. 생성만으로 연결/기동하지 않는다.""" from collections import OrderedDict, Counter from dataclasses import dataclass, field import hashlib import json import math import re import threading import time from app.publisher_metrics import PublisherMetrics @dataclass(frozen=True) class MetadataSettings: url: str = field(repr=False) namespace: str = "contextra:metadata:v1" max_channels: int = 64 max_bytes: int = 4 * 1024 * 1024 max_message_bytes: int = 128 * 1024 max_age: float = 1.0 io_timeout: float = 0.25 stream_length: int = 100 stream_ttl: int = 10 workers: int = 1 worker_mode: str = 'fixed' active_worker_limit: int = 16 def __post_init__(self): if not isinstance(self.url, str) or not self.url: raise ValueError("Metadata connection must be explicitly configured.") if not re.fullmatch(r"[a-zA-Z0-9:_-]{1,80}", self.namespace): raise ValueError("Invalid metadata namespace.") for v in (self.max_channels, self.max_bytes, self.max_message_bytes, self.stream_length, self.stream_ttl): if type(v) is not int or v <= 0: raise ValueError("Invalid metadata limit.") for v in (self.max_age, self.io_timeout): if not math.isfinite(v) or v <= 0: raise ValueError("Invalid metadata timeout.") if type(self.workers) is not int or not 1 <= self.workers <= 8: raise ValueError('Publisher workers must be between one and eight.') if self.worker_mode not in ('fixed', 'channel'): raise ValueError('Invalid publisher worker mode.') if type(self.active_worker_limit) is not int or not 1 <= self.active_worker_limit <= 16: raise ValueError('Active publisher worker limit must be between one and sixteen.') def camera_key(camera_id): if type(camera_id) not in (str, int) or not str(camera_id) or len(str(camera_id)) > 256: raise ValueError("Invalid camera reference.") return hashlib.sha256(str(camera_id).encode()).hexdigest() class RedisSnapshotSink: """명시 worker에서만 접속하며 기존 내부 Redis client와 공유하지 않는다.""" def __init__(self, settings, client=None): self.settings = settings self.client = client self.runs = {} def _connect(self): if self.client is None: import redis from redis.backoff import NoBackoff from redis.retry import Retry self.client = redis.from_url( self.settings.url, decode_responses=True, socket_connect_timeout=self.settings.io_timeout, socket_timeout=self.settings.io_timeout, max_connections=1, retry=Retry(NoBackoff(), 0), retry_on_timeout=False, ) def _commands(self, pipe, key, status, payload): stream = f"{self.settings.namespace}:camera:{key}:frames" if self.runs.get(key) != status["runId"] or status["state"] != "RUNNING": pipe.delete(stream) if payload is not None: pipe.xadd(stream, {"json": payload}, maxlen=self.settings.stream_length, approximate=False) pipe.expire(stream, self.settings.stream_ttl) pipe.hset(f"{self.settings.namespace}:channels", key, stream) pipe.set(f"{self.settings.namespace}:camera:{key}:status", json.dumps(dict(status, streamKey=stream), allow_nan=False), ex=self.settings.stream_ttl) pipe.expire(f"{self.settings.namespace}:channels", self.settings.stream_ttl) def write(self, key, status, payload): self._connect() pipe = self.client.pipeline(transaction=True) self._commands(pipe, key, status, payload) pipe.execute() self.runs[key] = status["runId"] def close(self): if self.client is not None: self.client.close() class MetadataPublisher: """채널당 최신 1개, 전체 byte cap. 채널 순환과 네트워크 I/O lock 분리를 유지한다.""" def __init__(self, settings, sink=None, clock=time.monotonic, *, sink_factory=None): self.settings = settings if sink is not None and (sink_factory is not None or settings.workers != 1): raise ValueError('Each publisher worker requires its own sink.') self._sinks = ([sink] if sink is not None else [(sink_factory(i) if sink_factory else RedisSnapshotSink(settings)) for i in range(settings.workers if settings.worker_mode == 'fixed' else 1)]) self.sink = self._sinks[0] self.clock = clock self._cv = threading.Condition() self._channels = {} self._pending = OrderedDict() self._provider_sequences = {} self._delivery = {} self._relay_timing = {} self._frame_timing = OrderedDict() self._bytes = 0 self._closed = False self._thread = None self._threads = [] self._lanes = {} self._metrics = Counter() self._heartbeat_at = self.clock() self._profile = PublisherMetrics(clock) def start(self): with self._cv: if self._closed: raise RuntimeError("Publisher is closed.") if self._thread is None: self._threads = [threading.Thread(target=self._run, args=(i,), name='metadata-publisher-' + str(i), daemon=True) for i in range(self.settings.workers)] self._thread = self._threads[0] for thread in self._threads: thread.start() def channel(self, camera_id, run_id, generation, state): key = camera_key(camera_id) if state not in {"STARTING", "RUNNING", "STOPPED", "ERROR"}: raise ValueError("Invalid channel state.") unstarted = state == 'STOPPED' and generation == 0 and run_id is None if not unstarted and (type(generation) is not int or generation < 1 or not isinstance(run_id, str) or not run_id): raise ValueError("Invalid run identity.") with self._cv: if self._closed: return False old = self._channels.get(key) if old and (generation < old["runGeneration"] or (generation == old["runGeneration"] and run_id != old["runId"])): return False if old and generation == old["runGeneration"] and old["state"] in {"STOPPED", "ERROR"} and state in {"RUNNING", "STARTING"}: return False if old is None and len(self._channels) >= self.settings.max_channels: self._metrics["channel_limit"] += 1 return False # Repeated RUNNING notifications precede every Frame; keep pending Frame data. if old and generation == old['runGeneration'] and run_id == old['runId'] and state == old['state']: if not self._same_state(key, state): return False self._profile.count(key, 'state_noop') return True if not self._admit(key, old, run_id, generation, state): return False seq = old["lastSeq"] if old and generation == old["runGeneration"] else -1 self._channels[key] = dict(cameraId=camera_id, runId=run_id, runGeneration=generation, state=state, lastSeq=seq) if old is None or generation != old['runGeneration'] or state != 'RUNNING': self._provider_sequences.pop(key, None) if old is None or generation != old['runGeneration']: self._delivery[key] = {} for pending in list(self._pending): if isinstance(pending, tuple) and pending[0] == key: self._bytes -= self._pending.pop(pending)[2] self._queue(key, None, 'state') return True def _admit(self, key, old, run_id, generation, state): if old is None: self._lanes[key] = len(self._channels) % self.settings.workers return True def _same_state(self, key, state): return True def wait_channel(self, camera_id, run_id, generation, timeout): return True def _queue(self, key, payload, reason='frame', sequence=None, provider=None): pending_key = (key, provider) if provider else key self._profile.count(key, 'queued_' + reason) previous = self._pending.get(pending_key) if previous: self._bytes -= previous[2] size = len(payload.encode()) if payload is not None else 0 if self._bytes + size > self.settings.max_bytes: # 상태 갱신은 남기고 이번 Frame만 버린다. self._pending[pending_key] = (self.clock(), None, 0) self._metrics["byte_limit"] += 1 self._cv.notify_all() return False if previous and previous[1] is not None: self._metrics["replaced"] += 1 self._profile.count(key, 'replaced_by_' + reason) self._profile.enqueue(key, sequence, reason, previous, self._lanes[key]) self._pending[pending_key] = (self.clock(), payload, size) self._bytes += size self._cv.notify_all() return True def note_frame_timing(self, frame, selected_ns, completed_ns): if self._profile.token() is None: return if not (type(selected_ns) is int and type(completed_ns) is int and 0 <= selected_ns <= completed_ns): raise ValueError('Invalid private Frame timing.') key = (camera_key(frame.source_id), frame.run_id, frame.run_generation, frame.frame_seq) with self._cv: if self._profile.token() is None: return self._frame_timing[key] = dict(selected_ns=selected_ns, completed_ns=completed_ns) while len(self._frame_timing) > 64: self._frame_timing.popitem(last=False) def submit(self, snapshot): payload = json.dumps(snapshot, ensure_ascii=False, allow_nan=False, separators=(",", ":")) key = camera_key(snapshot["cameraId"]) provider = snapshot.get('providerKey') if provider is not None and provider not in ('medium', 'pose', 'fire', 'crowd'): return False with self._cv: state = self._channels.get(key) if self._closed or not state or state["state"] != "RUNNING": self._metrics["inactive"] += 1 return False sequences = self._provider_sequences.setdefault(key, {}) last = sequences.get(provider, -1) if provider else state['lastSeq'] if (snapshot["runId"] != state["runId"] or snapshot["runGeneration"] != state["runGeneration"] or type(snapshot["frameSeq"]) is not int or snapshot["frameSeq"] <= last): self._metrics["stale_generation_or_seq"] += 1 return False if len(payload.encode()) > self.settings.max_message_bytes: self._metrics["message_limit"] += 1 return False accepted = self._queue(key, payload, sequence=snapshot['frameSeq'], provider=provider) sequences[provider] = snapshot['frameSeq'] state["lastSeq"] = max(state['lastSeq'], snapshot["frameSeq"]) return accepted def note_relay(self, provider, elapsed_ns): with self._cv: if self._profile.token() is None: return row = self._relay_timing.setdefault(provider, dict(count=0, sum_ns=0, max_ns=0)) row['count'] += 1 row['sum_ns'] += max(0, elapsed_ns) row['max_ns'] = max(row['max_ns'], elapsed_ns) def _run(self, lane=0): sink = self._sinks[lane] try: while True: with self._cv: self._cv.wait_for(lambda: self._closed or any(self._lanes[k[0] if isinstance(k, tuple) else k] == lane for k in self._pending), self.settings.stream_ttl / 3) if self._closed: return if lane == 0 and self.clock() - self._heartbeat_at >= self.settings.stream_ttl / 3: for channel in self._channels: if channel not in self._pending: self._queue(channel, None, 'heartbeat') self._heartbeat_at = self.clock() pending_key = next((k for k in self._pending if self._lanes[k[0] if isinstance(k, tuple) else k] == lane), None) if pending_key is None: continue key = pending_key[0] if isinstance(pending_key, tuple) else pending_key at, payload, size = self._pending.pop(pending_key) self._bytes -= size status = dict(self._channels[key]) status["publisher"] = dict(self._metrics) if payload is not None and self.clock() - at > self.settings.max_age: self._metrics["expired"] += 1 self._profile.count(key, 'expired') payload = None token = self._profile.token() kind = 'frame' if payload is not None else 'status' self._profile.duration(token, key, kind + '_queue', self.clock() - at) trace = self._profile.dequeue(key, kind, lane) if trace is not None and kind == 'frame': identity = (key, status['runId'], status['runGeneration'], trace.get('sequence')) trace.update(self._frame_timing.pop(identity, {})) if token is not None: write_wall, write_cpu = time.monotonic_ns(), time.thread_time_ns() succeeded = False try: sink.write(key, status, payload) succeeded = True except Exception: pass if token is not None: finished, cpu = time.monotonic_ns(), time.thread_time_ns() - write_cpu with self._cv: if token is not None: accounted = time.monotonic_ns() if succeeded: self._metrics["sent"] += 1 self._metrics['sent_frames' if payload is not None else 'sent_status'] += 1 if payload is not None: provider = pending_key[1] if isinstance(pending_key, tuple) else 'single' now = time.monotonic_ns() row = self._delivery.setdefault(key, {}).setdefault(provider, dict(count=0, first_at_ns=now, first_wall_ms=time.time_ns()//1_000_000, last_at_ns=now, interval_count=0, interval_sum_ns=0, max_interval_ns=0, histogram=[0]*11)) if row['count']: gap=now-row['last_at_ns'];row['interval_count']+=1 row['interval_sum_ns']+=gap;row['max_interval_ns']=max(row['max_interval_ns'],gap) bounds=(20,50,75,100,150,200,300,500,1000,2000) bucket=next((i for i,b in enumerate(bounds) if gap<=b*1_000_000),10) row['histogram'][bucket]+=1 row['count']+=1;row['last_at_ns']=now if token is not None and token == self._profile.token(): self._profile.count(key, 'sent_' + kind) else: self._metrics["send_error"] += 1 if token is not None and token == self._profile.token(): self._profile.count(key, 'send_error') if token is not None: if trace is not None: trace['write_succeeded'] = succeeded if succeeded and kind == 'frame' and trace is not None: for origin in ('selected', 'completed'): at_ns = trace.get(origin+'_ns') if at_ns is not None: self._profile.duration(token, key, origin+'_to_redis', (finished-at_ns)/1e9) self._profile.duration(token, key, kind + '_write_wall', (finished-write_wall)/1e9) self._profile.duration(token, key, kind + '_write_cpu', cpu/1e9) self._profile.duration(token, key, kind + '_completion_lock', (accounted-finished)/1e9) self._profile.complete(token, trace, write_wall, finished, cpu, accounted) finally: try: sink.close() except Exception: with self._cv: self._metrics["close_error"] += 1 def metrics(self, include_timeline=False): with self._cv: return dict(self._metrics, queued=len(self._pending), queuedBytes=self._bytes, relay_timing={k:dict(v) for k,v in self._relay_timing.items()}, channels=len(self._channels), workers=self.settings.workers, delivery={k:{p:dict(v,histogram=list(v['histogram'])) for p,v in ps.items()} for k,ps in self._delivery.items()}, profile=self._profile.snapshot(include_timeline)) def configure_metrics(self, enabled, seconds=30): with self._cv: if self._closed: raise RuntimeError('Publisher is closed.') self._profile.configure(enabled, seconds) self._relay_timing.clear() self._frame_timing.clear() return self._profile.snapshot() def close(self, timeout=1.0): if not math.isfinite(timeout) or timeout < 0: raise ValueError("Invalid shutdown timeout.") with self._cv: self._closed = True self._metrics["shutdown_drop"] += len(self._pending) self._pending.clear() self._bytes = 0 self._cv.notify_all() threads = tuple(self._threads) deadline = time.monotonic() + timeout for thread in threads: thread.join(max(0., deadline - time.monotonic())) with self._cv: return not any(t.is_alive() for t in threads) and not self._metrics["close_error"]