swf-monitor/scripts/enqueue-ops-message.py

0001 #!/usr/bin/env python3
0002 """Send one message to the epicprod ops queue — the cron-friendly trigger.
0003
0004 The cron side of scheduled automation is deliberately trivial: this script
0005 does nothing but enqueue a message for the prod-ops agent, which performs,
0006 logs, and times the actual work (see docs/EPICPROD_OPS_AGENT.md, action-stream
0007 logging). Uses the same ACTIVEMQ_* environment the agents use (source ~/.env).
0008
0009 Usage:
0010     enqueue-ops-message.py catalog_sync --created-by nightly_cron
0011     enqueue-ops-message.py association_sweep --created-by backfill --extra days=30
0012 """
0013
0014 import argparse
0015 import json
0016 import os
0017 import ssl
0018 import sys
0019
0020 import stomp
0021
0022 OPS_QUEUE = os.environ.get("EPICPROD_OPS_QUEUE", "/queue/epicprod.ops")
0023
0024
0025 def main():
0026     parser = argparse.ArgumentParser(description=__doc__)
0027     parser.add_argument('msg_type', help="ops message type, e.g. catalog_sync")
0028     parser.add_argument('--created-by', default='cron',
0029                         help="requester recorded in the action stream")
0030     parser.add_argument('--extra', action='append', default=[],
0031                         help="extra key=value message fields (int if numeric)")
0032     args = parser.parse_args()
0033
0034     msg = {'msg_type': args.msg_type, 'namespace': 'prodops',
0035            'created_by': args.created_by}
0036     for item in args.extra:
0037         key, _, value = item.partition('=')
0038         if not key or not value:
0039             sys.exit(f"bad --extra {item!r}, need key=value")
0040         msg[key] = int(value) if value.isdigit() else value
0041
0042     host = os.getenv('ACTIVEMQ_HOST', 'localhost')
0043     port = int(os.getenv('ACTIVEMQ_PORT', '61612'))
0044     conn = stomp.Connection(host_and_ports=[(host, port)], vhost=host,
0045                             try_loopback_connect=False)
0046     if os.getenv('ACTIVEMQ_USE_SSL', 'False').lower() == 'true':
0047         conn.transport.set_ssl(
0048             for_hosts=[(host, port)],
0049             ca_certs=os.getenv('ACTIVEMQ_SSL_CA_CERTS') or None,
0050             ssl_version=ssl.PROTOCOL_TLS_CLIENT,
0051         )
0052     conn.connect(os.getenv('ACTIVEMQ_USER', 'admin'),
0053                  os.getenv('ACTIVEMQ_PASSWORD', 'admin'), wait=True)
0054     try:
0055         conn.send(destination=OPS_QUEUE, body=json.dumps(msg))
0056     finally:
0057         conn.disconnect()
0058     print(f"enqueued {args.msg_type} to {OPS_QUEUE}")
0059
0060
0061 if __name__ == '__main__':
0062     main()