☰

AMQPS から Syslog

tag-icon • tag-icon syslog-ng rabbitmq
AMQPS から Syslog

現在、クラウド セキュリティ ベンダーからログを取得するために基本的な API を使用していますが、あまりクリーンなプロセスではありません。ベンダーが提供するイベント ストリーミング (AMQPS) サービスに移行したいと考えています。

キューはすでに設定済みで、クラウド側に関連情報もありますが、私が直面している問題は、そのストリームを受け入れて syslog にダンプする最善の方法を見つけることです。現在、syslog-ng を使用していますが、RabbitMQ に送信することしかできず、受け入れることはできないようです。

現在、Pika で AMQPS を受け入れてログ ファイルに送ることができるかどうかを確認中ですが、そこで問題が発生しています。どなたか助けていただければ幸いです。

ありがとう

答え1

walbitが作成したgithubページを見つけました（https://github.com/walbit/AMP_API) は、私の環境では正常に動作しているようです。私は「consume_event_stream_by_name.py」を使用しており、すべての入力が標準出力にダンプされます。これを syslog にも送信するように少し変更したところ、syslog-ng がこれを受け入れるようになりました。

現時点でこのコードに関して私が抱えている唯一の問題は、接続が来ないと時々タイムアウトしたり、全体的に停止したりすることです。そのため、何が起こっているのかを把握する必要があります。

#!/usr/bin/env python

import argparse
import pika
import pprint
import requests
import sys
import logging
import logging.handlers

# YOU NEED TO CREATE AN auth.py FILE WITH CLIENT_ID AND API_KEY STRINGS
from auth import CLIENT_ID, API_KEY

parser = argparse.ArgumentParser()
parser.add_argument('event_stream_name', metavar='event_stream_name',
                    nargs=1, help='event stream name')
parser.parse_args()
event_stream_name = parser.parse_args().event_stream_name[0]

api_endpoint = 'https://api.amp.cisco.com/v1/event_streams'

session = requests.Session()
session.auth = (CLIENT_ID, API_KEY)

event_streams = session.get(api_endpoint).json()['data']

event_stream = {}

for e in event_streams:
    if e['name'] is event_stream_name:
        event_stream = e

amqp_url = 'amqps://{user_name}:{password}@{host}:{port}'.format(
    **e['amqp_credentials'])
queue = e['amqp_credentials']['queue_name']
parameters = pika.URLParameters(amqp_url)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()


parameters = pika.URLParameters(amqp_url)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()

my_logger = logging.getLogger('MyLogger')
my_logger.setLevel(logging.INFO)
handler = logging.handlers.SysLogHandler(address = '/dev/log')
my_logger.addHandler(handler)

def callback(ch, method, properties, body):
#    print(" [x] Received meth:\t%r" % method)
#    print(" [x] Received prop:\t%r" % properties)
#    print(" [x] Received body:\t%r" % body)
#    print(body)
    my_logger.info('Fireamp: ' + body)


channel.basic_consume(callback, queue, no_ack=True)

print(" [*] Connecting to:\t%r" % amqp_url)
print(" [*] Waiting for messages. To exit press CTRL+C")
channel.start_consuming()

関連情報