Skip to main content

Mikkoo is a PgQ to RabbitMQ Relay

Project description

A PgQ to RabbitMQ relay. Mikkoo is a PgQ consumer that that publishes to RabbitMQ. In addition, it includes a built in auditing system that can be used to confirm that all PgQ events are received by RabbitMQ.

Mikkoo is named for the rabbit in the “Clever Rabbit and the Elephant” fable.

Version Downloads Status

Installation

Mikkoo is available on the Python Package Index and can be installed via pip:

pip install mikkoo

Once you’ve setup Skytools you may want to install the optional included utility functions in mikkoo.sql to make usage easier.

You can do this with a combination of curl and psql:

curl -L https://github.com/gmr/mikkoo/blob/main/mikkoo.sql | psql

This will install multiple stored procedures and an audit table in a mikkoo schema. Take a look at the DDL to get a good idea of what each funciton is and how it can be used.

PgQ Setup

  1. Install pgq into your database and create the queue:

    # CREATE EXTENSION pgq;
    CREATE EXTENSION
    # SELECT pgq.create_queue('test');
    create_queue
    --------------
                1
    (1 row)
  2. Ensure that pgqd is running.

PgQ Event to AMQP Mapping

When inserting events into a PgQ queue, the pgq.insert_event/7 function should be used with the following field mappings:

PgQ Event

AMQP

ev_type

Routing Key

ev_data

Message body

ev_extra1

Exchange

ev_extra2

Content-Type Property

ev_extra3

AMQP Properties [1]

ev_extra4

Headers [2]

ev_time

Headers.timetamp

ev_txid

Headers.txid

There is a convenience schema in the mikkoo.sql file that adds stored procedures for creating properly formatted mikkoo events in PgQ. In addition, there is are auditing functions that allow for the creation of an audit-log of events that were sent to PgQ.

AMQP Message Properties

The following table defines the available fields that can be set in a JSON blob in the ev_extra3 field when inserting an event.

Property

PgSQL Type

app_id

text

content_encoding

text

content_type

text

correlation_id

text

delivery_mode

int2

expiration

text

message_id

text

headers

text/json [3]

timestamp

int4

type

text

priority

int4

user_id

text

Values assigned in the JSON blob provided to ev_extra3 take precedence over the automatically assigned app_id, content_type, correlation_id, headers, and timestamp values created by Mikkoo at processing time.

As of 1.0, Mikkoo will automatically add four AMQP headers property values. These values will not overwrite any values with the same name specified in ev_extra4. overwriting values with the same name, even if they are specified in the The sequence value is a dynamically generated ID that attempts to provide fuzzy distributed ordering information. The timestamp value is the ISO-8601 representation of the ev_time field, which is created when an event is added to PgQ. The txid value is the ev_txid value, the PgQ transaction ID for the event. These values are added to help provide some level of deterministic ordering. The origin value is the hostname of the server that Mikkoo is running on.

Event Insertion Example

The following example inserts a JSON blob message body of {"foo": "bar"} that will be published to the postgres exchange in RabbitMQ using the test.routing-key routing key. The content type is specified in ev_extra2 and the AMQP type message property is specified in ev_extra3.

# SELECT pgq.insert_event('test', 'test.routing-key', '{"foo": "bar"}', 'postgres', 'application/json', '{"type": "example"}', '');
 insert_event
--------------
            4
(1 row)

When this message is received by RabbitMQ it will have a message body of:

{"foo": "bar"}

And it will have message properties similar to the following:

Property

Example Value

app_id

mikkoo

content_type

application/json

correlation_id

0ad6b212-4c84-4eb0-8782-9a44bdfe949f

headers

Key

Example Value

origin

mikkoo.domain.com

sequence

4539586784185129828

timestamp

2017-02-23 06:17:14.471682-00

txid

41356335556

timestamp

1449600290

type

example

Configuration

The Mikkoo configuration file uses YAML for markup and allows for one or more PgQ queue to be processed.

If you have a Sentry or a Sentry account, the Application/sentry_dsn setting will turn on sentry exception logging, if the raven client library is installed.

Queues are configured by name under the Application/workers stanza. The following example configures two workers for the processing of a queue named invoices. Each worker process connects to a local PostgreSQL and RabbitMQ instance using default credentials.

Application:
  workers:
     invoices:
       postgres_url: postgresql://localhost:5432/postgres
       rabbitmq:
         host: localhost
         port: 5671
         vhost: /
         ssl_options:
           protocol: 2
       confirm: false

Queue Configuration Options

The following table details the configuration options available per queue:

Key

Description

confirm

Enable/Disable RabbitMQ Publisher Confirmations. Default: True

consumer_name

Overwrite the default PgQ consumer name. Default: mikkoo

max_failures

Maximum failures before discarding an event. Default: 10

postgresql_url

The url for connecting to PostgreSQL

rabbitmq

Data structure for connection parameters to connect to RabbitMQ

retry_delay

How long in seconds until PgQ emits failed events. Default: 10

unregister

Unregister a consumer with PgQ on shutdown. Default: True

wait_duration

How long to wait before checking the queue after the last empty result. Default: 10

rabbitmq attributes:

Attribute

Description

host

The hostname or ip address of the RabbitMQ server (str)

port

The port of the RabbitMQ server (int)

vhost

The virtual host to connect to (str)

username

The username to connect as (str)

password

The password to use (str)

ssl_options

Optional: the SSL options for the SSL connection socket

heartbeat_interval

Optional: the AMQP heartbeat interval (int) default: 300 sec

ssl_options attributes:

Attribute

Description

ca_certs

The file path to the concatenated list of CA certificates (str)

ca_path

The directory path to the PEM formatted CA certificates (str)

ca_data

The PEM encoded CA certificates (str)

prototcol

The ssl PROTOCOL_* enum integer value. Default: 2 for enum PROTOCOL_TLS (int)

certfile

The file path to the PEM formatted certificate file (str)

keyfile

The file path to the certificate private key (str)

password

The password for decrypting the keyfile private key (str)

ciphers

The set of available ciphers in the OpenSSL cipher list format (str)

Example Configuration

The following is an example of a full configuration file:

Application:

  poll_interval: 10
  sentry_dsn: [YOUR SENTRY DSN]

  statsd:
    enabled: true
    host: localhost
    port: 8125

  workers:
    test:
      confirm: False
      consumer_name: my_consumer
      max_failures: 5
      postgres_url: postgresql://localhost:5432/postgres
      rabbitmq:
        host: localhost
        port: 5671
        username: guest
        password: guest
        ssl_options:
          protocol: 2
      retry_delay: 5
      unregister: False
      wait_duration: 5

Daemon:
  user: mikkoo
  pidfile: /var/run/mikkoo

Logging:
  version: 1
  formatters:
    verbose:
      format: '%(levelname) -10s %(asctime)s  %(process)-6d %(processName) -20s %(name) -18s: %(message)s'
      datefmt: '%Y-%m-%d %H:%M:%S'
  handlers:
    console:
      class: logging.StreamHandler
      formatter: verbose
      debug_only: True
  loggers:
    helper:
      handlers: [console]
      level: INFO
      propagate: true
    mikkoo:
      handlers: [console]
      level: INFO
      propagate: true
    pika:
      handlers: [console]
      level: ERROR
      propagate: true
    queries:
      handlers: [console]
      level: ERROR
      propagate: true
    tornado:
      handlers: [console]
      level: ERROR
      propagate: true
  root:
    handlers: [console]
    level: CRITICAL
    propagate: true
  disable_existing_loggers: true
  incremental: false

Running Mikkoo

After creating a configuration file for Mikkoo like the one above, simply run the mikkoo application providing the path to the configuration file:

mikkoo -c mikkoo.yml

The application will attempt to daemonize unless you use the -f foreground CLI switch.

Mikkoo’s CLI help can be invoked with --help and yields the following output:

$ mikkoo -h
usage: mikkoo [-h] [-c CONFIG] [-f]

Mikkoo is a PgQ to RabbitMQ Relay

optional arguments:
  -h, --help            show this help message and exit
  -c CONFIG, --config CONFIG
                        Path to the configuration file
  -f, --foreground      Run the application interactively

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

mikkoo-2.2.1.tar.gz (25.9 kB view hashes)

Uploaded source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page