ipmiscrape

#!/usr/bin/env python3
# ipmiscrape (part of ossobv/vcutil) // wdoekes/2022,2023 // Public Domain
#
# Fetch controller info from multiple IPMI boards in parallel.
# Fetch power usage from multiple IPMI boards in parallel.
#
# TODO: more documentation :)
#
import os
import sys
from collections import defaultdict, namedtuple
from multiprocessing import Process, Queue
from subprocess import PIPE, CalledProcessError, Popen, TimeoutExpired
from traceback import format_exc

VERBOSE = False
POOL_SIZE = 30
PROCESS_TIMEOUT = 1.5
IPMI_USER_PASS_BY_ALIAS = {'*': ('ADMIN', 'ADMIN')}  # ~/.config/ipmikvm/dict
IPMI_USER_PASS_BY_IP = {'*': ('ADMIN', 'ADMIN')}     # ~/.config/ipmikvm/dict


Machine = namedtuple('Machine', 'hostname ipmi_ip')
InfoReading = namedtuple(
    'InfoReading',
    'system_guid firmware_rev')
PowerReading = namedtuple(
    'PowerReading',
    'cur_watt max_watt min_watt avg_watt sample_duration power_state')
Result = namedtuple('Result', 'machine reading')


class MultiprocessingProgrammingError(Exception):
    @classmethod
    def from_exception(cls, msg, exception):
        # A traceback does not get pickled well.
        # Must use a factory/classmethod because we're pickling the
        # result: the copy will get the marshalled arguments.
        exc_type_name = exception.__class__.__name__
        exc_summary = str(exception)
        exc_tb = format_exc()
        return cls(msg, exc_type_name, exc_summary, exc_tb)

    def __str__(self):
        return (
            'MultiprocessingProgrammingError({!r}, {!r}, {!r})\n'
            '\n'
            '--- BEGIN CAUSE ---\n'
            '{}'
            '--- END CAUSE ---').format(*self.args)


def check_output(command, env, timeout):
    with Popen(command, env=env, stdout=PIPE, stderr=PIPE) as p:
        output, error = p.communicate(timeout=timeout)
        if p.returncode != 0:
            raise CalledProcessError(p.returncode, (command, output, error))
    return output


def get_machines(filename):
    # Provide machines as a CSV:
    # hostname,ipmi-ip
    with open(filename) as fp:
        lines = [line.strip() for line in fp.read().strip().split('\n')]
        lines = [line for line in lines if line and not line.startswith('#')]
        if lines[0].split(',', 2)[0:2] == ['hostname', 'ipmi-ip']:
            lines.pop(0)
        machines = [Machine(*line.split(',', 2)[0:2]) for line in lines]
    return machines


def get_ipmitool(ip, *ipmi_command):
    first_error = None
    user_pass = IPMI_USER_PASS_BY_IP.get(ip, IPMI_USER_PASS_BY_IP['*'])
    for password_attempt, (user, password) in enumerate(user_pass, 1):
        try:
            env = os.environ.copy()
            env['IPMI_PASSWORD'] = password  # using -E
            command = (
                'ipmitool', '-I', 'lan', '-H', ip, '-U', user, '-E'
                ) + ipmi_command
            if VERBOSE:
                strcmd = ' '.join(command)
                # print(f'#{password_attempt}# IPMI_PASSWORD={password}')
                print(f'#{password_attempt}# {strcmd}')
            res = check_output(command, timeout=PROCESS_TIMEOUT, env=env)
            res = res.decode('ascii', 'replace')
        except CalledProcessError as e:  # invalid username??
            # (cmd, out, err) = e.cmd
            # (exitcode, (cmd, out, err)) = e.args
            if not first_error or (
                    first_error.cmd[2].startswith((
                        b'Invalid user name:',          # bad user
                        b'Activate Session error:')     # bad pass
                    )):
                first_error = e
        else:
            break
    else:
        raise first_error

    return res


def get_ipmitool_mc_info(ip):
    res = {}

    # System GUID  : 3130xxxx-xxxx-xxxx-xxxc-xxxxxxxxxxxx
    # Timestamp    : 02/25/1996 10:51:48
    res.update(to_dict(get_ipmitool(ip, 'mc', 'guid')))

    # Device ID                 : 32
    # Device Revision           : 1
    # Firmware Revision         : 1.23
    # IPMI Version              : 2.0
    # Manufacturer ID           : 10876
    # Manufacturer Name         : Supermicro
    # Product ID                : 6683 (0x1a1b)
    # Product Name              : Unknown (0x1A1B)
    # Device Available          : yes
    # Provides Device SDRs      : no
    # Additional Device Support :
    #     Sensor Device
    # ...
    res.update(to_dict(get_ipmitool(ip, 'mc', 'info'), only_depth_0=True))

    return res


def get_ipmitool_power(ip):
    #     Instantaneous power reading:                   102 Watts
    #     Minimum during sampling period:                 22 Watts
    #     Maximum during sampling period:                166 Watts
    #     Average power reading over sample period:       71 Watts
    #     IPMI timestamp:                           Thu Dec  1 10:27:15 2022
    #     Sampling period:                          03698843 Seconds.
    #     Power reading state is:                   activated
    return to_dict(get_ipmitool(ip, 'dcmi', 'power', 'reading'))


def get_power_reading(machine):
    try:
        kv = get_ipmitool_power(machine.ipmi_ip)
        res = PowerReading(
            cur_watt=int(kv['Instantaneous power reading'].split()[0]),
            min_watt=int(kv['Minimum during sampling period'].split()[0]),
            max_watt=int(kv['Maximum during sampling period'].split()[0]),
            avg_watt=int(
                kv['Average power reading over sample period'].split()[0]),
            sample_duration=int(kv['Sampling period'].split()[0]),
            power_state=kv['Power reading state is'],
        )
    except CalledProcessError as e:
        return e
    except TimeoutExpired as e:
        return e
    except Exception as e:
        return MultiprocessingProgrammingError.from_exception(
            'unhandled {!r} on {!r}'.format(e.__class__.__name__, machine), e)

    return res


def get_info_reading(machine):
    try:
        kv = get_ipmitool_mc_info(machine.ipmi_ip)
        res = InfoReading(
            system_guid=kv['System GUID'],
            firmware_rev=kv['Firmware Revision'],
        )
    except CalledProcessError as e:
        return e
    except TimeoutExpired as e:
        return e
    except Exception as e:
        return MultiprocessingProgrammingError.from_exception(
            'unhandled {!r} on {!r}'.format(e.__class__.__name__, machine), e)

    return res


def to_dict(s, only_depth_0=False):
    if only_depth_0:
        lines = [line.split(':', 1) for line in s.split('\n') if line.strip()]
        lines = [kv for kv in lines if not kv[0].startswith((' ', '\t'))]
    else:
        lines = [
            line.strip().split(':', 1)
            for line in s.split('\n') if line.strip()]
    kv = dict((k.rstrip(), v.strip()) for k, v in lines)
    return kv


def run(jobq, resq, getfunc):
    while not jobq.empty():
        machine = jobq.get()
        result = Result(machine=machine, reading=getfunc(machine))
        resq.put(result)


def dump_results(results, dump_valid_func):
    valid, invalid = [], []
    for res in results:
        if isinstance(res.reading, Exception):
            invalid.append(res)
        else:
            valid.append(res)

    dump_invalid(invalid)
    dump_valid_func(valid)


def dump_info_results(results):
    print('hostname,ipmi-ip,firmware_rev')
    for res in results:
        print(','.join([
            res.machine.hostname, res.machine.ipmi_ip,
            res.reading.firmware_rev]))


def dump_power_results(results):
    print('hostname,ipmi-ip,avg_watt')
    for res in results:
        print(','.join([
            res.machine.hostname, res.machine.ipmi_ip,
            str(res.reading.avg_watt)]))


def dump_invalid(invalid):
    if not invalid:
        return
    print('-- there were some problems, see below --')
    for res in invalid:
        print(res)
    print('-- there were some problems, see above --')
    print()


def read_config_ipmikvm_dict():
    by_alias = defaultdict(list)
    by_ip = defaultdict(list)
    # <alias> <ip> <user> <pass>
    # * * ADMIN SIMPLEPASS
    # * * ADMIN OTHERPASS
    # machinex 1.2.3.4 ROOT ROOTPASS
    with open(os.path.expanduser('~/.config/ipmikvm/dict')) as fp:
        for line in fp:
            line = line.strip()
            if line and not line.startswith('#'):
                components = line.split()
                if len(components) >= 4:
                    alias, ip, user, password = components[0:4]
                    by_alias[alias].append((user, password))
                    by_ip[ip].append((user, password))
    return dict(by_alias), dict(by_ip)


def main(what, machines_filename):
    machines = get_machines(machines_filename)
    job_queue = Queue()
    [job_queue.put(machine) for machine in machines]
    res_queue = Queue()
    pool = []
    results = []

    getfunc = {
        'info': get_info_reading,
        'power': get_power_reading,
    }[what]
    dumpfunc = {
        'info': dump_info_results,
        'power': dump_power_results,
    }[what]

    for n in range(POOL_SIZE):
        p = Process(target=run, args=(job_queue, res_queue, getfunc))
        p.start()
        pool.append(p)

    for n in range(len(machines)):
        res = res_queue.get()
        results.append(res)

        # In VERBOSE mode, abort immediately upon seeing a programming error.
        if VERBOSE and (
                isinstance(res.reading, MultiprocessingProgrammingError)):
            try:
                for p in pool:
                    p.terminate()
                    p.join()
            finally:
                raise res.reading

    assert len(machines) == len(results), (len(machines), len(results))
    for p in pool:
        p.terminate()  # bonus.. otherwise things stall sometimes
        p.join()
    # #print()

    dump_results(results, dumpfunc)


if __name__ == '__main__':
    IPMI_USER_PASS_BY_ALIAS, IPMI_USER_PASS_BY_IP = read_config_ipmikvm_dict()
    # #print(IPMI_USER_PASS_BY_ALIAS)
    # #print(IPMI_USER_PASS_BY_IP)
    if sys.argv[1:2] == ['-v']:
        VERBOSE = True
        sys.argv.pop(1)
    main(*sys.argv[1:])  # info|power machines.csv