Find active sensu checks

sensu_checks.py

#!/usr/bin/env python

# The MIT License (MIT)

# Copyright (c) 2016 Michael-Keith Bernard

# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:

# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import os
import json
import shlex
import pipes
import argparse
import subprocess

__author__ = "Michael-Keith Bernard"
# See also: https://gist.github.com/SegFaultAX/85fb321af3cd151bff91

COMMAND_PREFIXES = [
    "/opt/sensu/embedded/bin/",
    "/usr/lib/nagios/plugins/",
]


def parse_args(args=None):
    parser = argparse.ArgumentParser(description="Find all active Sensu check commands")
    parser.add_argument("-p", "--path",
        default="/etc/sensu/conf.d/checks",
        help="path to sensu checks")
    parser.add_argument("-r", "--resolve",
        default=False,
        action="store_true",
        help="try to resolve non-absolute paths")
    parser.add_argument("-x", "--execute",
        default=False,
        action="store_true",
        help="run selected checks")

    return parser, parser.parse_args(args)


def run(cmd):
    args = shlex.split(cmd)
    proc = subprocess.Popen(
        args,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    stdout, stderr = proc.communicate()
    return stdout, stderr, proc.returncode


def find_checks(base):
    for root, dirs, files in os.walk(base):
        for f in files:
            if f.endswith(".json"):
                yield os.path.join(root, f)


def load_checks(files):
    checks = {}
    for f in files:
        try:
            with open(f) as fd:
                data = json.load(fd)
            checks[f] = data["checks"]
        except:
            continue
    return checks


def resolve_bin(name):
    if os.path.isabs(name):
        return name
    for prefix in COMMAND_PREFIXES:
        path = os.path.join(prefix, name)
        if os.path.exists(path):
            return path
    return name


def resolve_command(command):
    parts = shlex.split(command)
    exe = resolve_bin(parts[0])
    return " ".join(pipes.quote(p) for p in ([exe] + parts[1:]))


def main():
    _parser, args = parse_args()

    if os.path.isdir(args.path):
        files = list(find_checks(args.path))
    elif os.path.isfile(args.path):
        files = [args.path]
    else:
        raise RuntimeError(
            "{} is not a valid file or directory".format(args.path))

    checks = load_checks(files)

    for path, check_data in checks.iteritems():
        print path
        for check_name, check_contents in check_data.iteritems():
            valid = True
            command = check_contents.get("command").strip()
            if command and args.resolve:
                command = resolve_command(command)
            elif not command:
                command = "<no command supplied>"
                valid = False

            print "  {}: {}".format(check_name, command)
            if valid and args.execute:
                out, err, ret = run(command)
                print "  RETURN CODE: {}".format(ret)
                if out:
                    print "  STDOUT: {}".format(out.strip())
                if err:
                    print "  STDERR: {}".format(err.strip())
            print


if __name__ == "__main__":
    main()

since you've already shlex.split's the command. you don't need the shell=True in the call to Popen you get the shell parsing strings that you had originally quoted.

want to restructure this so it can give a report a the end.. [ like rspec or something. ] and also to add a filter for tests matching ....

also I have some incredibly slow checks. that sensu would have timed out. a timeout, and/or running in parallel would be nice.

feature creeper. 💥