127 lines
3.5 KiB
Python
Executable file
127 lines
3.5 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from common import DEFAULT_NAMESPACE, build_env, kafka_brokers, raw_near_topic, stderr
|
|
|
|
|
|
DEFAULT_OUTPUT_FORMAT = (
|
|
"%d{go[2006-01-02T15:04:05Z07:00]} topic=%t partition=%p offset=%o\n"
|
|
"%v\n"
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Live-inspect raw NEAR intents arriving in Redpanda."
|
|
)
|
|
parser.add_argument(
|
|
"--namespace",
|
|
default=DEFAULT_NAMESPACE,
|
|
help=f"Kubernetes namespace to inspect (default: {DEFAULT_NAMESPACE})",
|
|
)
|
|
parser.add_argument(
|
|
"--brokers",
|
|
default="",
|
|
help="Override Kafka brokers instead of reading them from unrip-config.",
|
|
)
|
|
parser.add_argument(
|
|
"--topic",
|
|
default="",
|
|
help="Topic to consume. Defaults to KAFKA_TOPIC_RAW_NEAR_INTENTS_QUOTE from unrip-config.",
|
|
)
|
|
parser.add_argument(
|
|
"--offset",
|
|
default="end",
|
|
help="Kafka offset selector passed to rpk (default: end)",
|
|
)
|
|
parser.add_argument(
|
|
"--num",
|
|
type=int,
|
|
default=0,
|
|
help="Number of records to read before exiting. 0 means unbounded (default: 0).",
|
|
)
|
|
parser.add_argument(
|
|
"--timeout",
|
|
type=float,
|
|
default=0,
|
|
help="Stop after this many seconds even if the stream is still open.",
|
|
)
|
|
parser.add_argument(
|
|
"--value-only",
|
|
action="store_true",
|
|
help="Print only the record value without topic/partition/offset metadata.",
|
|
)
|
|
parser.add_argument(
|
|
"--rpk-json",
|
|
action="store_true",
|
|
help="Use rpk's built-in JSON record formatter instead of the default metadata+value output.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def build_command(args: argparse.Namespace, brokers: str, topic: str) -> list[str]:
|
|
command = [
|
|
"kubectl",
|
|
"exec",
|
|
"-i",
|
|
"-n",
|
|
args.namespace,
|
|
"deploy/redpanda",
|
|
"--",
|
|
"rpk",
|
|
"topic",
|
|
"consume",
|
|
topic,
|
|
"--brokers",
|
|
brokers,
|
|
"--offset",
|
|
args.offset,
|
|
"--num",
|
|
str(args.num),
|
|
]
|
|
|
|
if args.rpk_json:
|
|
command.extend(["--format", "json", "--pretty-print=false"])
|
|
elif args.value_only:
|
|
command.extend(["--format", "%v\n"])
|
|
else:
|
|
command.extend(["--format", DEFAULT_OUTPUT_FORMAT])
|
|
|
|
return command
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
brokers = args.brokers or kafka_brokers(namespace=args.namespace)
|
|
topic = args.topic or raw_near_topic(namespace=args.namespace)
|
|
command = build_command(args, brokers, topic)
|
|
|
|
stderr(f"namespace={args.namespace} brokers={brokers} topic={topic} offset={args.offset} num={args.num}")
|
|
|
|
process = subprocess.Popen(command, env=build_env())
|
|
try:
|
|
process.wait(timeout=args.timeout if args.timeout and args.timeout > 0 else None)
|
|
except subprocess.TimeoutExpired:
|
|
process.send_signal(signal.SIGINT)
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=5)
|
|
return 124
|
|
except KeyboardInterrupt:
|
|
process.send_signal(signal.SIGINT)
|
|
return process.wait()
|
|
return process.returncode
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|