backing up - add parser for timestamps

Author: rerpha

pyproject.toml

@@ -8,6 +8,7 @@ dynamic = ["version"]
 dependencies = [
     "ess-streaming-data-types",
     "confluent-kafka",
+    "python-dateutil",
     "tzdata"
 ]
 readme = {file = "README.md", content-type = "text/markdown"}

src/saluki/main.py

@@ -2,6 +2,7 @@
 import logging
 import sys
 
+from dateutil.parser import parse, ParserError
 from saluki.consume import consume
 from saluki.listen import listen
 from saluki.play import play
@@ -17,6 +18,16 @@
 _SNIFF = "sniff"
 
 
+def _dateutil_parsable_or_unix_timestamp(inp: str) -> float:
+    try:
+        try:
+            return parse(inp).timestamp()
+        except ParserError:
+            return float(inp)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"timestamp {inp} is not parsable by dateutil.parse() and is not a unix timestamp")
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(
         prog="saluki",
@@ -47,7 +58,7 @@ def main() -> None:
     sub_parsers = parser.add_subparsers(help="sub-command help", required=True, dest="command")
 
     sniff_parser = sub_parsers.add_parser(_SNIFF, help="sniff - broker metadata")
-    sniff_parser.add_argument("broker", type=str)
+    sniff_parser.add_argument("broker", type=str, help="broker, optionally suffixed with a topic name")
 
     consumer_parser = argparse.ArgumentParser(add_help=False)
     consumer_parser.add_argument(
@@ -94,9 +105,7 @@ def main() -> None:
         type=int,
         nargs=2,
     )
-    g.add_argument(
-        "-t", "--timestamps", help="unix timestamps to replay between", type=str, nargs=2
-    )
+    g.add_argument("-t", "--timestamps", help='timestamps to replay between in ISO8601 or RFC3339 format ie. "2025-11-17 07:00:00"  ', type=_dateutil_parsable_or_unix_timestamp, nargs=2)
 
     if len(sys.argv) == 1:
         parser.print_help()
@@ -130,7 +139,11 @@ def main() -> None:
             args.timestamps,
         )
     elif args.command == _SNIFF:
-        sniff(args.broker)
+        try:
+            broker, topic = parse_kafka_uri(args.broker)
+            sniff(broker, topic)
+        except RuntimeError:
+            sniff(args.broker)
 
 
 if __name__ == "__main__":

src/saluki/sniff.py

@@ -7,22 +7,31 @@
 logger = logging.getLogger("saluki")
 
 
-def sniff(broker: str) -> None:
+def sniff(broker: str, topic: str | None = None) -> None:
     """
     Prints the broker and topic metadata for a given broker.
+    If a topic is given, only this topic's partitions and watermarks will be printed.
     :param broker: The broker address including port number.
+    :param topic: Optional topic to filter information to.
     """
     a = AdminClient({"bootstrap.servers": broker})
     c = Consumer({"bootstrap.servers": broker, "group.id": f"saluki-sniff-{uuid.uuid4()}"})
     t = a.list_topics(timeout=5)
-    logger.info(f"Cluster ID: {t.cluster_id}")
-    logger.info("Brokers:")
-    for value in t.brokers.values():
-        logger.info(f"\t{value}")
+    if topic is not None and topic not in t.topics.keys():
+        logger.warning(f"Topic {topic} not found on broker {broker}")
+        return
 
-    logger.info("Topics:")
+    if topic is None:
+        logger.info(f"Cluster ID: {t.cluster_id}")
+        logger.info("Brokers:")
+        for value in t.brokers.values():
+            logger.info(f"\t{value}")
+
+        logger.info("Topics:")
 
     for k, v in t.topics.items():
+        if topic is not None and k != topic:
+            continue
         partitions = v.partitions.keys()
         logger.info(f"\t{k}:")
         for p in partitions:

src/saluki/utils.py

@@ -79,8 +79,8 @@ def parse_kafka_uri(uri: str) -> Tuple[str, str]:
     If username is provided, a SASL mechanism must also be provided.
     Any other validation must be performed in the calling code.
     """
-    broker, topic = uri.split("/") if "/" in uri else (uri, "")
-    if not topic:
+    broker, topic = uri.split("/") if "/" in uri else (uri, None)
+    if topic is None:
         raise RuntimeError(
             f"Unable to parse URI {uri}, topic not defined. URI should be of form"
             f" broker[:port]/topic"

tests/test_sniff.py

@@ -1,29 +1,38 @@
+import pytest
 from unittest.mock import patch
 
 from confluent_kafka.admin import BrokerMetadata, ClusterMetadata, TopicMetadata
 
 from saluki.sniff import sniff
 
+@pytest.fixture()
+def fake_cluster_md():
+    """
+    Returns a fake cluster metadata object with two topics;
+    one with 1 partition and the other with 2.
+    """
+    fake_cluster_md = ClusterMetadata()
+    broker1 = BrokerMetadata()
+    broker1.id = "id1"  # type: ignore
+    broker1.host = "mybroker"  # type: ignore
+    broker1.port = 9093
+    fake_cluster_md.brokers = {0: broker1}
 
-def test_sniff_with_two_partitions_in_a_topic():
+    topic1 = TopicMetadata()
+    topic1.partitions = {0: {}}
+
+    topic2 = TopicMetadata()
+    topic2.partitions = {0: {}, 1: {}}
+
+    fake_cluster_md.topics = {"topic1": topic1, "topic2": topic2}
+    return fake_cluster_md
+
+def test_sniff_with_two_partitions_in_a_topic(fake_cluster_md):
     with (
         patch("saluki.sniff.AdminClient") as a,
         patch("saluki.sniff.Consumer") as c,
         patch("saluki.sniff.logger") as logger,
     ):
-        fake_cluster_md = ClusterMetadata()
-        broker1 = BrokerMetadata()
-        broker1.id = "id1"  # type: ignore
-        broker1.host = "mybroker"  # type: ignore
-        broker1.port = 9093
-        fake_cluster_md.brokers = {0: broker1}
-
-        topic1 = TopicMetadata()
-        topic1.partitions = {0: {}}
-        topic2 = TopicMetadata()
-        topic2.partitions = {0: {}, 1: {}}
-
-        fake_cluster_md.topics = {"topic1": topic1, "topic2": topic2}
         a().list_topics.return_value = fake_cluster_md
         c().get_watermark_offsets.return_value = 1, 2
         sniff("whatever")
@@ -40,3 +49,29 @@ def test_sniff_with_two_partitions_in_a_topic():
 
         topic2_call2 = logger.info.call_args_list[8]
         assert "1 - low:1, high:2, num_messages:1" in topic2_call2.args[0]
+
+def test_sniff_with_single_topic(fake_cluster_md):
+    with (
+        patch("saluki.sniff.AdminClient") as a,
+        patch("saluki.sniff.Consumer") as c,
+        patch("saluki.sniff.logger") as logger,
+    ):
+
+        a().list_topics.return_value = fake_cluster_md
+        c().get_watermark_offsets.return_value = 1, 2
+        sniff("mybroker:9093", "topic1")
+
+        assert "\ttopic1" in logger.info.call_args_list[0].args[0]
+        assert "\t\t0 - low:1, high:2, num_messages:1" in logger.info.call_args_list[1].args[0]
+
+
+def test_sniff_with_single_nonexistent_topic():
+    with (
+        patch("saluki.sniff.AdminClient") as a,
+        patch("saluki.sniff.Consumer"),
+        patch("saluki.sniff.logger") as logger,
+    ):
+        # Deliberately blank cluster metadata ie. no topics
+        a().list_topics.return_value = ClusterMetadata()
+        sniff("somebroker:9092", "sometopic")
+        logger.warning.assert_called_with("Topic sometopic not found on broker somebroker:9092")