Cassandra slow query log analysis tool

Question

Cassandra, when debugging is enabled, logs slow queries to the debug log file. Typical entries look like:

DEBUG [ScheduledTasks:1] 2017-02-16 18:58:44,342 MonitoringTask.java:572 - 4 operations were slow in the last 5010 msecs:
<SELECT  FROM foo.bar WHERE token(id) > token(9be90fe7-9a6d-45d5-ad11-e93cfd56def7) LIMIT 100>, time 1 msec - slow timeout 1 msec
<SELECT  FROM foo.bar WHERE token(id) > token(91faceee-a64b-4fd3-bb93-ef483acade88) LIMIT 100>, time 1 msec - slow timeout 1 msec
<SELECT  FROM foo.bar WHERE token(id) > token(47250d17-573a-4d76-9039-d2771a19ff10) LIMIT 100>, time 1 msec - slow timeout 1 msec
<SELECT  FROM foo.bar WHERE token(id) > token(e04fc6d0-18b8-4ac0-b5f9-df42cd3a03c5) LIMIT 100>, time 1 msec - slow timeout 1 msec

The actual format is only documented in code.

For MySQL, the mysqldumpslow tool parses the logs and prints the queries (and related statistics) in a readable manner. I'm trying to write a similar tool for Cassandra, for the feature request in CASSANDRA-13000.

The goals I set are:

1. Use similar options to mysqldumpslow, where applicable, so I've to implement these options:

   --help  Display help message and exit
   -g  Only consider statements that match the pattern
   -r  Reverse the sort order
   -s  How to sort output
   -t  Display only first num queries
   

   Sorting options:

   • t, at: Sort by query time or average query time
   • c: Sort by count

   Of these, the -g option is yet to be implemented, since there are some problems in how the queries are logged.

   I'm also adding long-form variants of these (--sort, --reverse, etc.) consistently.

2. Support JSON encoded input, in a streaming fashion. This is for another related patch I'm submitting, where the queries are dumped with JSON encoding for easier parsing by external tools. The JSON-encoded entry will look like:

   {
     "operation": "SELECT  FROM foo.bar WHERE token(id) > token(60bad0b3-551f-46c7-addc-4e3105561a21) LIMIT 100",
     "totalTime": 1,
     "timeout": 1,
     "isCrossNode": false,
     "numTimesReported": 1,
     "minTime": 1,
     "maxTime": 1,
     "keyspace": "foo",
     "table": "bar"
   }
   

3. Keep compatibility with Python 2 and 3

The code:

csqldumpslow.py:

#! /usr/bin/env python3

from __future__ import print_function

import re
import sys
import getopt
import json


def usage():
    msg = """Usage:
    {} [OPTION] ... [FILE] ...

Provide a summary of the slow queries listed in Cassandra debug logs.
Multiple log files can be provided, in which case, the logs are combined.
If no file is specified, logs/debugs.log is assumed. Use - for stdin.

  -h, --help          Print this message
  -s, --sort=type     Sort the input
                          t   - total time
                          at  - average time
                          c   - count
  -r, --reverse       Reverse the sort order
  -t, --top=N         Print only the top N queries (only useful when sorting)
  -j, --json          Assume input consists of slow queries encoded in JSON
  -o, --output=FILE   Save output to FILE

"""
    print(msg.format(sys.argv[0]))


class query_stats:
    def __init__(self, time=0, avg=0, mintime=0, maxtime=0, count=1):
        if count == 1:
            self.time = self.avg = self.mintime = self.maxtime = time
            self.count = 1
        else:
            self.avg = avg
            self.mintime = mintime
            self.maxtime = maxtime
            self.count = count
            self.time = time

    def __str__(self):
        if self.count == 1:
            return "{}ms".format(self.time)
        else:
            return "{}ms ({}ms) Min: {}ms Max: {}ms".format(
                    self.avg,
                    self.time,
                    self.mintime,
                    self.maxtime
                    )


class slow_query:
    def __init__(self, operation, stats, timeout,
                 keyspace=None, table=None, is_cross_node=False):
        self.operation = operation
        self.stats = stats
        self.timeout = timeout
        self.keyspace = keyspace
        self.table = table
        self.is_cross_node = is_cross_node

    def __str__(self):
        return "  Time: {} {} Timeout: {}\n\t{}\n".format(
            self.stats,
            "(cross-node)" if self.is_cross_node else "",
            self.timeout,
            self.operation)


class log_parser:
    regexes = {
            'start': re.compile('DEBUG.*- (\d+) operations were slow in the last (\d+) msecs:$'), # noqa
            'single': re.compile('<(.*)>, time (\d+) msec - slow timeout (\d+) msec(/cross-node)?$'), # noqa
            'multi': re.compile('<(.*)>, was slow (\d+) times: avg/min/max (\d+)/(\d+)/(\d+) msec - slow timeout (\d+) msec(/cross-node)?$'), # noqa
            }

    def __init__(self, sort, key, reverse, top, top_count, json_input):
        self.queries = []
        self.sort = sort
        self.key = key
        self.reverse = reverse
        self.top = top
        self.top_count = top_count
        self.json_input = json_input

    def process_query(self, query):
        # If we're not sorting, we can print the queries directly. If we are
        # sorting, save the query.
        if self.sort:
            self.queries.append(query)
        else:
            # If we have to print only N entries, exit after doing so
            if self.top:
                if self.top_count > 0:
                    self.top_count -= 1
                else:
                    sys.exit()
            print(query)

    def parse_slow_query_stats(self, line):
        match = log_parser.regexes['single'].match(line)
        if match is not None:
            self.process_query(slow_query(
                operation=match.group(1),
                stats=query_stats(int(match.group(2))),
                timeout=int(match.group(3)),
                is_cross_node=(match.group(4) is None)
                ))
            return
        match = log_parser.regexes['multi'].match(line)
        if match is not None:
            self.process_query(slow_query(
                operation=match.group(1),
                stats=query_stats(
                    count=int(match.group(2)),
                    avg=int(match.group(3)),
                    time=int(match.group(3))*int(match.group(2)),
                    mintime=int(match.group(4)),
                    maxtime=int(match.group(5))
                    ),
                timeout=match.group(6),
                is_cross_node=(match.group(7) is None)
                ))
            return
        print("Could not parse: " + line, file=sys.stderr)
        sys.exit(1)

    def get_json_objects(self, infile):
        # Since Python's json doesn't support streaming, try accumulating line
        # by line, and parsing.
        prev = ""
        for line in infile:
            try:
                yield json.loads(prev + line)
            except json.JSONDecodeError:
                prev += line

    def parse_json(self, infile):
        for obj in self.get_json_objects(infile):
            self.process_query(slow_query(
                operation=obj["operation"],
                stats=query_stats(
                    count=obj["numTimesReported"],
                    time=obj["totalTime"],
                    avg=obj["totalTime"]/obj["numTimesReported"],
                    mintime=obj["minTime"],
                    maxtime=obj["maxTime"]
                    ),
                timeout=obj["timeout"],
                is_cross_node=obj["isCrossNode"]
                ))

    def parse_log(self, infile):
        if self.json_input:
            self.parse_json(infile)
        else:
            # How many queries does the current log entry list?
            current_count = 0
            for line in infile:
                line = line.rstrip()
                if current_count > 0:
                    self.parse_slow_query_stats(line)
                    current_count -= 1
                else:
                    match = log_parser.regexes['start'].match(line)
                    if match is None:
                        continue
                    current_count = int(match.group(1))

    def sort_queries(self):
        # Sort by total time, default
        if self.key is None or self.key == 't':
            self.queries.sort(key=lambda x: x.stats.time, reverse=self.reverse)
        # Sort by avergae time
        elif self.key == 'at':
            self.queries.sort(key=lambda x: x.stats.avg, reverse=self.reverse)
        # Sort by count
        elif self.key == 'c':
            self.queries.sort(key=lambda x: x.stats.count, reverse=self.reverse) # noqa
        return

    def end(self):
        # Sort and print
        if self.sort:
            self.sort_queries()
            if self.top:
                self.queries = self.queries[:self.top_count]
            for q in self.queries:
                print(q)


def main():
    opts, args = getopt.gnu_getopt(
            sys.argv[1:],
            'hs:rt:jo:',
            [
                'help',
                'sort=',
                'reverse',
                'top=',
                'json',
                'output=',
            ]
            )
    # Defaults:
    # Do not sort
    sort = False
    key = None
    # Do not reverse
    reverse = False
    # Print all lines, top_count is ignored if top is unset
    top = False
    top_count = 0
    # Assume debug.log-style input, not JSON
    json_input = False

    for opt, arg in opts:
        if opt in ["-h", "--help"]:
            usage()
            sys.exit()
        elif opt in ["-s", "--sort"]:
            sort = True
            key = arg
        elif opt in ["-r", "--reverse"]:
            reverse = True
        elif opt in ["-t", "--top"]:
            top = True
            top_count = int(arg)
        elif opt in ["-j", "--json"]:
            json_input = True
        elif opt in ["-o", "--output"]:
            sys.stdout = open(arg, "a")
        else:
            print("Not yet implemented: " + opt)
            sys.exit(1)

    if len(args) == 0:
        # Default to reading the debug.log
        args = ['logs/debug.log']
    parser = log_parser(sort, key, reverse, top, top_count, json_input)
    for arg in args:
        if arg == '-':
            print("Reading from standard input")
            parser.parse_log(sys.stdin)
        else:
            with open(arg) as infile:
                print("Reading from " + arg)
                parser.parse_log(infile)
    parser.end()


if __name__ == "__main__":
    main()

Answer 1

Here are some notes about the code (both performance and code style related):

• since you are initializing a lot of slow_query and query_stats (also see note about the naming below) class instances on the fly, to improve the memory usage and performance, use __slots__:

  class slow_query:
      __slots__ = ["operation", "stats", "timeout", "keyspace", "table", "is_cross_node"]
      # ...
  

• switching from json to ujson may dramatically improve the JSON parsing speed

• or, you can try the PyPy and simplejson combination (ujson won't work on PyPy since it is written in C, simplejson is a fast pure-python parser)

• think about the capturing groups in your regular expressions, you can avoid capturing more things than you actually need. For example, in the "start" regular expression you have 2 capturing groups, but you actually use only the first one:

  r'DEBUG.*- (\d+) operations were slow in the last \d+ msecs:$'
                                       no group here^
  

• the wild card matches in the regular expressions can be non-greedy - .*? instead of .* (not sure if it will have a measurable impact on performance)

• class names should use a "CamelCase" convention (PEP8 reference)

• the .get_json_objects() method can be static
• for the CLI parameter parsing I would use argparse module - you would avoid the boilerplate code you have in the main() and usage() functions
• use 2 spaces before the # for the inline comment (PEP8 reference)
• fix typo "avergae" -> "average"

• you can improve the readability of the sort_queries() method by introducing a mapping between the key and the sort attribute name, something along these lines:

  def sort_queries(self):
      """Sorts "queries" in place, default sort is "by time"."""
      sort_attributes = {
          't': 'time',
          'at': 'avg',
          'c': 'count'
      }
      sort_attribute = sort_attributes.get(self.key, 't')
  
      self.queries.sort(key=lambda x: getattr(x.stats, sort_attribute), 
                        reverse=self.reverse)
  

  It though feels like this mapping should be defined as a constant beforehand.

• improve on documentation: add meaningful docstrings to the class methods, put comments whenever you think the reader may have difficulties to understand the code - remember, the code is being read much more often than written

Note that this is what I can see by looking at the code. Of course, to really identify the bottleneck(s), you should profile the code properly on a large input.