diff --git a/kustomization.yaml b/kustomization.yaml index 195142b..53039f9 100644 --- a/kustomization.yaml +++ b/kustomization.yaml @@ -1,8 +1,9 @@ configMapGenerator: - name: env-injector-files - namespace: downstream-llm-d + namespace: llm-d-thibrahi files: - sitecustomize.py - profiler_config.yaml + - metrics_observer.py generatorOptions: disableNameSuffixHash: true diff --git a/metrics_observer.py b/metrics_observer.py new file mode 100644 index 0000000..f1847f3 --- /dev/null +++ b/metrics_observer.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +""" +Prometheus metrics observer for vLLM saturation detection. + +Runs as a sidecar container alongside vLLM. Polls the /metrics endpoint, +tracks generation token throughput, and writes a signal file when throughput +plateaus (indicating the system is saturated and in steady state). + +The signal file triggers profiling in sitecustomize.py. + +Configuration via environment variables (see defaults below). +No external dependencies — stdlib only. +""" + +import json +import os +import sys +import time +import urllib.request +import urllib.error +from datetime import datetime, timezone + +PORT = int(os.environ.get("METRICS_OBSERVER_PORT", "8000")) +POLL_INTERVAL = float(os.environ.get("METRICS_OBSERVER_POLL_INTERVAL", "5")) +WINDOW_SIZE = int(os.environ.get("METRICS_OBSERVER_WINDOW_SIZE", "6")) +CV_THRESHOLD = float(os.environ.get("METRICS_OBSERVER_CV_THRESHOLD", "0.10")) +MIN_THROUGHPUT = float(os.environ.get("METRICS_OBSERVER_MIN_THROUGHPUT", "1.0")) +SIGNAL_FILE = os.environ.get("METRICS_OBSERVER_SIGNAL_FILE", "/tmp/profiler_start") +OUTPUT_FILE = os.environ.get("METRICS_OBSERVER_OUTPUT", "/tmp/metrics_observer.json") +WARMUP = float(os.environ.get("METRICS_OBSERVER_WARMUP", "30")) + +METRICS_URL = f"http://localhost:{PORT}/metrics" + +TRACKED_COUNTERS = ["vllm:generation_tokens", "vllm:prompt_tokens"] +TRACKED_GAUGES = [ + "vllm:num_requests_running", + "vllm:num_requests_waiting", + "vllm:kv_cache_usage_perc", +] + + +def log(msg): + print(f"[metrics-observer] {msg}", file=sys.stderr, flush=True) + + +def parse_metric(text, name): + """Extract first matching metric value from Prometheus text format.""" + for line in text.split("\n"): + if line.startswith("#"): + continue + if not line.startswith(name): + continue + # name or name{labels} value + parts = line.split() + if len(parts) >= 2: + try: + return float(parts[-1]) + except ValueError: + continue + return None + + +def fetch_metrics(): + """Fetch and parse relevant metrics from vLLM.""" + try: + req = urllib.request.Request(METRICS_URL, method="GET") + with urllib.request.urlopen(req, timeout=5) as resp: + text = resp.read().decode("utf-8") + except (urllib.error.URLError, OSError): + return None + + result = {} + for name in TRACKED_COUNTERS + TRACKED_GAUGES: + val = parse_metric(text, name) + if val is not None: + result[name] = val + return result if result else None + + +def detect_plateau(throughput_history): + """Return True if throughput has plateaued (low coefficient of variation).""" + if len(throughput_history) < WINDOW_SIZE: + return False + recent = throughput_history[-WINDOW_SIZE:] + if all(t < MIN_THROUGHPUT for t in recent): + return False + mean = sum(recent) / len(recent) + if mean < MIN_THROUGHPUT: + return False + variance = sum((t - mean) ** 2 for t in recent) / len(recent) + cv = (variance ** 0.5) / mean + return cv < CV_THRESHOLD + + +def main(): + log(f"Starting: port={PORT} poll={POLL_INTERVAL}s window={WINDOW_SIZE} " + f"cv_thresh={CV_THRESHOLD} min_tput={MIN_THROUGHPUT} warmup={WARMUP}s") + log(f"Signal file: {SIGNAL_FILE}") + log(f"Metrics URL: {METRICS_URL}") + + if os.path.exists(SIGNAL_FILE): + os.remove(SIGNAL_FILE) + log("Removed stale signal file") + + log(f"Waiting {WARMUP}s for vLLM warmup...") + time.sleep(WARMUP) + + samples = [] + throughput_history = [] + prev_gen_tokens = None + prev_time = None + signal_sent = False + + config = { + "port": PORT, + "poll_interval": POLL_INTERVAL, + "window_size": WINDOW_SIZE, + "cv_threshold": CV_THRESHOLD, + "min_throughput": MIN_THROUGHPUT, + "signal_file": SIGNAL_FILE, + "warmup": WARMUP, + } + + while True: + metrics = fetch_metrics() + now = time.time() + + if metrics is None: + log("Failed to fetch metrics, retrying...") + time.sleep(POLL_INTERVAL) + continue + + gen_tokens = metrics.get("vllm:generation_tokens", 0) + throughput = 0.0 + + if prev_gen_tokens is not None and prev_time is not None: + dt = now - prev_time + if dt > 0: + throughput = (gen_tokens - prev_gen_tokens) / dt + + prev_gen_tokens = gen_tokens + prev_time = now + + if throughput > 0 or throughput_history: + throughput_history.append(throughput) + + sample = { + "timestamp": now, + "generation_tokens": gen_tokens, + "prompt_tokens": metrics.get("vllm:prompt_tokens", 0), + "throughput_tok_s": round(throughput, 2), + "num_requests_running": metrics.get("vllm:num_requests_running", 0), + "num_requests_waiting": metrics.get("vllm:num_requests_waiting", 0), + "kv_cache_usage_perc": metrics.get("vllm:kv_cache_usage_perc", 0), + } + samples.append(sample) + + running = sample["num_requests_running"] + waiting = sample["num_requests_waiting"] + log(f"tput={throughput:.1f} tok/s running={running} " + f"waiting={waiting} samples={len(throughput_history)}/{WINDOW_SIZE}") + + if not signal_sent and detect_plateau(throughput_history): + log(f"Throughput plateau detected! " + f"mean={sum(throughput_history[-WINDOW_SIZE:])/WINDOW_SIZE:.1f} tok/s") + + with open(SIGNAL_FILE, "w") as f: + f.write(json.dumps({ + "triggered_at": datetime.now(timezone.utc).isoformat(), + "throughput_tok_s": round(throughput, 2), + "mean_throughput": round( + sum(throughput_history[-WINDOW_SIZE:]) / WINDOW_SIZE, 2 + ), + "samples_collected": len(samples), + })) + log(f"Signal file written: {SIGNAL_FILE}") + signal_sent = True + + output = { + "config": config, + "signal_triggered_at": datetime.now(timezone.utc).isoformat(), + "throughput_at_trigger": round(throughput, 2), + "samples": samples, + } + try: + with open(OUTPUT_FILE, "w") as f: + json.dump(output, f, indent=2) + log(f"Metrics dump written: {OUTPUT_FILE}") + except Exception as e: + log(f"Failed to write output: {e}") + + log("Observer done. Exiting.") + break + + time.sleep(POLL_INTERVAL) + + +if __name__ == "__main__": + main() diff --git a/sitecustomize.py b/sitecustomize.py index 8bfa665..2cc1dbb 100644 --- a/sitecustomize.py +++ b/sitecustomize.py @@ -51,6 +51,9 @@ def __init__(self): self.target_class: str = "Worker" self.target_method: str = "execute_model" self.debug: bool = False + self.signal_mode: bool = False + self.signal_file: str = "/tmp/profiler_start" + self.profile_duration: int = 50 self._load_config() @@ -160,6 +163,14 @@ def _load_from_env(self): if 'VLLM_PROFILER_DEBUG' in os.environ: self.debug = os.environ['VLLM_PROFILER_DEBUG'].lower() in ('true', '1', 'yes') + # Signal mode + if 'VLLM_PROFILER_SIGNAL_MODE' in os.environ: + self.signal_mode = os.environ['VLLM_PROFILER_SIGNAL_MODE'].lower() in ('true', '1', 'yes') + if 'VLLM_PROFILER_SIGNAL_FILE' in os.environ: + self.signal_file = os.environ['VLLM_PROFILER_SIGNAL_FILE'] + if 'VLLM_PROFILER_DURATION' in os.environ: + self.profile_duration = int(os.environ['VLLM_PROFILER_DURATION']) + def _parse_ranges(self, ranges_str: str) -> List[Tuple[int, int]]: """ Parse profiling ranges from string format. @@ -199,7 +210,86 @@ def get_output_filename(self, pid: Optional[int] = None, rank: Optional[int] = N # ============================================================================== -# Import Hook +# NIXL Handshake Recorder +# ============================================================================== + +_NIXL_CONNECTOR_MODULE = "vllm.distributed.kv_transfer.kv_connector.v1.nixl_connector" + +class NixlHandshakeRecorder: + """Records per-call timing of NixlConnector._nixl_handshake.""" + + def __init__(self): + self.records = [] + self.output_path = os.environ.get( + 'VLLM_PROFILER_NIXL_HANDSHAKE_OUTPUT', + '/tmp/nixl_handshake_timings.json' + ) + self.enabled = os.environ.get( + 'VLLM_PROFILER_NIXL_HANDSHAKE', '' + ).lower() in ('true', '1', 'yes') + + def wrap(self, original_func): + import functools + import time + recorder = self + + @functools.wraps(original_func) + def wrapped(self_connector, host, port, remote_tp_size, expected_engine_id): + start = time.monotonic() + error_msg = None + result = None + try: + result = original_func( + self_connector, host, port, + remote_tp_size, expected_engine_id + ) + return result + except Exception as e: + error_msg = str(e) + raise + finally: + elapsed = time.monotonic() - start + record = { + "timestamp": time.time(), + "pid": os.getpid(), + "host": host, + "port": port, + "remote_tp_size": remote_tp_size, + "expected_engine_id": expected_engine_id, + "duration_s": round(elapsed, 6), + "success": error_msg is None, + "error": error_msg, + "remote_agents": ( + list(result.values()) if result else None + ), + } + recorder.records.append(record) + recorder._flush() + print( + f"[nixl-handshake] {host}:{port} " + f"tp={remote_tp_size} " + f"engine={expected_engine_id[:12]}... " + f"dur={elapsed:.3f}s " + f"{'OK' if error_msg is None else 'FAIL'}", + file=sys.stderr + ) + + return wrapped + + def _flush(self): + import json + try: + with open(self.output_path, 'w') as f: + json.dump(self.records, f, indent=2) + except Exception as e: + print(f"[nixl-handshake] flush error: {e}", file=sys.stderr) + + +_nixl_recorder = NixlHandshakeRecorder() + + +# ============================================================================== +# Import Hooks # ============================================================================== class PostImportLoader(importlib.abc.Loader): @@ -238,25 +328,104 @@ def find_spec(self, fullname, path, target=None): return None -# Install the import hook +class NixlPostImportLoader(importlib.abc.Loader): + def __init__(self, loader, recorder): + self.loader = loader + self.recorder = recorder + + def create_module(self, spec): + if hasattr(self.loader, "create_module"): + return self.loader.create_module(spec) + return None + + def exec_module(self, module): + self.loader.exec_module(module) + nixl_cls = getattr(module, "NixlConnector", None) + if nixl_cls is None: + print(f"[nixl-handshake] NixlConnector not found in {module.__name__}", file=sys.stderr) + return + original = getattr(nixl_cls, "_nixl_handshake", None) + if original is None: + print(f"[nixl-handshake] _nixl_handshake not found on NixlConnector", file=sys.stderr) + return + setattr(nixl_cls, "_nixl_handshake", self.recorder.wrap(original)) + print(f"[nixl-handshake] Wrapped NixlConnector._nixl_handshake", file=sys.stderr) + + +class NixlPostImportFinder(importlib.abc.MetaPathFinder): + def __init__(self, recorder): + self.recorder = recorder + + def find_spec(self, fullname, path, target=None): + if fullname != _NIXL_CONNECTOR_MODULE: + return None + + sys.meta_path.remove(self) + try: + spec = importlib.util.find_spec(fullname) + finally: + sys.meta_path.insert(0, self) + + if spec and spec.loader: + spec.loader = NixlPostImportLoader(spec.loader, self.recorder) + return spec + return None + + +# Install import hooks sys.meta_path.insert(0, PostImportFinder()) +if _nixl_recorder.enabled: + sys.meta_path.insert(0, NixlPostImportFinder(_nixl_recorder)) # ============================================================================== # Profiler Wrapper # ============================================================================== +def _make_profiler(activities): + """Create a new torch.profiler.profile instance.""" + from torch.profiler import profile + return profile( + activities=activities, + record_shapes=_config.record_shapes, + with_stack=_config.with_stack, + profile_memory=_config.profile_memory, + with_modules=_config.with_modules + ) + + +def _stop_and_export(prof, start, end): + """Stop profiler, print stats, export trace.""" + prof.stop() + + if _config.print_stats: + print("===== begin profiler output") + if _config.table_enabled: + print(prof.key_averages().table( + sort_by=_config.table_sort_by, + row_limit=_config.table_row_limit + )) + print("===== end profiler output") + + if _config.export_chrome_trace: + output_file = _config.get_output_filename(range_start=start, range_end=end) + prof.export_chrome_trace(output_file) + print(f"[profiler] Exported trace to: {output_file}") + else: + print(f"[profiler] Chrome trace export disabled (export_chrome_trace=false)") + + def wrap_func_with_profiler(original_func): """ - Wraps a function with PyTorch profiler that activates for configured ranges. + Wraps a function with PyTorch profiler. - Supports multiple profiling windows, e.g., calls 50-100 and 200-300. + Two modes: + - Range mode (default): profiles fixed call ranges (e.g., 100-150) + - Signal mode: waits for signal file, then profiles for N calls """ - import torch import functools - from torch.profiler import profile, ProfilerActivity + from torch.profiler import ProfilerActivity - # Parse activities activities = [] for activity in _config.activities: if activity.upper() == "CPU": @@ -264,74 +433,61 @@ def wrap_func_with_profiler(original_func): elif activity.upper() == "CUDA": activities.append(ProfilerActivity.CUDA) - # Create profiler instance - prof = profile( - activities=activities, - record_shapes=_config.record_shapes, - with_stack=_config.with_stack, - profile_memory=_config.profile_memory, - with_modules=_config.with_modules - ) - - # Track call count and current profiling range index + prof = _make_profiler(activities) count = 0 current_range_idx = 0 profiling_active = False + signal_consumed = False + signal_start = 0 + signal_end = 0 @functools.wraps(original_func) def wrapped_func(*args, **kwargs): nonlocal count, current_range_idx, profiling_active, prof + nonlocal signal_consumed, signal_start, signal_end count += 1 - # Check if we should start profiling - if not profiling_active and current_range_idx < len(_config.ranges): - start, end = _config.ranges[current_range_idx] - if count == start: - print(f"[profiler] Starting profiler for range {start}-{end} (call #{count})") - prof.start() - profiling_active = True - - # Check if we should stop profiling - if profiling_active: - start, end = _config.ranges[current_range_idx] - if count == end: - print(f"[profiler] Stopping profiler for range {start}-{end} (call #{count})") - prof.stop() + if _config.signal_mode: + # Signal mode: wait for signal file to start profiling + if not profiling_active and not signal_consumed: + if os.path.exists(_config.signal_file): + signal_start = count + signal_end = count + _config.profile_duration + print(f"[profiler] Signal received! Starting profiler for {_config.profile_duration} calls " + f"(calls {signal_start}-{signal_end}, call #{count})") + prof.start() + profiling_active = True + + if profiling_active and count >= signal_end: + print(f"[profiler] Stopping profiler (call #{count}, range {signal_start}-{signal_end})") + _stop_and_export(prof, signal_start, signal_end) profiling_active = False + signal_consumed = True + try: + os.remove(_config.signal_file) + except OSError: + pass + else: + # Range mode: original behavior + if not profiling_active and current_range_idx < len(_config.ranges): + start, end = _config.ranges[current_range_idx] + if count == start: + print(f"[profiler] Starting profiler for range {start}-{end} (call #{count})") + prof.start() + profiling_active = True + + if profiling_active: + start, end = _config.ranges[current_range_idx] + if count == end: + print(f"[profiler] Stopping profiler for range {start}-{end} (call #{count})") + _stop_and_export(prof, start, end) + profiling_active = False + current_range_idx += 1 + + if current_range_idx < len(_config.ranges): + prof = _make_profiler(activities) - # Print and export results - if _config.print_stats: - print("===== begin profiler output") - if _config.table_enabled: - print(prof.key_averages().table( - sort_by=_config.table_sort_by, - row_limit=_config.table_row_limit - )) - print("===== end profiler output") - - # Optionally export Chrome trace file - if _config.export_chrome_trace: - output_file = _config.get_output_filename(range_start=start, range_end=end) - prof.export_chrome_trace(output_file) - print(f"[profiler] Exported trace to: {output_file}") - else: - print(f"[profiler] Chrome trace export disabled (export_chrome_trace=false)") - - # Move to next range - current_range_idx += 1 - - # Create new profiler for next range if exists - if current_range_idx < len(_config.ranges): - prof = profile( - activities=activities, - record_shapes=_config.record_shapes, - with_stack=_config.with_stack, - profile_memory=_config.profile_memory, - with_modules=_config.with_modules - ) - - # Call original function result = original_func(*args, **kwargs) return result @@ -385,5 +541,10 @@ def unwrap_function(): # Startup # ============================================================================== -print(f"[profiler] vLLM profiler installed - will profile ranges: {_config.ranges}", file=sys.stderr) +if _config.signal_mode: + print(f"[profiler] vLLM profiler installed - signal mode: waiting for {_config.signal_file} (duration={_config.profile_duration} calls)", file=sys.stderr) +else: + print(f"[profiler] vLLM profiler installed - will profile ranges: {_config.ranges}", file=sys.stderr) print(f"[profiler] Target: {_config.target_module}.{_config.target_class}.{_config.target_method}", file=sys.stderr) +if _nixl_recorder.enabled: + print(f"[profiler] NIXL handshake recorder enabled → {_nixl_recorder.output_path}", file=sys.stderr) diff --git a/webhook.py b/webhook.py index 106a18a..083055a 100644 --- a/webhook.py +++ b/webhook.py @@ -32,8 +32,13 @@ FILE_KEYS = [ {"key": "sitecustomize.py", "mountPath": "/home/vllm/profiler/sitecustomize.py"}, {"key": "profiler_config.yaml", "mountPath": "/home/vllm/profiler/profiler_config.yaml"}, + {"key": "metrics_observer.py", "mountPath": "/home/vllm/profiler/metrics_observer.py"}, ] +OBSERVER_SIDECAR_IMAGE = os.getenv("OBSERVER_SIDECAR_IMAGE", "python:3.11-slim") +OBSERVER_SHARED_VOLUME_NAME = "profiler-signal" +OBSERVER_SHARED_MOUNT_PATH = "/tmp" + # Profiler configuration annotation prefix PROFILER_ANNOTATION_PREFIX = "vllm.profiler/" @@ -97,6 +102,11 @@ def extract_profiler_env_from_annotations(annotations: Dict[str, str]) -> List[D "output": "VLLM_PROFILER_OUTPUT", "export-trace": "VLLM_PROFILER_EXPORT_TRACE", "debug": "VLLM_PROFILER_DEBUG", + "nixl-handshake": "VLLM_PROFILER_NIXL_HANDSHAKE", + "nixl-handshake-output": "VLLM_PROFILER_NIXL_HANDSHAKE_OUTPUT", + "signal-mode": "VLLM_PROFILER_SIGNAL_MODE", + "signal-file": "VLLM_PROFILER_SIGNAL_FILE", + "duration": "VLLM_PROFILER_DURATION", } for annotation_suffix, env_name in annotation_to_env.items(): @@ -251,6 +261,111 @@ def build_files_volume_patch_for_pod(pod: Dict[str, Any]) -> List[Dict[str, Any] return patch +def extract_observer_env_from_annotations(annotations: Dict[str, str]) -> List[Dict[str, str]]: + """Extract observer-specific env vars from pod annotations.""" + observer_annotation_to_env = { + "observer-poll-interval": "METRICS_OBSERVER_POLL_INTERVAL", + "observer-window-size": "METRICS_OBSERVER_WINDOW_SIZE", + "observer-cv-threshold": "METRICS_OBSERVER_CV_THRESHOLD", + "observer-min-throughput": "METRICS_OBSERVER_MIN_THROUGHPUT", + "observer-warmup": "METRICS_OBSERVER_WARMUP", + "observer-port": "METRICS_OBSERVER_PORT", + } + env_vars = [] + for suffix, env_name in observer_annotation_to_env.items(): + key = f"{PROFILER_ANNOTATION_PREFIX}{suffix}" + if key in annotations: + env_vars.append({"name": env_name, "value": annotations[key]}) + return env_vars + + +def detect_vllm_port(pod: Dict[str, Any]) -> str: + """Auto-detect vLLM metrics port from pod spec.""" + for c in pod.get("spec", {}).get("containers", []): + if c.get("name") == "vllm": + for p in c.get("ports", []): + if p.get("name") == "vllm": + return str(p.get("containerPort", 8000)) + return "8000" + + +def build_observer_sidecar_patch(pod: Dict[str, Any], annotations: Dict[str, str]) -> List[Dict[str, Any]]: + """ + Build JSON patch to inject metrics-observer sidecar container and shared /tmp volume. + """ + patch: List[Dict[str, Any]] = [] + spec = pod.get("spec", {}) or {} + containers = spec.get("containers", []) + volumes = spec.get("volumes", []) + + # Check if observer already injected + if any(c.get("name") == "metrics-observer" for c in containers): + logger.debug("metrics-observer sidecar already present; skipping") + return patch + + # Shared /tmp volume + shared_vol_present = any(v.get("name") == OBSERVER_SHARED_VOLUME_NAME for v in volumes) + if not shared_vol_present: + vol_entry = {"name": OBSERVER_SHARED_VOLUME_NAME, "emptyDir": {}} + if volumes: + patch.append({"op": "add", "path": "/spec/volumes/-", "value": vol_entry}) + else: + patch.append({"op": "add", "path": "/spec/volumes", "value": [vol_entry]}) + + # Mount shared /tmp into existing vllm container(s) + for idx, c in enumerate(containers): + if c.get("name") != "vllm": + continue + mounts = c.get("volumeMounts", []) + already_mounted = any( + m.get("name") == OBSERVER_SHARED_VOLUME_NAME for m in mounts + ) + if not already_mounted: + mount = { + "name": OBSERVER_SHARED_VOLUME_NAME, + "mountPath": OBSERVER_SHARED_MOUNT_PATH, + } + if mounts: + patch.append({"op": "add", "path": f"/spec/containers/{idx}/volumeMounts/-", "value": mount}) + else: + patch.append({"op": "add", "path": f"/spec/containers/{idx}/volumeMounts", "value": [mount]}) + + # Build observer env vars + observer_env = extract_observer_env_from_annotations(annotations) + vllm_port = detect_vllm_port(pod) + # Auto-set port if not overridden by annotation + if not any(e["name"] == "METRICS_OBSERVER_PORT" for e in observer_env): + observer_env.append({"name": "METRICS_OBSERVER_PORT", "value": vllm_port}) + + # Build sidecar container + sidecar = { + "name": "metrics-observer", + "image": OBSERVER_SIDECAR_IMAGE, + "command": ["python", "/home/vllm/profiler/metrics_observer.py"], + "env": observer_env, + "volumeMounts": [ + { + "name": OBSERVER_SHARED_VOLUME_NAME, + "mountPath": OBSERVER_SHARED_MOUNT_PATH, + }, + { + "name": FILES_VOLUME_NAME, + "mountPath": "/home/vllm/profiler/metrics_observer.py", + "subPath": "metrics_observer.py", + "readOnly": True, + }, + ], + "resources": { + "requests": {"cpu": "50m", "memory": "64Mi"}, + "limits": {"cpu": "200m", "memory": "128Mi"}, + }, + } + + patch.append({"op": "add", "path": "/spec/containers/-", "value": sidecar}) + logger.debug("Observer sidecar patch prepared with port=%s", vllm_port) + return patch + + @app.route("/healthz", methods=["GET"]) def healthz(): return "ok", 200 @@ -348,6 +463,12 @@ def mutate(): # Mount configuration files patch_ops.extend(build_files_volume_patch_for_pod(obj)) + # Inject observer sidecar if annotation present + observer_annotation = f"{PROFILER_ANNOTATION_PREFIX}observer" + if annotations.get(observer_annotation, "").lower() in ("true", "1", "yes"): + logger.debug("Observer sidecar requested via annotation") + patch_ops.extend(build_observer_sidecar_patch(obj, annotations)) + if patch_ops: logger.debug("Emitting JSONPatch with %d operation(s)", len(patch_ops)) patch_str = json.dumps(patch_ops)