backing up

Author: rerpha

README.md

@@ -1,19 +1,42 @@
 ![](https://github.com/ISISComputingGroup/saluki/blob/main/resources/logo.png)
 
-Serialise/deserialise flatbuffers blobs from kafka. 
-This currently deserialises https://github.com/ess-dmsc/python-streaming-data-types, but I am working to make it agnostic. Python bindings for the respective schema will need to be generated. 
+ISIS-specific Kafka tools.
+Deserialises [the ESS flatbuffers blobs](https://github.com/ess-dmsc/python-streaming-data-types) from Kafka. 
+
+Also allows replaying data in a topic. 
 
 # Usage
 See `saluki --help` for all options. 
 
 ## Listen to a topic for updates
 `saluki listen mybroker:9092/mytopic` - This will listen for updates for `mytopic` on `mybroker`. 
 
+### Filter to specific schemas
+
+TODO
+
 ## Consume from a topic
 `saluki consume mybroker:9092/mytopic -p 1 -o 123456 -m 10` - This will print 9 messages before (and inclusively the offset specified) offset `123456` of `mytopic` on `mybroker`, in partition 1.
 
 Use the `-g` flag to go the other way, ie. in the above example to consume the 9 messages _after_ offset 123456
 
+### Consume X of a certain schema(s)
+
+TODO
+
+## List all topics and their high, low watermarks and number of messages
+TODO
+
+## Replay data from one topic to another
+
+### Between offsets
+
+TODO
+
+### Between timestamps 
+
+TODO
+
 # Install 
 `pip install saluki`
 

src/saluki/listen.py

@@ -7,7 +7,7 @@
 logger = logging.getLogger("saluki")
 
 
-def listen(broker: str, topic: str, partition: int | None = None) -> None:
+def listen(broker: str, topic: str, partition: int | None = None, filter: list[str] | None = None) -> None:
     """
     Listen to a topic and deserialise each message
     :param broker: the broker address, including the port

src/saluki/main.py

@@ -11,8 +11,10 @@
 
 _LISTEN = "listen"
 _CONSUME = "consume"
-_REPLAY = "replay"
-
+_PLAY = "play"
+_SNIFF = "sniff"
+_BURY = "bury"
+_DIG = "dig"
 
 def main() -> None:
     parser = argparse.ArgumentParser(
@@ -21,6 +23,8 @@ def main() -> None:
     )
 
     parent_parser = argparse.ArgumentParser(add_help=False)
+    #TODO this needs restructuring. consider having a topic_parser with partition options etc. 
+    # because -m and -g does not make sense for saluki listen. 
     parent_parser.add_argument("topic", type=str, help="Kafka topic. format is broker<:port>/topic")
 
     parent_parser.add_argument(
@@ -41,6 +45,9 @@ 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)
+
     consumer_parser = argparse.ArgumentParser(add_help=False)
     consumer_parser.add_argument(
         "-e",
@@ -67,37 +74,44 @@ def main() -> None:
     consumer_mode_parser.add_argument("-s", "--schema", required=False, default="auto", type=str)
     consumer_mode_parser.add_argument("-g", "--go-forwards", required=False, action="store_true")
     consumer_mode_parser.add_argument("-p", "--partition", required=False, type=int, default=0)
-    # TODO make this allow multiple comma-split args
-    consumer_mode_parser.add_argument("-f", "--filter", required=False, type=str, nargs='+')
+    consumer_mode_parser.add_argument("-f", "--filter", required=False, action="append")
 
     listen_parser = sub_parsers.add_parser(
         _LISTEN,
         help="listen mode - listen until KeyboardInterrupt",
         parents=[parent_parser, consumer_parser],
     )
     listen_parser.add_argument("-p", "--partition", required=False, type=int, default=None)
-    # TODO make filtering work for this as well
 
     #### NEW FEATURES HERE PLZ
     # replay from, to offset
-    # saluki replay -o FROMOFFSET TOOFFSET srcbroker/srctopic destbroker/desttopic
+    # saluki play -o FROMOFFSET TOOFFSET srcbroker/srctopic destbroker/desttopic
 
     # replay from, to timestamp
-    # saluki replay -t FROMTIMESTAMP TOTIMESTAMP srcbroker/srctopic destbroker/desttopic
+    # saluki play -t FROMTIMESTAMP TOTIMESTAMP srcbroker/srctopic destbroker/desttopic
 
     # saluki consume x messages of y schema
     # saluki consume -f pl72 mybroker:9092/XXX_runInfo -m 10 # get the last pl72 run starts
 
     # saluki consume x messages of y or z schema
-    # saluki consume -f pl72,6s4t mybroker:9092/XXX_runInfo -m 10 # get the last pl72 run starts or 6s4t run stops
+    # saluki consume -f pl72,6s4t mybroker:9092/XXX_runInfo -m 10 # get the last pl72 run starts or 6s4t run stops#
+
+    # saluki bury - dump data on topic to file
+    # saluki bury mybroker:9092/topicname -p 0 -f offsetortimestamp -t offsetortimestamp outputfile
+
+    # saluki dig - push data from dump generated by saluki bury to topic 
+    # saluki dig mybroker:9092/topicname -p 0 outputfile
 
-    replay_parser = sub_parsers.add_parser(
-        _REPLAY,
+    # saluki sniff - broker metadata ie. topic watermarks and num_messages. 
+    # saluki sniff mybroker:9092 
+
+    play_parser = sub_parsers.add_parser(
+        _PLAY,
         help="replay mode - replay data into another topic",
         parents=[parent_parser],
     )
-    replay_parser.add_argument("-o", "--offset", help="replay between offsets", type=bool, required=False, default="store_false")
-    replay_parser.add_argument("-t", "--timestamp", help="replay between timestamps", type=bool, required=False)
+    play_parser.add_argument("-o", "--offset", help="offsets to replay between (inclusive)", type=int, nargs=2)
+    play_parser.add_argument("-t", "--timestamp", help="timestamps to replay between", type=str, nargs=2)
 
 
     if len(sys.argv) == 1:
@@ -114,7 +128,7 @@ def main() -> None:
         logger.addHandler(logging.FileHandler(args.log_file.name))
 
     if args.command == _LISTEN:
-        listen(broker, topic, args.partition)
+        listen(broker, topic, args.partition, args.filter)
     elif args.command == _CONSUME:
         consume(
             broker,
@@ -124,7 +138,15 @@ def main() -> None:
             args.offset,
             args.go_forwards,
         )
-    elif args.command == _REPLAY:
+    elif args.command == _PLAY:
+        pass
+        #play(src_broker, src_topic, dest_broker, dest_topic, args.offset, args.timestamp)
+    elif args.command == _SNIFF:
+        print(args.broker)
+        pass
+    elif args.command == _BURY:
+        pass
+    elif args.command == _DIG:
         pass
 
 if __name__ == "__main__":

src/saluki/utils.py

@@ -31,7 +31,7 @@ def fallback_deserialiser(payload: bytes) -> str:
     return schema, ret
 
 
-def deserialise_and_print_messages(msgs: List[Message], partition: int | None) -> None:
+def deserialise_and_print_messages(msgs: List[Message], partition: int | None, filter: list[str] | None) -> None:
     for msg in msgs:
         try:
             if msg is None:
@@ -42,6 +42,9 @@ def deserialise_and_print_messages(msgs: List[Message], partition: int | None) -
             if partition is not None and msg.partition() != partition:
                 continue
             schema, deserialised = _try_to_deserialise_message(msg.value())
+            if filter is not None and schema not in filter:
+                # ignore
+                break
             time = _parse_timestamp(msg)
             logger.info(f"{msg.offset()} ({time}):({schema}) {deserialised}")
         except Exception as e: