verify-loader

#!/usr/bin/env python3
#
# ViaQ logging load generator
#
# Copyright 2021 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

import sys
import time
import argparse
import signal
import io
import fcntl
from collections import defaultdict
try:
    from systemd import journal
except ImportError:
    journal_available = False
else:
    journal_available = True

F_SETPIPE_SZ = 1031  # Linux 2.6.35+
F_GETPIPE_SZ = 1032  # Linux 2.6.35+

REPORT_INTERVAL = 10
MB = 1024 * 1024


# If running verify-loader as a background process, we can still do:
#   kill -INT ${pid}
#   kill -TERM ${pid}
# to make it exit the main loop and print the statistics.
def handle_sig_term(signum, frame):
    raise KeyboardInterrupt


class Context(object):

    def __init__(self):
        self.prev = None
        self.count = 0
        self.bytes = 0
        self.report_count = 0
        self.report_bytes = 0
        self.duplicates = 0
        self.skips = 0
        self.report_skips = 0
        self.report_duplicates = 0

    def msg(self, seq, length):
        self.report_count += 1
        self.count += 1
        self.report_bytes += length
        self.bytes += length

        if self.prev is None:
            self.prev = seq
            return None
        elif seq == (self.prev + 1):
            # Normal and expected code path
            self.prev = seq
            return None
        else:
            ret_prev = self.prev
            if seq <= self.prev:
                self.duplicates += 1
                self.report_duplicates += 1
            else:
                assert seq > (self.prev + 1), (
                    "Logic bomb! Should not be possible"
                )
                self.skips += 1
                self.report_skips += 1
                # Since the sequence jumped ahead, save the new value as the
                # previous in order to be sure to stay with the jump.
                self.prev = seq
            return ret_prev

    def report(self):
        if self.report_count > 0:
            self.count += self.report_count
            self.bytes += self.report_bytes
            self.report_count = 0
            self.report_bytes = 0
            self.report_skips = 0
            self.report_duplicates = 0
            ret_val = True
        else:
            ret_val = False
        return ret_val


def print_stats(invocid, ctx, payload):
    now = time.time()
    try:
        stats, _ = payload.rsplit(b' ', 1)
        rawvals = stats.strip()[:-1].split(b' ')
        vals = []
        for val in rawvals:
            if val:
                vals.append(val)
        if vals[1][-1] == 115:
            timestamp, statseq = float(vals[1][:-1]), int(vals[2])
            if statseq != ctx.prev:
                raise Exception(
                    f"Statistics sequence number, {statseq:d}, doest not"
                    f" match sequence from log record itself, {ctx.prev:d}"
                )
        else:
            raise Exception(
                "Invalid floating point timestamp value encountered:"
                f" {vals[1]}"
            )
        if len(vals) == 7:
            lclrate, gblrate = float(vals[3]), float(vals[4])
            skp, skr = int(vals[5]), int(vals[6])
        elif len(vals) == 10:
            lclrate, gblrate = float(vals[3]), float(vals[7])
            skp, skr = int(vals[8]), int(vals[9])
        else:
            raise Exception("Logic bomb!")
    except Exception as exc:
        print(
            f"Error parsing payload statistics: {exc!r}, payload: {payload!r}",
            file=sys.stderr,
        )
        sys.exit(1)
    else:
        offset = now - timestamp
        print(
            f"verify-loader - loader stats ({invocid}): {now:.2f}s"
            f" ({offset:4.2f}s) {lclrate:12.3f}"
            f" {gblrate:12.3f} {statseq:d} {ctx.count:d} {ctx.skips:d}"
            f" {ctx.duplicates:d} {skp:d} {skr:d}",
            flush=True
        )


def verify(
    input_gen, report_interval=REPORT_INTERVAL, emit_loader_stats=False,
    debug=False
):
    signal.signal(signal.SIGTERM, handle_sig_term)

    ret_val = 0

    ignored_bytes = 0
    ignored_count = 0

    report_bytes = 0
    report_ignored_bytes = 0
    report_ignored_count = 0

    contexts = defaultdict(Context)
    start = time.time()
    report_start = start
    rptperiod = (report_interval, start)

    try:
        for line in input_gen:
            line_len = len(line)
            if not line.startswith(b"loader seq - "):
                if debug:
                    print(f"verify-loader -- ignored {line[:80]}...")
                report_ignored_bytes += line_len
                report_ignored_count += 1
            else:
                try:
                    _, invocid, seqval, payload = line.split(b'-', 4)
                except Exception:
                    report_ignored_bytes += line_len
                    report_ignored_count += 1
                else:
                    try:
                        seq = int(seqval)
                    except Exception:
                        report_ignored_bytes += line_len
                        report_ignored_count += 1
                    else:
                        report_bytes += line_len
                        invocid = invocid.strip()
                        ctx = contexts[invocid]
                        prev = ctx.msg(seq, line_len)
                        if debug and (prev is not None):
                            # Bad record encountered, flag it
                            print(
                                f"verify-loader -- {invocid}: bad record,"
                                f" current sequence #: {seq:d}, previous"
                                f" sequence #: {prev:d}",
                                flush=True
                            )
                        if emit_loader_stats and payload.startswith(
                            b" (stats:"
                        ):
                            print_stats(invocid, ctx, payload)
            now = time.time()
            if now >= rptperiod[1]:
                rptperiod = (
                    rptperiod[0], rptperiod[1] + rptperiod[0]
                )

                ignored_bytes += report_ignored_bytes
                ignored_count += report_ignored_count

                total_bytes = 0
                total_count = 0
                total_skips = 0
                total_dupes = 0
                report_count = 0
                report_skips = 0
                report_dupes = 0

                for invocid, ctx in contexts.items():
                    report_count += ctx.report_count
                    report_skips += ctx.report_skips
                    report_dupes += ctx.report_duplicates
                    if ctx.report():
                        print(
                            f"verify-loader - loader {invocid}: {ctx.count:d}"
                            f" {ctx.skips:d} {ctx.duplicates:d}",
                            flush=True
                        )
                    total_bytes += ctx.bytes
                    total_count += ctx.count
                    total_skips += ctx.skips
                    total_dupes += ctx.duplicates

                print(
                    "verify-loader interval %.3f - %.3f :"
                    " read rate: %.3f MB/sec, %.3f /sec, %d skips,"
                    " %d duplicates "
                    "(ignored %.3f MB/sec, %.3f /sec); "
                    "overall read rate: %.3f MB/sec, %.3f /sec, %d skips,"
                    " %d duplicates "
                    "(ignored %.3f MB/sec, %.3f /sec)"
                    % (
                        report_start,
                        now,
                        (report_bytes / MB) / (now - report_start),
                        (report_count / (now - report_start)),
                        report_skips,
                        report_dupes,
                        (report_ignored_bytes / MB) / (now - report_start),
                        (report_ignored_count / (now - report_start)),
                        (total_bytes / MB) / (now - start),
                        (total_count / (now - start)),
                        total_skips,
                        total_dupes,
                        (ignored_bytes / MB) / (now - start),
                        (ignored_count / (now - start)),
                    ),
                    flush=True
                )
                report_bytes = 0
                report_ignored_bytes = 0
                report_ignored_count = 0
                # In case outputing the report takes a significant amount of
                # time, we reset the check for the next report here.
                report_start = time.time()
    except KeyboardInterrupt:
        pass
    finally:
        now = time.time()
        ignored_bytes += report_ignored_bytes
        ignored_count += report_ignored_count
        total_bytes = 0
        total_count = 0
        tot_skips = 0
        tot_dupes = 0
        for invocid, ctx in contexts.items():
            ctx.report()
            total_bytes += ctx.bytes
            total_count += ctx.count
            tot_skips += ctx.skips
            tot_dupes += ctx.duplicates
        if ignored_count + total_count > 0:
            for invocid, ctx in contexts.items():
                print(
                    f"verify-loader - loader {invocid}: {ctx.count:d}"
                    f" {ctx.skips:d} {ctx.duplicates:d}"
                )
            print(
                "verify-loader overall %.3f - %.3f read rate: %.3f MB/sec,"
                " %.3f /sec, %d skips, %d duplicates "
                "(ignored %.3f MB/sec, %.3f /sec)"
                % (
                    start,
                    now,
                    (total_bytes / MB) / (now - start),
                    (total_count / (now - start)),
                    tot_skips,
                    tot_dupes,
                    (ignored_bytes / MB) / (now - start),
                    (ignored_count / (now - start)),
                )
            )
        if tot_skips + tot_dupes > 0:
            ret_val = 1
    return ret_val


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Message payload generator.')
    parser.add_argument(
        'file', metavar='FILE', nargs='?',
        default='-',
        help='file to read, if empty, stdin is used'
    )
    parser.add_argument(
        '--report-interval', metavar='INTERVAL', dest='reportint', type=int,
        default=REPORT_INTERVAL,
        help=(
            'the # seconds between reports'
            f' (defaults to {REPORT_INTERVAL} seconds)'
        )
    )
    parser.add_argument(
        '--emit-loader-stats', dest='emitloaderstats', action="store_true",
        help=(
            'emit statistics embedded in logs from the loader'
            ' (defaults to False)'
        )
    )
    parser.add_argument(
        '--read-journal', dest='readjournal', action="store_true",
        help='read directly from the systemd journal (defaults to False)'
    )
    parser.add_argument(
        '--debug', dest='debug', action="store_true",
        help='Turn on debugging (defaults to False)'
    )
    args = parser.parse_args()

    if args.readjournal and not journal_available:
        print(
            "systemd journal module not available, most likely running using"
            " pypy3",
            file=sys.stderr
        )
        sys.exit(1)

    if args.file == '-':
        if args.readjournal:
            j = journal.Reader()
            # The fileno() method ends up creating the inotify file descriptor
            # which in turn prevents this client from leaking open journal log
            # files.
            j.fileno()
            j.seek_tail()
            j.get_previous()

            def jreader():
                while True:
                    for entry in j:
                        yield entry['MESSAGE'].encode('utf-8')
                    # We have to wait in a loop here because the journal code
                    # runs as C code, where Python exception handling is not
                    # in play.
                    ret = journal.NOP
                    while ret == journal.NOP:
                        ret = j.wait(0.1)
            input_gen = jreader()
        else:
            r_fd = sys.stdin.fileno()
            try:
                buffer_size = fcntl.fcntl(r_fd, F_GETPIPE_SZ)
            except OSError:
                # Default when not a pipe.
                buffer_size = 65536
            assert buffer_size >= 4096, (
                f"Expected pipe size to be >= 4096, was {buffer_size!r}"
            )
            fo = io.FileIO(r_fd)
            input_gen = io.BufferedReader(fo, buffer_size=buffer_size)
    else:
        if args.readjournal:
            print(
                "*** Warning *** ignoring request to read from the systemd"
                " journal (--read-journal) since we have an actual file"
                f" argument provided ({args.file}) ..."
            )
        input_gen = open(args.file, 'rb', 1)
    sys.exit(
        verify(input_gen, args.reportint, args.emitloaderstats, args.debug)
    )