Authored by BuddyJack

Fix

  1 +import collectd
  2 +import pika
  3 +import time
  4 +
  5 +SendCon = None
  6 +
  7 +RecvCon = None
  8 +
  9 +RabbitConfig = {
  10 + 'host': '127.0.0.1',
  11 + 'port': 5672,
  12 + 'user': 'guest',
  13 + 'pwd': 'guest',
  14 + 'vhost': '/',
  15 + 'tag': 'rabbit_01'
  16 +}
  17 +
  18 +readFlag = False
  19 +
  20 +def init_SendCon():
  21 + global SendCon
  22 + if SendCon == None or SendCon.is_open == False:
  23 + credentials = pika.PlainCredentials(RabbitConfig['user'], RabbitConfig['pwd'])
  24 + SendCon = pika.BlockingConnection(pika.ConnectionParameters(host=RabbitConfig['host'], port=RabbitConfig['port'], credentials=credentials, virtual_host=RabbitConfig['vhost']))
  25 + if SendCon.is_open == True:
  26 + collectd.info("connect success...")
  27 + else:
  28 + return
  29 +
  30 +
  31 +def init_RecvCon():
  32 + global RecvCon
  33 + if RecvCon == None or RecvCon.is_open == False:
  34 + credentials = pika.PlainCredentials(RabbitConfig['user'], RabbitConfig['pwd'])
  35 + RecvCon = pika.BlockingConnection(pika.ConnectionParameters(host=RabbitConfig['host'], port=RabbitConfig['port'], credentials=credentials, virtual_host=RabbitConfig['vhost']))
  36 + else:
  37 + return
  38 +
  39 +
  40 +def send_check():
  41 + init_SendCon()
  42 + global SendCon
  43 + collectd.info("start send")
  44 + channel = SendCon.channel()
  45 + channel.queue_declare(queue='ctest_'+RabbitConfig['host'])
  46 + channel.basic_publish(exchange='',
  47 + routing_key='ctest_'+RabbitConfig['host'],
  48 + body='quit')
  49 + pass
  50 +
  51 +
  52 +def call_back(ch, method, properties, body):
  53 + global readFlag
  54 + readFlag = True
  55 + ch.basic_cancel(consumer_tag='consumer_'+RabbitConfig['host'])
  56 + ch.stop_consuming()
  57 + pass
  58 +
  59 +
  60 +def recv_check():
  61 + init_RecvCon()
  62 + global RecvCon
  63 + collectd.info("start read")
  64 + channel = RecvCon.channel()
  65 + channel.queue_declare(queue='ctest_'+RabbitConfig['host'])
  66 + channel.basic_consume(call_back,
  67 + queue='ctest_'+RabbitConfig['host'],
  68 + no_ack=True,
  69 + consumer_tag='consumer_'+RabbitConfig['host'])
  70 + channel.start_consuming()
  71 + pass
  72 +
  73 +
  74 +def config_callback(conf=None):
  75 + global RabbitConfig
  76 + for c in conf.children:
  77 + key = c.key.lower()
  78 + value = c.values[0]
  79 + if key == 'port':
  80 + RabbitConfig[key] = int(value)
  81 + else:
  82 + RabbitConfig[key] = str(value)
  83 + pass
  84 +
  85 +
  86 +def read_callback():
  87 + val = collectd.Values(plugin='rabbit-health', plugin_instance=RabbitConfig['tag'])
  88 + val.type = 'rabbitmq_state'
  89 + val.type_instance = ''
  90 + send_check()
  91 + recv_check()
  92 + global readFlag
  93 + if readFlag:
  94 + val.values = [1]
  95 + else:
  96 + val.values = [-1]
  97 + val.dispatch()
  98 + readFlag = False
  99 + pass
  100 +
  101 +
  102 +collectd.register_config(config_callback)
  103 +
  104 +collectd.register_read(read_callback)
  105 +
  106 +
  107 +
  108 +
  109 +
  110 +
  111 +
  112 +