''' collectd plugin for zookeeper collectd rrd data will be stored 'coordinator-port' folder under current collectd hostname @author james.kim collectd.conf sample --------------------------------------- LoadPlugin python ModulePath "/collectd/share/python/" LogTraces true Interactive false Import zookeeper_stat_plugin HOST "localhost" PORT 2181 CONN_TIMEOUT 1.0 #RRDSTORE_HOSTNAME "put.a.host.name" collectd types.db --------------------------------------- coordinator_stat sessions_count:GAUGE:0:U, min_latency:GAUGE:0:U, avg_latency:GAUGE:0:U, max_latency:GAUGE:0:U, received:GAUGE:0:U, sent:GAUGE:0:U, outstanding:GAUGE:0:U, node_count:GAUGE:0:U ''' import time import collectd import os, re, copy import zookeeper import StringIO import telnetlib import socket import sys, traceback class Plugin: HOST = 'localhost' PORT = 2181 CONN_TIMEOUT = 1.0 RRDSTORE_HOSTNAME = None plugin = 'coordinator' type = 'coordinator_stat' request_command = "stat" full_request_command = "" thisfilename = "" def configure(self, conf): """Receive configuration block""" thisfile = os.path.abspath(__file__) self.thisfilename = os.path.basename(thisfile) for node in conf.children: if node.key == 'HOST': self.HOST = node.values[0] elif node.key == 'PORT': self.PORT = int(node.values[0]) elif node.key == 'CONN_TIMEOUT': self.CONN_TIMEOUT = float(node.values[0]) elif node.key == 'RRDSTORE_HOSTNAME': self.RRDSTORE_HOSTNAME = node.values[0] else: self.log_info('Unknown config key: %s.' % (node.key)) if not self.HOST or not self.PORT: raise Exception("%s: invalid configuration; HOST=%s, PORT=%s" % (self.thisfilename, self.HOST, self.PORT)) self.plugin = "%s-%s" % (self.plugin, self.PORT) self.full_request_command = "%s:%s %s" % (self.HOST, self.PORT, self.request_command) self.log_info("configuration complete") def initConnection(self): pass def getConnection(self): self.log_info("trying to get a new connection; timeout:%s sec" % (self.CONN_TIMEOUT)) #connection = telnetlib.Telnet(self.HOST, self.PORT, timeout=self.CONN_TIMEOUT) # this line blocks for 1sec. and affected to other plugins of pausing collecting connection = telnetlib.Telnet(self.HOST, self.PORT) return connection def request_data(self, request_command): self.log_debug("requesting data") connection = self.getConnection() if connection == None: self.log_warning("no connection") return connection.write(request_command) response = connection.read_all() connection.close() self.log_debug("response; response:%s" % (response)) return response def fetch_data(self, request_command): self.log_debug("fetching data success") data = None try: data = self.request_data(request_command) except Exception as e: ''' connection should be disabled here. ''' self.log_debug("clear connection ") self.connection = None raise Exception("fetch data error:%s, trace:%s" % (str(e), traceback.format_exc().splitlines())) if data == None or data == "": raise Exception("empty response") return data def submit(self, values): metric = collectd.Values(); metric.plugin = self.plugin metric.type = self.type metric.values = values if self.RRDSTORE_HOSTNAME: metric.host = self.RRDSTORE_HOSTNAME metric.dispatch() self.log_debug("submit success") def teardown(self): self.log_info("closing connection to... %s %s" % (self.HOST, self.PORT)) if self.connection: self.connection.close() def log_debug(self, string): collectd.debug("%s: %s, url:%s" % (self.thisfilename, string, self.full_request_command)) def log_info(self, string): collectd.info("%s: %s, url:%s" % (self.thisfilename, string, self.full_request_command)) def log_warning(self, string): collectd.warning("%s: %s, url:%s" % (self.thisfilename, string, self.full_request_command)) def log_error(self, string): collectd.error("%s: url:%s message:%s" % (self.thisfilename, self.full_request_command, string)) #---------------------------------------------------------------------------------------------- def parse_data(self, stat_str): sio = StringIO.StringIO(stat_str) firstline = sio.readline() # :Zookeeper version: 3.3.3-1073969, built on 02/23/2011 22:27 GMT #collectd.debug("%s: firstline:%s" %(self.thisfilename, firstline)) if firstline.find("This ZooKeeper instance is not currently serving requests") > -1 : raise Exception(firstline) sio.readline() #Clients: self.sessions_count = 0 for line in sio: if not line.strip(): #:/127.0.0.1:44726[1](queued=0,recved=15,sent=15) break self.sessions_count = self.sessions_count + 1 #collectd.info("zookeeper_stat_plugin.py: sessions_count %s"%(self.sessions_count)) #Latency, Reveived, Send, Outstanding, Zxid, Mode, Node Count for line in sio: attr, value = line.split(':') attr = attr.strip().replace(" ", "_").replace("/", "_").lower() self.__dict__[attr] = value.strip() #collectd.info("%s:%s"%(attr,value)) self.min_latency, self.avg_latency, self.max_latency = self.latency_min_avg_max.split("/") return [self.sessions_count, self.min_latency, self.avg_latency, self.max_latency, self.received, self.sent, self.outstanding, self.node_count] def dispatcher(self): try: data = self.fetch_data(self.request_command) values = self.parse_data(data) if values == None or len(values) == 0: self.log_warning("empty response data") return self.submit(values) except Exception as e: self.log_error("dispatch data error:%s, trace:%s" % (str(e), traceback.format_exc().splitlines())) return self.log_info("dispatching success") plugin = Plugin() #== Hook Callbacks, Order is important! ==# collectd.register_config(plugin.configure) collectd.register_init(plugin.initConnection) collectd.register_read(plugin.dispatcher) collectd.register_shutdown(plugin.teardown)