From 81d8da625b19749c257541b25b42ccd55a2d0b28 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 26 Nov 2025 13:53:14 +0000 Subject: [PATCH] Queue dumper CLI - might be useful for debug --- trustgraph-cli/pyproject.toml | 1 + trustgraph-cli/trustgraph/cli/dump_queues.py | 216 +++++++++++++++++++ 2 files changed, 217 insertions(+) create mode 100644 trustgraph-cli/trustgraph/cli/dump_queues.py diff --git a/trustgraph-cli/pyproject.toml b/trustgraph-cli/pyproject.toml index 9eaf60cd..9e1226d1 100644 --- a/trustgraph-cli/pyproject.toml +++ b/trustgraph-cli/pyproject.toml @@ -34,6 +34,7 @@ tg-delete-mcp-tool = "trustgraph.cli.delete_mcp_tool:main" tg-delete-kg-core = "trustgraph.cli.delete_kg_core:main" tg-delete-tool = "trustgraph.cli.delete_tool:main" tg-dump-msgpack = "trustgraph.cli.dump_msgpack:main" +tg-dump-queues = "trustgraph.cli.dump_queues:main" tg-get-flow-class = "trustgraph.cli.get_flow_class:main" tg-get-kg-core = "trustgraph.cli.get_kg_core:main" tg-graph-to-turtle = "trustgraph.cli.graph_to_turtle:main" diff --git a/trustgraph-cli/trustgraph/cli/dump_queues.py b/trustgraph-cli/trustgraph/cli/dump_queues.py new file mode 100644 index 00000000..43a7fc6a --- /dev/null +++ b/trustgraph-cli/trustgraph/cli/dump_queues.py @@ -0,0 +1,216 @@ +""" +Multi-queue Pulsar message dumper for debugging TrustGraph message flows. + +This utility monitors multiple Pulsar queues simultaneously and logs all messages +to a file with timestamps and pretty-printed formatting. Useful for debugging +message flows, diagnosing stuck services, and understanding system behavior. +""" + +import pulsar +import sys +import json +from datetime import datetime +import argparse + +def format_message(queue_name, msg): + """Format a message with timestamp and queue name.""" + timestamp = datetime.now().isoformat() + + # Try to parse as JSON and pretty-print + try: + value = msg.value() + + # If it's bytes, decode it + if isinstance(value, bytes): + value = value.decode('utf-8') + + # If it's a string, try to parse as JSON + if isinstance(value, str): + try: + parsed = json.loads(value) + body = json.dumps(parsed, indent=2) + except (json.JSONDecodeError, TypeError): + body = value + else: + # Try to convert to dict for pretty printing + try: + # Pulsar schema objects have __dict__ or similar + if hasattr(value, '__dict__'): + parsed = {k: v for k, v in value.__dict__.items() + if not k.startswith('_')} + else: + parsed = str(value) + body = json.dumps(parsed, indent=2, default=str) + except (TypeError, AttributeError): + body = str(value) + + except Exception as e: + body = f"\n{msg.value()}" + + # Format the output + header = f"\n{'='*80}\n[{timestamp}] Queue: {queue_name}\n{'='*80}\n" + return header + body + "\n" + +def main(): + parser = argparse.ArgumentParser( + prog='tg-dump-queues', + description='Monitor and dump messages from multiple Pulsar queues', + epilog=""" +Examples: + # Monitor agent and prompt queues + tg-dump-queues non-persistent://tg/request/agent:default \\ + non-persistent://tg/request/prompt:default + + # Monitor with custom output file + tg-dump-queues non-persistent://tg/request/agent:default \\ + --output debug.log + + # Append to existing log file + tg-dump-queues non-persistent://tg/request/agent:default \\ + --output queue.log --append + +Common queue patterns: + - Agent requests: non-persistent://tg/request/agent:default + - Agent responses: non-persistent://tg/response/agent:default + - Prompt requests: non-persistent://tg/request/prompt:default + - Prompt responses: non-persistent://tg/response/prompt:default + - LLM requests: non-persistent://tg/request/text-completion:default + - LLM responses: non-persistent://tg/response/text-completion:default + """, + formatter_class=argparse.RawDescriptionHelpFormatter + ) + + parser.add_argument( + 'queues', + nargs='+', + help='Pulsar queue names to monitor' + ) + + parser.add_argument( + '--output', '-o', + default='queue.log', + help='Output file (default: queue.log)' + ) + + parser.add_argument( + '--append', '-a', + action='store_true', + help='Append to output file instead of overwriting' + ) + + parser.add_argument( + '--pulsar-host', + default='pulsar://localhost:6650', + help='Pulsar host URL (default: pulsar://localhost:6650)' + ) + + parser.add_argument( + '--listener-name', + default='localhost', + help='Pulsar listener name (default: localhost)' + ) + + parser.add_argument( + '--subscriber', + default='debug', + help='Subscriber name for queue subscription (default: debug)' + ) + + args = parser.parse_args() + + # Filter out any accidentally included flags + queues = [q for q in args.queues if not q.startswith('--')] + + if not queues: + parser.error("No queues specified") + + print(f"TrustGraph Queue Dumper") + print(f"Monitoring {len(queues)} queue(s):") + for q in queues: + print(f" - {q}") + print(f"Output file: {args.output}") + print(f"Mode: {'append' if args.append else 'overwrite'}") + print(f"Press Ctrl+C to stop\n") + + # Connect to Pulsar + try: + client = pulsar.Client(args.pulsar_host, listener_name=args.listener_name) + except Exception as e: + print(f"Error connecting to Pulsar at {args.pulsar_host}: {e}", file=sys.stderr) + sys.exit(1) + + # Subscribe to all queues + consumers = [] + for queue in queues: + try: + consumer = client.subscribe( + queue, + args.subscriber, + consumer_type=pulsar.ConsumerType.Shared + ) + consumers.append((queue, consumer)) + print(f"✓ Subscribed to: {queue}") + except Exception as e: + print(f"✗ Error subscribing to {queue}: {e}", file=sys.stderr) + + if not consumers: + print("\nNo consumers created. Exiting.", file=sys.stderr) + client.close() + sys.exit(1) + + print(f"\nListening for messages...\n") + + # Open output file (overwrite by default, append if --append specified) + mode = 'a' if args.append else 'w' + try: + with open(args.output, mode) as f: + f.write(f"\n{'#'*80}\n") + f.write(f"# Session started: {datetime.now().isoformat()}\n") + f.write(f"# Monitoring queues: {', '.join(queues)}\n") + f.write(f"{'#'*80}\n") + f.flush() + + try: + # Process messages from all consumers + while True: + for queue_name, consumer in consumers: + try: + # Non-blocking receive with timeout + msg = consumer.receive(timeout_millis=100) + consumer.acknowledge(msg) + + # Format and write + output = format_message(queue_name, msg) + f.write(output) + f.flush() + + # Also print to console (truncated) + print(f"[{datetime.now().strftime('%H:%M:%S')}] {queue_name}: Message received") + + except Exception as e: + # Timeout is normal, just continue silently + if "TimeOut" not in str(e) and "Timeout" not in str(e): + error_msg = f"Error receiving from {queue_name}: {e}\n" + f.write(error_msg) + f.flush() + print(error_msg, file=sys.stderr) + + except KeyboardInterrupt: + print("\n\nStopping...") + f.write(f"\n{'#'*80}\n") + f.write(f"# Session ended: {datetime.now().isoformat()}\n") + f.write(f"{'#'*80}\n") + + except IOError as e: + print(f"Error writing to {args.output}: {e}", file=sys.stderr) + sys.exit(1) + + # Cleanup + for _, consumer in consumers: + consumer.close() + client.close() + + print(f"\nMessages logged to: {args.output}") + +if __name__ == '__main__': + main()