Redo s3 stuff (#53)

gcn_monitor/cli.py

@@ -40,7 +40,8 @@ def host_port(host_port_str):
     show_default=True,
     help="Log level",
 )
-def main(prometheus, loglevel):
+@click.option("--bucket-name", help="Bucket Name")
+def main(prometheus, loglevel, bucket_name):
     """Monitor connectivity of a Kafka client.
 
     Specify the Kafka client configuration in environment variables using the
@@ -60,4 +61,4 @@ def main(prometheus, loglevel):
     )
     log.info("Prometheus listening on %s", prometheus.netloc)
 
-    kafka.run()
+    kafka.run(bucket_name)

gcn_monitor/kafka.py

@@ -9,6 +9,7 @@
 
 import json
 import logging
+from base64 import b64encode
 
 import boto3
 import gcn_kafka
@@ -63,7 +64,7 @@ def parse_filenames(message):
     return file_name, message_key_file_name, headers_file_name
 
 
-def run():
+def run(bucket_name):
     log.info("Creating consumer")
     config = gcn_kafka.config_from_env()
     config["stats_cb"] = stats_cb
@@ -78,6 +79,31 @@ def run():
     while True:
         for message in consumer.consume(timeout=1):
             topic = message.topic()
+            file_name, message_key_file_name, headers_file_name = parse_filenames(
+                message
+            )
+            s3_client.put_object(
+                Bucket=bucket_name,
+                Key=file_name,
+                Body=message.value(),
+            )
+
+            if message_key_file_name is not None:
+                s3_client.put_object(
+                    Bucket=bucket_name,
+                    Key=message_key_file_name,
+                    Body=message.key(),
+                )
+
+            if headers_file_name is not None:
+                s3_client.put_object(
+                    Bucket=bucket_name,
+                    Key=headers_file_name,
+                    Body={
+                        key: b64encode(value).decode()
+                        for key, value in message.headers()
+                    },
+                )
             if error := message.error():
                 log.error("topic %s: got error %s", topic, error)
             else: