WIP: AMQP support

[ghost]

AMQP support is feature-complete, needs still to be tested on development machines and some unittests are needed.

• All features for intelmqctl
• Unit tests
• Documentation
• real world tests
• clearing queues
• paths
• lazy queues

[codecov-io]

Codecov Report

Merging #1179 into develop will decrease coverage by 0.03%.
The diff coverage is 21.01%.

@@             Coverage Diff             @@
##           develop    #1179      +/-   ##
===========================================
- Coverage    75.11%   75.08%   -0.04%     
===========================================
  Files          318      288      -30     
  Lines        14402    13267    -1135     
  Branches      1911     1793     -118     
===========================================
- Hits         10818     9961     -857     
+ Misses        3137     2909     -228     
+ Partials       447      397      -50

Impacted Files	Coverage Δ
intelmq/lib/pipeline.py	48.16% <22.58%> (-19.59%)	⬇️
intelmq/bin/intelmqctl.py	9.86% <7.14%> (-0.18%)	⬇️
intelmq/bots/parsers/zoneh/parser.py	28.57% <0%> (-71.43%)	⬇️
intelmq/bots/parsers/shodan/parser.py	19.64% <0%> (-70.53%)	⬇️
intelmq/bots/parsers/calidog/parser_certstream.py	36% <0%> (-40%)	⬇️
...q/bots/collectors/microsoft/collector_interflow.py	27.41% <0%> (-19.5%)	⬇️
intelmq/bots/parsers/n6/parser_n6stomp.py	62.1% <0%> (-14.43%)	⬇️
...sts/bots/parsers/webinspektor/test_webinspektor.py	91.3% <0%> (-8.7%)	⬇️
...q/tests/bots/parsers/malwareurl/test_malwareurl.py	91.3% <0%> (-8.7%)	⬇️
...tests/bots/parsers/threatminer/test_threatminer.py	91.3% <0%> (-8.7%)	⬇️
... and 150 more

[ghost]

There are speed problems and problems with unacked messages. Postponing this

[ghost]

  File "/home/sebastianw/intelmq/intelmq/lib/pipeline.py", line 293, in __init__
    raise exceptions.ConfigurationError("To use AMQP you must install the 'pika' library.")
TypeError: __init__() missing 1 required positional argument: 'argument'

[SYNchroACK]

@wagner-certat what do you think about using "lazy queues" feature?
https://www.rabbitmq.com/lazy-queues.html

I think is something that we miss a lot due a lot of messages that stuck on experts (like cymru-whois, gethostbyname and deduplicator).

I think (not sure), we only need to pass the x-queue-mode value inside the args parameter here: https://pika.readthedocs.io/en/latest/modules/channel.html?highlight=channel#pika.channel.Channel.queue_declare

[SYNchroACK]

$ cat /opt/intelmq/etc/defaults.conf 
{
    "destination_pipeline_broker": "Amqp",
    "destination_pipeline_db": 2,
    "destination_pipeline_host": "127.0.0.1",
    "destination_pipeline_password": null,
    "destination_pipeline_port": 5672,

  ....

    "source_pipeline_broker": "Amqp",
    "source_pipeline_db": 2,
    "source_pipeline_host": "127.0.0.1",
    "source_pipeline_password": null,
    "source_pipeline_port": 5672
}

$ intelmqctl list queues
Traceback (most recent call last):
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 174, in _read_from_socket
    data = recv(self._sock, socket_read_size)
  File "/usr/lib/python3/dist-packages/redis/_compat.py", line 79, in recv
    return sock.recv(*args, **kwargs)
ConnectionResetError: [Errno 104] Connection reset by peer

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3/dist-packages/redis/client.py", line 667, in execute_command
    connection.send_command(*args)
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 610, in send_command
    self.send_packed_command(self.pack_command(*args))
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 585, in send_packed_command
    self.connect()
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 493, in connect
    self.on_connect()
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 567, in on_connect
    if nativestr(self.read_response()) != 'OK':
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 624, in read_response
    response = self._parser.read_response()
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 284, in read_response
    response = self._buffer.readline()
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 216, in readline
    self._read_from_socket()
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 191, in _read_from_socket
    (e.args,))
redis.exceptions.ConnectionError: Error while reading from socket: (104, 'Connection reset by peer')

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib/python3/dist-packages/redis/connection.py", line 174, in _read_from_socket
    data = recv(self._sock, socket_read_size)
  File "/usr/lib/python3/dist-packages/redis/_compat.py", line 79, in recv
    return sock.recv(*args, **kwargs)
ConnectionResetError: [Errno 104] Connection reset by peer

On this line (https://github.com/certtools/intelmq/blob/dev-amqp/intelmq/lib/pipeline.py#L39) it seems that we need to add .title()

Also, on this line (https://github.com/certtools/intelmq/blob/dev-amqp/intelmq/lib/pipeline.py#L41) it seems now we have three broker options, and I guess the one on line 41 will be required by intelmqctl

[ghost]

Ad lazy queues: Thanks for the pointer, definitely

On this line (https://github.com/certtools/intelmq/blob/dev-amqp/intelmq/lib/pipeline.py#L39) it seems that we need to add .title()

yes, thanks.

Also, on this line (https://github.com/certtools/intelmq/blob/dev-amqp/intelmq/lib/pipeline.py#L41) it seems now we have three broker options, and I guess the one on line 41 will be required by intelmqctl

What do you mean?