121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
import time
|
|
import uuid
|
|
import jaeger_client
|
|
import opentracing
|
|
from opentracing.ext import tags
|
|
from opentracing.propagation import Format
|
|
|
|
import logging
|
|
import sys
|
|
import pika
|
|
import json
|
|
|
|
from sentry_sdk.integrations.logging import LoggingIntegration
|
|
import sentry_sdk
|
|
|
|
from config import Config
|
|
|
|
from feeder import Feeder
|
|
|
|
|
|
def message_callback_factory(feeder: Feeder):
|
|
def message_callback(channel, method, properties, body):
|
|
try:
|
|
msg = json.loads(body.decode('utf-8'))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as e:
|
|
logging.warning(f"Invalid message received: {e}")
|
|
channel.basic_ack(delivery_tag=method.delivery_tag) # We don't want this to be requeue
|
|
return
|
|
|
|
logging.debug(f"Handling message: {msg}")
|
|
|
|
if 'tag' not in msg:
|
|
logging.warning(f"Invalid message received: No tag field")
|
|
channel.basic_ack(delivery_tag=method.delivery_tag) # We don't want this to be requeue
|
|
return
|
|
tag = msg['tag']
|
|
mime_type = msg.get('mime_type', 'audio/wave')
|
|
|
|
span_ctx = opentracing.tracer.extract(Format.TEXT_MAP, msg)
|
|
span_tags = {
|
|
tags.SPAN_KIND: tags.SPAN_KIND_CONSUMER,
|
|
'sample_tag': tag
|
|
}
|
|
|
|
with opentracing.tracer.start_active_span(
|
|
'main.handleMessage', finish_on_close=True, child_of=span_ctx, tags=span_tags
|
|
) as scope:
|
|
try:
|
|
feeder.do_feed(tag, mime_type)
|
|
except Exception as e:
|
|
logging.error(f"Something went wrong during handling sample {tag} run: {e}; msg: {msg}")
|
|
logging.exception(e)
|
|
channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
|
|
else:
|
|
# successful
|
|
channel.basic_ack(delivery_tag=method.delivery_tag)
|
|
|
|
return message_callback
|
|
|
|
|
|
def main():
|
|
# setup logging
|
|
|
|
logging.basicConfig(
|
|
stream=sys.stdout,
|
|
format="%(asctime)s - %(name)s [%(levelname)s]: %(message)s",
|
|
level=Config.LOG_LEVEL
|
|
)
|
|
|
|
# setup observability stuffs
|
|
|
|
if Config.SENTRY_DSN:
|
|
sentry_logging = LoggingIntegration(
|
|
level=logging.DEBUG, # Capture info and above as breadcrumbs
|
|
event_level=logging.ERROR # Send errors as events
|
|
)
|
|
sentry_sdk.init(
|
|
dsn=Config.SENTRY_DSN,
|
|
integrations=[sentry_logging],
|
|
traces_sample_rate=1.0,
|
|
send_default_pii=True,
|
|
release=Config.RELEASE_ID,
|
|
environment=Config.RELEASEMODE,
|
|
_experiments={"auto_enabling_integrations": True}
|
|
)
|
|
|
|
jaeger_client.Config(config={}, service_name='storage-service-feeder', validate=True).initialize_tracer()
|
|
|
|
# Start the memes
|
|
logging.info("Preparing feeder...")
|
|
feeder = Feeder()
|
|
|
|
logging.info("Connecting to MQ service...")
|
|
connection = pika.BlockingConnection(pika.connection.URLParameters(Config.PIKA_URL))
|
|
channel = connection.channel()
|
|
channel.exchange_declare(exchange=Config.PIKA_SAMPLE_CACHE_EXCHANGE, exchange_type='direct')
|
|
|
|
queue_declare_result = channel.queue_declare(queue=str(uuid.uuid4()), exclusive=False)
|
|
queue_name = queue_declare_result.method.queue
|
|
|
|
channel.queue_bind(exchange=Config.PIKA_SAMPLE_CACHE_EXCHANGE, routing_key='cache', queue=queue_name)
|
|
|
|
channel.basic_qos(prefetch_count=1)
|
|
|
|
channel.basic_consume(queue=queue_name, on_message_callback=message_callback_factory(feeder), auto_ack=False)
|
|
|
|
logging.info("Connection complete! Listening to messages...")
|
|
try:
|
|
channel.start_consuming()
|
|
except KeyboardInterrupt:
|
|
logging.info("SIGINT Received! Stopping stuff...")
|
|
channel.stop_consuming()
|
|
|
|
time.sleep(2) # lol
|
|
opentracing.tracer.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|