cassandra_stress_exporter.py

#
# Run cassandra-stress like this:
#
#    cassandra-stress (...) -log file=cs.log
#
# Run this exporter like this:
#
#    tail -f cs.log | sudo ./cassandra_stress_exporter &
#
# You don't need to restart exporter across c-s restarts.
#
# Exports the following metrics:
#
#    cassandra_stress-0/gauge-ops
#    cassandra_stress-0/gauge-lat_mean
#    cassandra_stress-0/gauge-lat_perc_50
#    cassandra_stress-0/gauge-lat_perc_95
#    cassandra_stress-0/gauge-lat_perc_99
#    cassandra_stress-0/gauge-lat_perc_99
#    cassandra_stress-0/gauge-lat_perc_max
#    cassandra_stress-0/gauge-errors
#

import sys
import socket
import os
import time

hostname = socket.gethostname().split('.')[0]

server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
server.connect('/var/run/collectd-unixsock')

def readlines(sock, recv_buffer=4096, delim='\n'):
    buffer = ''
    data = True
    while data:
        data = sock.recv(recv_buffer)
        buffer += data
        while buffer.find(delim) != -1:
            line, buffer = buffer.split('\n', 1)
            yield line
    return

class ValuePack:
    def __init__(self):
        self.messages = []
        self.time = time.time()
    def add(self, plugin, plugin_instance, type, type_instance, val):
        self.messages.append('PUTVAL "%s/%s-%s/%s-%s" %f:%f\n' % (hostname, plugin, plugin_instance, type, type_instance, self.time, al))
    def send(self):
        for msg in self.messages:
            server.send(msg)
        lines = readlines(server)
        for msg in self.messages:
            reply = next(lines)
            if not reply.startswith('0 Success'):
                print('collectd error: ' + reply)

start_time = time.time()
plugin = 'cassandra_stress'
plugin_instance = '0'

with os.fdopen(sys.stdin.fileno(), 'r', 1) as input:
    while True:
        line = sys.stdin.readline()

        if not line.startswith('total,'):
            continue

        if time.time() - start_time < 1.0: # Skip existing lines
            continue

        cols = line.split()

        ops = float(cols[2].rstrip(','))
        lat_mean = float(cols[5].rstrip(','))
        lat_med = float(cols[6].rstrip(','))
        lat_perc_95 = float(cols[7].rstrip(','))
        lat_perc_99 = float(cols[8].rstrip(','))
        lat_perc_999 = float(cols[9].rstrip(','))
        lat_max = float(cols[10].rstrip(','))
        errors = int(cols[13].rstrip(','))

        pack = ValuePack()
        pack.add(plugin, plugin_instance, 'gauge', 'ops', ops)
        pack.add(plugin, plugin_instance, 'gauge', 'lat_mean', lat_mean)
        pack.add(plugin, plugin_instance, 'gauge', 'lat_perc_50', lat_med)
        pack.add(plugin, plugin_instance, 'gauge', 'lat_perc_95', lat_perc_95)
        pack.add(plugin, plugin_instance, 'gauge', 'lat_perc_99', lat_perc_99)
        pack.add(plugin, plugin_instance, 'gauge', 'lat_perc_99.9', lat_perc_999)
        pack.add(plugin, plugin_instance, 'gauge', 'lat_perc_max', lat_max)
        pack.add(plugin, plugin_instance, 'gauge', 'errors', errors)
        pack.send()

    server.close()