RabbitMQ recovery scripts

rdqdump2

#!/usr/bin/env python3

import struct
import sys

# Directly inspired by Jeff Bryner's rdqdump:
# https://github.com/jeffbryner/rdqdump
#
# This does essentially the same thing, but works with the binary data directly.

MARKER = bytes.fromhex("395f316c000000016d0000")

def read_record(data, start):
    pos = data.find(MARKER, start)
    if pos >= 0:
        len_pos = pos + len(MARKER)
        (payload_len,) = struct.unpack("!H", data[len_pos:len_pos+2])
        start_pos = len_pos+2
        end_pos = start_pos+payload_len
        payload = data[start_pos:end_pos]
        return payload, end_pos
    else:
        return None, -1

def extract_all(data):
    pos = 0
    while pos >= 0:
        payload, pos = read_record(data, pos)
        if payload:
            print(payload.hex())
            #print("read {} bytes, next record at {}".format(len(payload), pos))

data = None
with open(sys.argv[1], "rb") as fh:
    data = fh.read()
    extract_all(data)

restore_msgs

#!/usr/bin/env python3

import os
import sys

import msgpack
import pika

class Sender:
    def __init__(self, conn, queue):
        self.queue = queue
        self.channel = conn.channel()
        self.props = pika.BasicProperties(delivery_mode=2)

    def send(self, raw):
        self.channel.basic_publish(exchange='',
                                   routing_key=self.queue,
                                   body=raw,
                                   properties=self.props)

def validate(raw):
    try:
        msgpack.unpackb(raw)
        return True
    except:
        return False

queue = os.environ['QUEUE']
params = pika.URLParameters(os.environ['RABBITMQ'])
conn = pika.BlockingConnection(params)
sender = Sender(conn, queue)

n_sent = 0
n_invalid = 0

for line in sys.stdin:
    raw = bytes.fromhex(line[:-1])
    if validate(raw):
        sender.send(raw)
        #print("Sent {}-byte message.".format(len(raw)))
        n_sent += 1
        if n_sent % 1000 == 0:
            print("Sent {} messages.".format(n_sent))
    else:
        print("Invalid {}-byte message.".format(len(raw)))
        n_invalid += 1

print("{} valid messages, {} invalid.".format(n_sent, n_invalid))