[ceph.git] / ceph / src / tools / cephfs / top / cephfs-top

#!/usr/bin/python3

import argparse
import sys
import curses
import errno
import json
import signal
import time

from collections import OrderedDict
from datetime import datetime
from enum import Enum, unique
from threading import Event

import rados


class FSTopException(Exception):
    def __init__(self, msg=''):
        self.error_msg = msg

    def get_error_msg(self):
        return self.error_msg


@unique
class MetricType(Enum):
    METRIC_TYPE_NONE = 0
    METRIC_TYPE_PERCENTAGE = 1
    METRIC_TYPE_LATENCY = 2
    METRIC_TYPE_SIZE = 3


FS_TOP_PROG_STR = 'cephfs-top'

# version match b/w fstop and stats emitted by mgr/stats
FS_TOP_SUPPORTED_VER = 1

ITEMS_PAD_LEN = 1
ITEMS_PAD = " " * ITEMS_PAD_LEN
DEFAULT_REFRESH_INTERVAL = 1
# min refresh interval allowed
MIN_REFRESH_INTERVAL = 0.5

# metadata provided by mgr/stats
FS_TOP_MAIN_WINDOW_COL_CLIENT_ID = "client_id"
FS_TOP_MAIN_WINDOW_COL_MNT_ROOT = "mount_root"
FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR = "mount_point@host/addr"

MAIN_WINDOW_TOP_LINE_ITEMS_START = [ITEMS_PAD,
                                    FS_TOP_MAIN_WINDOW_COL_CLIENT_ID,
                                    FS_TOP_MAIN_WINDOW_COL_MNT_ROOT]
MAIN_WINDOW_TOP_LINE_ITEMS_END = [FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR]

# adjust this map according to stats version and maintain order
# as emitted by mgr/stast
MAIN_WINDOW_TOP_LINE_METRICS = OrderedDict([
    ("CAP_HIT", MetricType.METRIC_TYPE_PERCENTAGE),
    ("READ_LATENCY", MetricType.METRIC_TYPE_LATENCY),
    ("WRITE_LATENCY", MetricType.METRIC_TYPE_LATENCY),
    ("METADATA_LATENCY", MetricType.METRIC_TYPE_LATENCY),
    ("DENTRY_LEASE", MetricType.METRIC_TYPE_PERCENTAGE),
    ("OPENED_FILES", MetricType.METRIC_TYPE_NONE),
    ("PINNED_ICAPS", MetricType.METRIC_TYPE_NONE),
    ("OPENED_INODES", MetricType.METRIC_TYPE_NONE),
    ("READ_IO_SIZES", MetricType.METRIC_TYPE_SIZE),
    ("WRITE_IO_SIZES", MetricType.METRIC_TYPE_SIZE),
])
MGR_STATS_COUNTERS = list(MAIN_WINDOW_TOP_LINE_METRICS.keys())

FS_TOP_VERSION_HEADER_FMT = '{prog_name} - {now}'
FS_TOP_CLIENT_HEADER_FMT = 'Client(s): {num_clients} - {num_mounts} FUSE, '\
    '{num_kclients} kclient, {num_libs} libcephfs'

CLIENT_METADATA_KEY = "client_metadata"
CLIENT_METADATA_MOUNT_POINT_KEY = "mount_point"
CLIENT_METADATA_MOUNT_ROOT_KEY = "root"
CLIENT_METADATA_IP_KEY = "IP"
CLIENT_METADATA_HOSTNAME_KEY = "hostname"
CLIENT_METADATA_VALID_METRICS_KEY = "valid_metrics"

GLOBAL_METRICS_KEY = "global_metrics"
GLOBAL_COUNTERS_KEY = "global_counters"

last_time = time.time()
last_read_size = {}
last_write_size = {}


def calc_perc(c):
    if c[0] == 0 and c[1] == 0:
        return 0.0
    return round((c[0] / (c[0] + c[1])) * 100, 2)


def calc_lat(c):
    return round(c[0] + c[1] / 1000000000, 2)


# in MB
def calc_size(c):
    return round(c[1] / (1024 * 1024), 2)


# in MB
def calc_avg_size(c):
    if c[0] == 0:
        return 0.0
    return round(c[1] / (c[0] * 1024 * 1024), 2)


# in MB/s
def calc_speed(size, duration):
    if duration == 0:
        return 0.0
    return round(size / (duration * 1024 * 1024), 2)


def wrap(s, sl):
    """return a '+' suffixed wrapped string"""
    if len(s) < sl:
        return s
    return f'{s[0:sl-1]}+'


class FSTop(object):
    def __init__(self, args):
        self.rados = None
        self.stdscr = None  # curses instance
        self.client_name = args.id
        self.cluster_name = args.cluster
        self.conffile = args.conffile
        self.refresh_interval_secs = args.delay
        self.exit_ev = Event()

    def refresh_window_size(self):
        self.height, self.width = self.stdscr.getmaxyx()

    def handle_signal(self, signum, _):
        self.exit_ev.set()

    def init(self):
        try:
            if self.conffile:
                r_rados = rados.Rados(rados_id=self.client_name, clustername=self.cluster_name,
                                      conffile=self.conffile)
            else:
                r_rados = rados.Rados(rados_id=self.client_name, clustername=self.cluster_name)
            r_rados.conf_read_file()
            r_rados.connect()
            self.rados = r_rados
        except rados.Error as e:
            if e.errno == errno.ENOENT:
                raise FSTopException(f'cluster {self.cluster_name} does not exist')
            else:
                raise FSTopException(f'error connecting to cluster: {e}')
        self.verify_perf_stats_support()
        signal.signal(signal.SIGTERM, self.handle_signal)
        signal.signal(signal.SIGINT, self.handle_signal)

    def fini(self):
        if self.rados:
            self.rados.shutdown()
            self.rados = None

    def selftest(self):
        stats_json = self.perf_stats_query()
        if not stats_json['version'] == FS_TOP_SUPPORTED_VER:
            raise FSTopException('perf stats version mismatch!')
        missing = [m for m in stats_json["global_counters"] if m.upper() not in MGR_STATS_COUNTERS]
        if missing:
            raise FSTopException('Cannot handle unknown metrics from \'ceph fs perf stats\': '
                                 f'{missing}')

    def setup_curses(self, win):
        self.stdscr = win
        curses.use_default_colors()
        curses.start_color()
        try:
            curses.curs_set(0)
        except curses.error:
            # If the terminal do not support the visibility
            # requested it will raise an exception
            pass
        self.run_display()

    def verify_perf_stats_support(self):
        mon_cmd = {'prefix': 'mgr module ls', 'format': 'json'}
        try:
            ret, buf, out = self.rados.mon_command(json.dumps(mon_cmd), b'')
        except Exception as e:
            raise FSTopException(f'error checking \'stats\' module: {e}')
        if ret != 0:
            raise FSTopException(f'error checking \'stats\' module: {out}')
        if 'stats' not in json.loads(buf.decode('utf-8'))['enabled_modules']:
            raise FSTopException('\'stats\' module not enabled. Use \'ceph mgr module '
                                 'enable stats\' to enable')

    def perf_stats_query(self):
        mgr_cmd = {'prefix': 'fs perf stats', 'format': 'json'}
        try:
            ret, buf, out = self.rados.mgr_command(json.dumps(mgr_cmd), b'')
        except Exception as e:
            raise FSTopException(f'error in \'perf stats\' query: {e}')
        if ret != 0:
            raise FSTopException(f'error in \'perf stats\' query: {out}')
        return json.loads(buf.decode('utf-8'))

    def items(self, item):
        if item == "CAP_HIT":
            return "chit"
        if item == "READ_LATENCY":
            return "rlat"
        if item == "WRITE_LATENCY":
            return "wlat"
        if item == "METADATA_LATENCY":
            return "mlat"
        if item == "DENTRY_LEASE":
            return "dlease"
        if item == "OPENED_FILES":
            return "ofiles"
        if item == "PINNED_ICAPS":
            return "oicaps"
        if item == "OPENED_INODES":
            return "oinodes"
        if item == "READ_IO_SIZES":
            return "rtio"
        if item == "WRITE_IO_SIZES":
            return "wtio"
        else:
            # return empty string for none type
            return ''

    def mtype(self, typ):
        if typ == MetricType.METRIC_TYPE_PERCENTAGE:
            return "(%)"
        elif typ == MetricType.METRIC_TYPE_LATENCY:
            return "(s)"
        elif typ == MetricType.METRIC_TYPE_SIZE:
            return "(MB)"
        else:
            # return empty string for none type
            return ''

    def avg_items(self, item):
        if item == "READ_IO_SIZES":
            return "raio"
        if item == "WRITE_IO_SIZES":
            return "waio"
        else:
            # return empty string for none type
            return ''

    def speed_items(self, item):
        if item == "READ_IO_SIZES":
            return "rsp"
        if item == "WRITE_IO_SIZES":
            return "wsp"
        else:
            # return empty string for none type
            return ''

    def speed_mtype(self, typ):
        if typ == MetricType.METRIC_TYPE_SIZE:
            return "(MB/s)"
        else:
            # return empty string for none type
            return ''

    def refresh_top_line_and_build_coord(self):
        if self.topl is None:
            return

        xp = 0
        x_coord_map = {}

        heading = []
        for item in MAIN_WINDOW_TOP_LINE_ITEMS_START:
            heading.append(item)
            nlen = len(item) + len(ITEMS_PAD)
            x_coord_map[item] = (xp, nlen)
            xp += nlen

        for item, typ in MAIN_WINDOW_TOP_LINE_METRICS.items():
            it = f'{self.items(item)}{self.mtype(typ)}'
            heading.append(it)
            nlen = len(it) + len(ITEMS_PAD)
            x_coord_map[item] = (xp, nlen)
            xp += nlen

            if item == "READ_IO_SIZES" or item == "WRITE_IO_SIZES":
                # average io sizes
                it = f'{self.avg_items(item)}{self.mtype(typ)}'
                heading.append(it)
                nlen = len(it) + len(ITEMS_PAD)
                if item == "READ_IO_SIZES":
                    x_coord_map["READ_IO_AVG"] = (xp, nlen)
                if item == "WRITE_IO_SIZES":
                    x_coord_map["WRITE_IO_AVG"] = (xp, nlen)
                xp += nlen

                # io speeds
                it = f'{self.speed_items(item)}{self.speed_mtype(typ)}'
                heading.append(it)
                nlen = len(it) + len(ITEMS_PAD)
                if item == "READ_IO_SIZES":
                    x_coord_map["READ_IO_SPEED"] = (xp, nlen)
                if item == "WRITE_IO_SIZES":
                    x_coord_map["WRITE_IO_SPEED"] = (xp, nlen)
                xp += nlen

        for item in MAIN_WINDOW_TOP_LINE_ITEMS_END:
            heading.append(item)
            nlen = len(item) + len(ITEMS_PAD)
            x_coord_map[item] = (xp, nlen)
            xp += nlen
        title = ITEMS_PAD.join(heading)
        hlen = min(self.width - 2, len(title))
        self.topl.addnstr(0, 0, title, hlen, curses.A_STANDOUT | curses.A_BOLD)
        self.topl.refresh()
        return x_coord_map

    @staticmethod
    def has_metric(metadata, metrics_key):
        return metrics_key in metadata

    @staticmethod
    def has_metrics(metadata, metrics_keys):
        for key in metrics_keys:
            if not FSTop.has_metric(metadata, key):
                return False
        return True

    def refresh_client(self, client_id, metrics, counters, client_meta, x_coord_map, y_coord):
        global last_time
        cur_time = time.time()
        duration = cur_time - last_time
        last_time = cur_time
        remaining_hlen = self.width - 1
        for item in MAIN_WINDOW_TOP_LINE_ITEMS_START:
            coord = x_coord_map[item]
            hlen = coord[1] - len(ITEMS_PAD)
            hlen = min(hlen, remaining_hlen)
            if remaining_hlen < coord[1]:
                remaining_hlen = 0
            else:
                remaining_hlen -= coord[1]
            if item == FS_TOP_MAIN_WINDOW_COL_CLIENT_ID:
                self.mainw.addnstr(y_coord, coord[0],
                                   wrap(client_id.split('.')[1], hlen),
                                   hlen)
            elif item == FS_TOP_MAIN_WINDOW_COL_MNT_ROOT:
                if FSTop.has_metric(client_meta, CLIENT_METADATA_MOUNT_ROOT_KEY):
                    self.mainw.addnstr(y_coord, coord[0],
                                       wrap(client_meta[CLIENT_METADATA_MOUNT_ROOT_KEY], hlen),
                                       hlen)
                else:
                    self.mainw.addnstr(y_coord, coord[0], "N/A", hlen)

            if remaining_hlen == 0:
                return

        cidx = 0
        client_id = x_coord_map[FS_TOP_MAIN_WINDOW_COL_CLIENT_ID]
        for item in counters:
            coord = x_coord_map[item]
            hlen = coord[1] - len(ITEMS_PAD)
            hlen = min(hlen, remaining_hlen)
            if remaining_hlen < coord[1]:
                remaining_hlen = 0
            else:
                remaining_hlen -= coord[1]
            m = metrics[cidx]
            key = MGR_STATS_COUNTERS[cidx]
            typ = MAIN_WINDOW_TOP_LINE_METRICS[key]
            if item.lower() in client_meta.get(CLIENT_METADATA_VALID_METRICS_KEY, []):
                if typ == MetricType.METRIC_TYPE_PERCENTAGE:
                    self.mainw.addnstr(y_coord, coord[0], f'{calc_perc(m)}', hlen)
                elif typ == MetricType.METRIC_TYPE_LATENCY:
                    self.mainw.addnstr(y_coord, coord[0], f'{calc_lat(m)}', hlen)
                elif typ == MetricType.METRIC_TYPE_SIZE:
                    self.mainw.addnstr(y_coord, coord[0], f'{calc_size(m)}', hlen)

                    # average io sizes
                    if remaining_hlen == 0:
                        return
                    if key == "READ_IO_SIZES":
                        coord = x_coord_map["READ_IO_AVG"]
                    elif key == "WRITE_IO_SIZES":
                        coord = x_coord_map["WRITE_IO_AVG"]
                    hlen = coord[1] - len(ITEMS_PAD)
                    hlen = min(hlen, remaining_hlen)
                    if remaining_hlen < coord[1]:
                        remaining_hlen = 0
                    else:
                        remaining_hlen -= coord[1]
                    self.mainw.addnstr(y_coord, coord[0], f'{calc_avg_size(m)}', hlen)

                    # io speeds
                    if remaining_hlen == 0:
                        return
                    if key == "READ_IO_SIZES":
                        coord = x_coord_map["READ_IO_SPEED"]
                    elif key == "WRITE_IO_SIZES":
                        coord = x_coord_map["WRITE_IO_SPEED"]
                    hlen = coord[1] - len(ITEMS_PAD)
                    hlen = min(hlen, remaining_hlen)
                    if remaining_hlen < coord[1]:
                        remaining_hlen = 0
                    else:
                        remaining_hlen -= coord[1]
                    if key == "READ_IO_SIZES":
                        global last_read_size
                        last_size = last_read_size.get(client_id, 0)
                        size = m[1] - last_size
                        last_read_size[client_id] = m[1]
                    if key == "WRITE_IO_SIZES":
                        global last_write_size
                        last_size = last_write_size.get(client_id, 0)
                        size = m[1] - last_size
                        last_write_size[client_id] = m[1]
                    self.mainw.addnstr(y_coord, coord[0],
                                       f'{calc_speed(size, duration)}',
                                       hlen)
                else:
                    # display 0th element from metric tuple
                    self.mainw.addnstr(y_coord, coord[0], f'{m[0]}', hlen)
            else:
                self.mainw.addnstr(y_coord, coord[0], "N/A", hlen)
            cidx += 1

            if remaining_hlen == 0:
                return

        for item in MAIN_WINDOW_TOP_LINE_ITEMS_END:
            coord = x_coord_map[item]
            hlen = coord[1] - len(ITEMS_PAD)
            # always place the FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR in the
            # last, it will be a very long string to display
            if item == FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR:
                if FSTop.has_metrics(client_meta, [CLIENT_METADATA_MOUNT_POINT_KEY,
                                                   CLIENT_METADATA_HOSTNAME_KEY,
                                                   CLIENT_METADATA_IP_KEY]):
                    self.mainw.addnstr(y_coord, coord[0],
                                       f'{client_meta[CLIENT_METADATA_MOUNT_POINT_KEY]}@'
                                       f'{client_meta[CLIENT_METADATA_HOSTNAME_KEY]}/'
                                       f'{client_meta[CLIENT_METADATA_IP_KEY]}',
                                       remaining_hlen)
                else:
                    self.mainw.addnstr(y_coord, coord[0], "N/A", remaining_hlen)
            hlen = min(hlen, remaining_hlen)
            if remaining_hlen < coord[1]:
                remaining_hlen = 0
            else:
                remaining_hlen -= coord[1]
            if remaining_hlen == 0:
                return

    def refresh_clients(self, x_coord_map, stats_json):
        counters = [m.upper() for m in stats_json[GLOBAL_COUNTERS_KEY]]
        y_coord = 0
        for client_id, metrics in stats_json[GLOBAL_METRICS_KEY].items():
            self.refresh_client(client_id,
                                metrics,
                                counters,
                                stats_json[CLIENT_METADATA_KEY][client_id],
                                x_coord_map,
                                y_coord)
            y_coord += 1

    def refresh_main_window(self, x_coord_map, stats_json):
        if self.mainw is None:
            return
        self.refresh_clients(x_coord_map, stats_json)
        self.mainw.refresh()

    def refresh_header(self, stats_json):
        hlen = self.width - 2
        if not stats_json['version'] == FS_TOP_SUPPORTED_VER:
            self.header.addnstr(0, 0, 'perf stats version mismatch!', hlen)
            return False
        client_metadata = stats_json[CLIENT_METADATA_KEY]
        num_clients = len(client_metadata)
        num_mounts = len([client for client, metadata in client_metadata.items() if
                          CLIENT_METADATA_MOUNT_POINT_KEY in metadata
                          and metadata[CLIENT_METADATA_MOUNT_POINT_KEY] != 'N/A'])
        num_kclients = len([client for client, metadata in client_metadata.items() if
                            "kernel_version" in metadata])
        num_libs = num_clients - (num_mounts + num_kclients)
        now = datetime.now().ctime()
        self.header.addnstr(0, 0,
                            FS_TOP_VERSION_HEADER_FMT.format(prog_name=FS_TOP_PROG_STR, now=now),
                            hlen, curses.A_STANDOUT | curses.A_BOLD)
        self.header.addnstr(1, 0, FS_TOP_CLIENT_HEADER_FMT.format(num_clients=num_clients,
                                                                  num_mounts=num_mounts,
                                                                  num_kclients=num_kclients,
                                                                  num_libs=num_libs), hlen)
        self.header.refresh()
        return True

    def run_display(self):
        while not self.exit_ev.is_set():
            # use stdscr.clear() instead of clearing each window
            # to avoid screen blinking.
            self.stdscr.clear()
            self.refresh_window_size()
            if self.width <= 2 or self.width <= 2:
                self.exit_ev.wait(timeout=self.refresh_interval_secs)
                continue

            # coordinate constants for windowing -- (height, width, y, x)
            # NOTE: requires initscr() call before accessing COLS, LINES.
            try:
                HEADER_WINDOW_COORD = (2, self.width - 1, 0, 0)
                self.header = curses.newwin(*HEADER_WINDOW_COORD)
                if self.height >= 3:
                    TOPLINE_WINDOW_COORD = (1, self.width - 1, 3, 0)
                    self.topl = curses.newwin(*TOPLINE_WINDOW_COORD)
                else:
                    self.topl = None
                if self.height >= 5:
                    MAIN_WINDOW_COORD = (self.height - 4, self.width - 1, 4, 0)
                    self.mainw = curses.newwin(*MAIN_WINDOW_COORD)
                else:
                    self.mainw = None
            except curses.error:
                # this may happen when creating the sub windows the
                # terminal window size changed, just retry it
                continue

            stats_json = self.perf_stats_query()
            try:
                if self.refresh_header(stats_json):
                    x_coord_map = self.refresh_top_line_and_build_coord()
                    self.refresh_main_window(x_coord_map, stats_json)
                self.exit_ev.wait(timeout=self.refresh_interval_secs)
            except curses.error:
                # this may happen when addstr the terminal window
                # size changed, just retry it
                pass


if __name__ == '__main__':
    def float_greater_than(x):
        value = float(x)
        if value < MIN_REFRESH_INTERVAL:
            raise argparse.ArgumentTypeError(
                f'Refresh interval should be greater than or equal to {MIN_REFRESH_INTERVAL}')
        return value

    parser = argparse.ArgumentParser(description='Ceph Filesystem top utility')
    parser.add_argument('--cluster', nargs='?', const='ceph', default='ceph',
                        help='Ceph cluster to connect (default: ceph)')
    parser.add_argument('--id', nargs='?', const='fstop', default='fstop',
                        help='Ceph user to use to connection (default: fstop)')
    parser.add_argument('--conffile', nargs='?', default=None,
                        help='Path to cluster configuration file')
    parser.add_argument('--selftest', dest='selftest', action='store_true',
                        help='Run in selftest mode')
    parser.add_argument('-d', '--delay', nargs='?', default=DEFAULT_REFRESH_INTERVAL,
                        type=float_greater_than, help='Interval to refresh data '
                        f'(default: {DEFAULT_REFRESH_INTERVAL})')

    args = parser.parse_args()
    err = False
    ft = FSTop(args)
    try:
        ft.init()
        if args.selftest:
            ft.selftest()
            sys.stdout.write("selftest ok\n")
        else:
            curses.wrapper(ft.setup_curses)
    except FSTopException as fst:
        err = True
        sys.stderr.write(f'{fst.get_error_msg()}\n')
    except Exception as e:
        err = True
        sys.stderr.write(f'exception: {e}\n')
    finally:
        ft.fini()
    sys.exit(0 if not err else -1)
Commit	Line	Data
f67539c2 TL	1	#!/usr/bin/python3
	2
	3	import argparse
	4	import sys
	5	import curses
	6	import errno
	7	import json
	8	import signal
a4b75251	9	import time
f67539c2 TL	10
	11	from collections import OrderedDict
	12	from datetime import datetime
	13	from enum import Enum, unique
	14	from threading import Event
	15
	16	import rados
	17
	18
	19	class FSTopException(Exception):
	20	def __init__(self, msg=''):
	21	self.error_msg = msg
	22
	23	def get_error_msg(self):
	24	return self.error_msg
	25
	26
	27	@unique
	28	class MetricType(Enum):
	29	METRIC_TYPE_NONE = 0
	30	METRIC_TYPE_PERCENTAGE = 1
	31	METRIC_TYPE_LATENCY = 2
a4b75251	32	METRIC_TYPE_SIZE = 3
f67539c2 TL	33
	34
	35	FS_TOP_PROG_STR = 'cephfs-top'
	36
	37	# version match b/w fstop and stats emitted by mgr/stats
	38	FS_TOP_SUPPORTED_VER = 1
	39
	40	ITEMS_PAD_LEN = 1
b3b6e05e	41	ITEMS_PAD = " " * ITEMS_PAD_LEN
f67539c2 TL	42	DEFAULT_REFRESH_INTERVAL = 1
	43	# min refresh interval allowed
	44	MIN_REFRESH_INTERVAL = 0.5
	45
	46	# metadata provided by mgr/stats
b3b6e05e TL	47	FS_TOP_MAIN_WINDOW_COL_CLIENT_ID = "client_id"
	48	FS_TOP_MAIN_WINDOW_COL_MNT_ROOT = "mount_root"
	49	FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR = "mount_point@host/addr"
f67539c2 TL	50
	51	MAIN_WINDOW_TOP_LINE_ITEMS_START = [ITEMS_PAD,
	52	FS_TOP_MAIN_WINDOW_COL_CLIENT_ID,
	53	FS_TOP_MAIN_WINDOW_COL_MNT_ROOT]
	54	MAIN_WINDOW_TOP_LINE_ITEMS_END = [FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR]
	55
	56	# adjust this map according to stats version and maintain order
	57	# as emitted by mgr/stast
	58	MAIN_WINDOW_TOP_LINE_METRICS = OrderedDict([
	59	("CAP_HIT", MetricType.METRIC_TYPE_PERCENTAGE),
	60	("READ_LATENCY", MetricType.METRIC_TYPE_LATENCY),
	61	("WRITE_LATENCY", MetricType.METRIC_TYPE_LATENCY),
	62	("METADATA_LATENCY", MetricType.METRIC_TYPE_LATENCY),
	63	("DENTRY_LEASE", MetricType.METRIC_TYPE_PERCENTAGE),
	64	("OPENED_FILES", MetricType.METRIC_TYPE_NONE),
	65	("PINNED_ICAPS", MetricType.METRIC_TYPE_NONE),
	66	("OPENED_INODES", MetricType.METRIC_TYPE_NONE),
a4b75251 TL	67	("READ_IO_SIZES", MetricType.METRIC_TYPE_SIZE),
a4b75251 TL	68	("WRITE_IO_SIZES", MetricType.METRIC_TYPE_SIZE),
f67539c2 TL	69	])
	70	MGR_STATS_COUNTERS = list(MAIN_WINDOW_TOP_LINE_METRICS.keys())
	71
	72	FS_TOP_VERSION_HEADER_FMT = '{prog_name} - {now}'
	73	FS_TOP_CLIENT_HEADER_FMT = 'Client(s): {num_clients} - {num_mounts} FUSE, '\
	74	'{num_kclients} kclient, {num_libs} libcephfs'
	75
	76	CLIENT_METADATA_KEY = "client_metadata"
	77	CLIENT_METADATA_MOUNT_POINT_KEY = "mount_point"
	78	CLIENT_METADATA_MOUNT_ROOT_KEY = "root"
	79	CLIENT_METADATA_IP_KEY = "IP"
	80	CLIENT_METADATA_HOSTNAME_KEY = "hostname"
	81	CLIENT_METADATA_VALID_METRICS_KEY = "valid_metrics"
	82
	83	GLOBAL_METRICS_KEY = "global_metrics"
	84	GLOBAL_COUNTERS_KEY = "global_counters"
	85
a4b75251 TL	86	last_time = time.time()
	87	last_read_size = {}
	88	last_write_size = {}
	89
f67539c2 TL	90
	91	def calc_perc(c):
	92	if c[0] == 0 and c[1] == 0:
	93	return 0.0
	94	return round((c[0] / (c[0] + c[1])) * 100, 2)
	95
	96
	97	def calc_lat(c):
	98	return round(c[0] + c[1] / 1000000000, 2)
	99
	100
a4b75251 TL	101	# in MB
	102	def calc_size(c):
	103	return round(c[1] / (1024 * 1024), 2)
	104
	105
	106	# in MB
	107	def calc_avg_size(c):
	108	if c[0] == 0:
	109	return 0.0
	110	return round(c[1] / (c[0] * 1024 * 1024), 2)
	111
	112
	113	# in MB/s
	114	def calc_speed(size, duration):
	115	if duration == 0:
	116	return 0.0
	117	return round(size / (duration * 1024 * 1024), 2)
	118
	119
f67539c2 TL	120	def wrap(s, sl):
	121	"""return a '+' suffixed wrapped string"""
	122	if len(s) < sl:
	123	return s
	124	return f'{s[0:sl-1]}+'
	125
	126
	127	class FSTop(object):
	128	def __init__(self, args):
	129	self.rados = None
	130	self.stdscr = None # curses instance
	131	self.client_name = args.id
	132	self.cluster_name = args.cluster
	133	self.conffile = args.conffile
	134	self.refresh_interval_secs = args.delay
	135	self.exit_ev = Event()
	136
b3b6e05e TL	137	def refresh_window_size(self):
	138	self.height, self.width = self.stdscr.getmaxyx()
	139
f67539c2 TL	140	def handle_signal(self, signum, _):
	141	self.exit_ev.set()
	142
	143	def init(self):
	144	try:
	145	if self.conffile:
	146	r_rados = rados.Rados(rados_id=self.client_name, clustername=self.cluster_name,
	147	conffile=self.conffile)
	148	else:
	149	r_rados = rados.Rados(rados_id=self.client_name, clustername=self.cluster_name)
	150	r_rados.conf_read_file()
	151	r_rados.connect()
	152	self.rados = r_rados
	153	except rados.Error as e:
	154	if e.errno == errno.ENOENT:
	155	raise FSTopException(f'cluster {self.cluster_name} does not exist')
	156	else:
	157	raise FSTopException(f'error connecting to cluster: {e}')
	158	self.verify_perf_stats_support()
	159	signal.signal(signal.SIGTERM, self.handle_signal)
	160	signal.signal(signal.SIGINT, self.handle_signal)
	161
	162	def fini(self):
	163	if self.rados:
	164	self.rados.shutdown()
	165	self.rados = None
	166
	167	def selftest(self):
	168	stats_json = self.perf_stats_query()
	169	if not stats_json['version'] == FS_TOP_SUPPORTED_VER:
	170	raise FSTopException('perf stats version mismatch!')
	171	missing = [m for m in stats_json["global_counters"] if m.upper() not in MGR_STATS_COUNTERS]
	172	if missing:
	173	raise FSTopException('Cannot handle unknown metrics from \'ceph fs perf stats\': '
	174	f'{missing}')
	175
b3b6e05e TL	176	def setup_curses(self, win):
	177	self.stdscr = win
	178	curses.use_default_colors()
	179	curses.start_color()
	180	try:
	181	curses.curs_set(0)
	182	except curses.error:
	183	# If the terminal do not support the visibility
	184	# requested it will raise an exception
	185	pass
	186	self.run_display()
f67539c2 TL	187
	188	def verify_perf_stats_support(self):
	189	mon_cmd = {'prefix': 'mgr module ls', 'format': 'json'}
	190	try:
	191	ret, buf, out = self.rados.mon_command(json.dumps(mon_cmd), b'')
	192	except Exception as e:
	193	raise FSTopException(f'error checking \'stats\' module: {e}')
	194	if ret != 0:
	195	raise FSTopException(f'error checking \'stats\' module: {out}')
	196	if 'stats' not in json.loads(buf.decode('utf-8'))['enabled_modules']:
	197	raise FSTopException('\'stats\' module not enabled. Use \'ceph mgr module '
	198	'enable stats\' to enable')
	199
	200	def perf_stats_query(self):
	201	mgr_cmd = {'prefix': 'fs perf stats', 'format': 'json'}
	202	try:
	203	ret, buf, out = self.rados.mgr_command(json.dumps(mgr_cmd), b'')
	204	except Exception as e:
	205	raise FSTopException(f'error in \'perf stats\' query: {e}')
	206	if ret != 0:
	207	raise FSTopException(f'error in \'perf stats\' query: {out}')
	208	return json.loads(buf.decode('utf-8'))
	209
b3b6e05e TL	210	def items(self, item):
	211	if item == "CAP_HIT":
	212	return "chit"
	213	if item == "READ_LATENCY":
	214	return "rlat"
	215	if item == "WRITE_LATENCY":
	216	return "wlat"
	217	if item == "METADATA_LATENCY":
	218	return "mlat"
	219	if item == "DENTRY_LEASE":
	220	return "dlease"
	221	if item == "OPENED_FILES":
	222	return "ofiles"
	223	if item == "PINNED_ICAPS":
	224	return "oicaps"
	225	if item == "OPENED_INODES":
	226	return "oinodes"
a4b75251 TL	227	if item == "READ_IO_SIZES":
	228	return "rtio"
	229	if item == "WRITE_IO_SIZES":
	230	return "wtio"
b3b6e05e TL	231	else:
	232	# return empty string for none type
	233	return ''
	234
f67539c2 TL	235	def mtype(self, typ):
	236	if typ == MetricType.METRIC_TYPE_PERCENTAGE:
	237	return "(%)"
	238	elif typ == MetricType.METRIC_TYPE_LATENCY:
	239	return "(s)"
a4b75251 TL	240	elif typ == MetricType.METRIC_TYPE_SIZE:
	241	return "(MB)"
	242	else:
	243	# return empty string for none type
	244	return ''
	245
	246	def avg_items(self, item):
	247	if item == "READ_IO_SIZES":
	248	return "raio"
	249	if item == "WRITE_IO_SIZES":
	250	return "waio"
	251	else:
	252	# return empty string for none type
	253	return ''
	254
	255	def speed_items(self, item):
	256	if item == "READ_IO_SIZES":
	257	return "rsp"
	258	if item == "WRITE_IO_SIZES":
	259	return "wsp"
	260	else:
	261	# return empty string for none type
	262	return ''
	263
	264	def speed_mtype(self, typ):
	265	if typ == MetricType.METRIC_TYPE_SIZE:
	266	return "(MB/s)"
f67539c2 TL	267	else:
	268	# return empty string for none type
	269	return ''
	270
	271	def refresh_top_line_and_build_coord(self):
b3b6e05e TL	272	if self.topl is None:
	273	return
	274
f67539c2 TL	275	xp = 0
	276	x_coord_map = {}
	277
	278	heading = []
	279	for item in MAIN_WINDOW_TOP_LINE_ITEMS_START:
	280	heading.append(item)
	281	nlen = len(item) + len(ITEMS_PAD)
	282	x_coord_map[item] = (xp, nlen)
	283	xp += nlen
	284
	285	for item, typ in MAIN_WINDOW_TOP_LINE_METRICS.items():
b3b6e05e	286	it = f'{self.items(item)}{self.mtype(typ)}'
f67539c2 TL	287	heading.append(it)
	288	nlen = len(it) + len(ITEMS_PAD)
	289	x_coord_map[item] = (xp, nlen)
	290	xp += nlen
	291
a4b75251 TL	292	if item == "READ_IO_SIZES" or item == "WRITE_IO_SIZES":
	293	# average io sizes
	294	it = f'{self.avg_items(item)}{self.mtype(typ)}'
	295	heading.append(it)
	296	nlen = len(it) + len(ITEMS_PAD)
	297	if item == "READ_IO_SIZES":
	298	x_coord_map["READ_IO_AVG"] = (xp, nlen)
	299	if item == "WRITE_IO_SIZES":
	300	x_coord_map["WRITE_IO_AVG"] = (xp, nlen)
	301	xp += nlen
	302
	303	# io speeds
	304	it = f'{self.speed_items(item)}{self.speed_mtype(typ)}'
	305	heading.append(it)
	306	nlen = len(it) + len(ITEMS_PAD)
	307	if item == "READ_IO_SIZES":
	308	x_coord_map["READ_IO_SPEED"] = (xp, nlen)
	309	if item == "WRITE_IO_SIZES":
	310	x_coord_map["WRITE_IO_SPEED"] = (xp, nlen)
	311	xp += nlen
	312
f67539c2 TL	313	for item in MAIN_WINDOW_TOP_LINE_ITEMS_END:
	314	heading.append(item)
	315	nlen = len(item) + len(ITEMS_PAD)
	316	x_coord_map[item] = (xp, nlen)
	317	xp += nlen
b3b6e05e TL	318	title = ITEMS_PAD.join(heading)
	319	hlen = min(self.width - 2, len(title))
	320	self.topl.addnstr(0, 0, title, hlen, curses.A_STANDOUT \| curses.A_BOLD)
	321	self.topl.refresh()
f67539c2 TL	322	return x_coord_map
	323
	324	@staticmethod
	325	def has_metric(metadata, metrics_key):
	326	return metrics_key in metadata
	327
	328	@staticmethod
	329	def has_metrics(metadata, metrics_keys):
	330	for key in metrics_keys:
	331	if not FSTop.has_metric(metadata, key):
	332	return False
	333	return True
	334
	335	def refresh_client(self, client_id, metrics, counters, client_meta, x_coord_map, y_coord):
a4b75251 TL	336	global last_time
	337	cur_time = time.time()
	338	duration = cur_time - last_time
	339	last_time = cur_time
b3b6e05e	340	remaining_hlen = self.width - 1
f67539c2 TL	341	for item in MAIN_WINDOW_TOP_LINE_ITEMS_START:
	342	coord = x_coord_map[item]
	343	hlen = coord[1] - len(ITEMS_PAD)
b3b6e05e TL	344	hlen = min(hlen, remaining_hlen)
	345	if remaining_hlen < coord[1]:
	346	remaining_hlen = 0
	347	else:
	348	remaining_hlen -= coord[1]
f67539c2	349	if item == FS_TOP_MAIN_WINDOW_COL_CLIENT_ID:
b3b6e05e TL	350	self.mainw.addnstr(y_coord, coord[0],
	351	wrap(client_id.split('.')[1], hlen),
	352	hlen)
f67539c2 TL	353	elif item == FS_TOP_MAIN_WINDOW_COL_MNT_ROOT:
f67539c2 TL	354	if FSTop.has_metric(client_meta, CLIENT_METADATA_MOUNT_ROOT_KEY):
b3b6e05e TL	355	self.mainw.addnstr(y_coord, coord[0],
	356	wrap(client_meta[CLIENT_METADATA_MOUNT_ROOT_KEY], hlen),
	357	hlen)
f67539c2	358	else:
b3b6e05e TL	359	self.mainw.addnstr(y_coord, coord[0], "N/A", hlen)
	360
	361	if remaining_hlen == 0:
	362	return
	363
f67539c2	364	cidx = 0
a4b75251	365	client_id = x_coord_map[FS_TOP_MAIN_WINDOW_COL_CLIENT_ID]
f67539c2 TL	366	for item in counters:
f67539c2 TL	367	coord = x_coord_map[item]
b3b6e05e TL	368	hlen = coord[1] - len(ITEMS_PAD)
	369	hlen = min(hlen, remaining_hlen)
	370	if remaining_hlen < coord[1]:
	371	remaining_hlen = 0
	372	else:
	373	remaining_hlen -= coord[1]
f67539c2	374	m = metrics[cidx]
a4b75251 TL	375	key = MGR_STATS_COUNTERS[cidx]
a4b75251 TL	376	typ = MAIN_WINDOW_TOP_LINE_METRICS[key]
f67539c2 TL	377	if item.lower() in client_meta.get(CLIENT_METADATA_VALID_METRICS_KEY, []):
f67539c2 TL	378	if typ == MetricType.METRIC_TYPE_PERCENTAGE:
b3b6e05e	379	self.mainw.addnstr(y_coord, coord[0], f'{calc_perc(m)}', hlen)
f67539c2	380	elif typ == MetricType.METRIC_TYPE_LATENCY:
b3b6e05e	381	self.mainw.addnstr(y_coord, coord[0], f'{calc_lat(m)}', hlen)
a4b75251 TL	382	elif typ == MetricType.METRIC_TYPE_SIZE:
	383	self.mainw.addnstr(y_coord, coord[0], f'{calc_size(m)}', hlen)
	384
	385	# average io sizes
	386	if remaining_hlen == 0:
	387	return
	388	if key == "READ_IO_SIZES":
	389	coord = x_coord_map["READ_IO_AVG"]
	390	elif key == "WRITE_IO_SIZES":
	391	coord = x_coord_map["WRITE_IO_AVG"]
	392	hlen = coord[1] - len(ITEMS_PAD)
	393	hlen = min(hlen, remaining_hlen)
	394	if remaining_hlen < coord[1]:
	395	remaining_hlen = 0
	396	else:
	397	remaining_hlen -= coord[1]
	398	self.mainw.addnstr(y_coord, coord[0], f'{calc_avg_size(m)}', hlen)
	399
	400	# io speeds
	401	if remaining_hlen == 0:
	402	return
	403	if key == "READ_IO_SIZES":
	404	coord = x_coord_map["READ_IO_SPEED"]
	405	elif key == "WRITE_IO_SIZES":
	406	coord = x_coord_map["WRITE_IO_SPEED"]
	407	hlen = coord[1] - len(ITEMS_PAD)
	408	hlen = min(hlen, remaining_hlen)
	409	if remaining_hlen < coord[1]:
	410	remaining_hlen = 0
	411	else:
	412	remaining_hlen -= coord[1]
	413	if key == "READ_IO_SIZES":
	414	global last_read_size
	415	last_size = last_read_size.get(client_id, 0)
	416	size = m[1] - last_size
	417	last_read_size[client_id] = m[1]
	418	if key == "WRITE_IO_SIZES":
	419	global last_write_size
	420	last_size = last_write_size.get(client_id, 0)
	421	size = m[1] - last_size
	422	last_write_size[client_id] = m[1]
	423	self.mainw.addnstr(y_coord, coord[0],
	424	f'{calc_speed(size, duration)}',
	425	hlen)
f67539c2 TL	426	else:
f67539c2 TL	427	# display 0th element from metric tuple
b3b6e05e	428	self.mainw.addnstr(y_coord, coord[0], f'{m[0]}', hlen)
f67539c2	429	else:
b3b6e05e	430	self.mainw.addnstr(y_coord, coord[0], "N/A", hlen)
f67539c2 TL	431	cidx += 1
f67539c2 TL	432
b3b6e05e TL	433	if remaining_hlen == 0:
	434	return
	435
	436	for item in MAIN_WINDOW_TOP_LINE_ITEMS_END:
	437	coord = x_coord_map[item]
	438	hlen = coord[1] - len(ITEMS_PAD)
	439	# always place the FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR in the
	440	# last, it will be a very long string to display
	441	if item == FS_TOP_MAIN_WINDOW_COL_MNTPT_HOST_ADDR:
	442	if FSTop.has_metrics(client_meta, [CLIENT_METADATA_MOUNT_POINT_KEY,
	443	CLIENT_METADATA_HOSTNAME_KEY,
	444	CLIENT_METADATA_IP_KEY]):
	445	self.mainw.addnstr(y_coord, coord[0],
	446	f'{client_meta[CLIENT_METADATA_MOUNT_POINT_KEY]}@'
	447	f'{client_meta[CLIENT_METADATA_HOSTNAME_KEY]}/'
	448	f'{client_meta[CLIENT_METADATA_IP_KEY]}',
	449	remaining_hlen)
	450	else:
	451	self.mainw.addnstr(y_coord, coord[0], "N/A", remaining_hlen)
	452	hlen = min(hlen, remaining_hlen)
	453	if remaining_hlen < coord[1]:
	454	remaining_hlen = 0
	455	else:
	456	remaining_hlen -= coord[1]
	457	if remaining_hlen == 0:
	458	return
	459
f67539c2 TL	460	def refresh_clients(self, x_coord_map, stats_json):
	461	counters = [m.upper() for m in stats_json[GLOBAL_COUNTERS_KEY]]
	462	y_coord = 0
	463	for client_id, metrics in stats_json[GLOBAL_METRICS_KEY].items():
	464	self.refresh_client(client_id,
	465	metrics,
	466	counters,
	467	stats_json[CLIENT_METADATA_KEY][client_id],
	468	x_coord_map,
	469	y_coord)
	470	y_coord += 1
	471
	472	def refresh_main_window(self, x_coord_map, stats_json):
b3b6e05e TL	473	if self.mainw is None:
b3b6e05e TL	474	return
f67539c2	475	self.refresh_clients(x_coord_map, stats_json)
b3b6e05e	476	self.mainw.refresh()
f67539c2 TL	477
f67539c2 TL	478	def refresh_header(self, stats_json):
b3b6e05e	479	hlen = self.width - 2
f67539c2	480	if not stats_json['version'] == FS_TOP_SUPPORTED_VER:
b3b6e05e	481	self.header.addnstr(0, 0, 'perf stats version mismatch!', hlen)
f67539c2 TL	482	return False
	483	client_metadata = stats_json[CLIENT_METADATA_KEY]
	484	num_clients = len(client_metadata)
	485	num_mounts = len([client for client, metadata in client_metadata.items() if
	486	CLIENT_METADATA_MOUNT_POINT_KEY in metadata
	487	and metadata[CLIENT_METADATA_MOUNT_POINT_KEY] != 'N/A'])
	488	num_kclients = len([client for client, metadata in client_metadata.items() if
	489	"kernel_version" in metadata])
	490	num_libs = num_clients - (num_mounts + num_kclients)
	491	now = datetime.now().ctime()
b3b6e05e TL	492	self.header.addnstr(0, 0,
	493	FS_TOP_VERSION_HEADER_FMT.format(prog_name=FS_TOP_PROG_STR, now=now),
	494	hlen, curses.A_STANDOUT \| curses.A_BOLD)
	495	self.header.addnstr(1, 0, FS_TOP_CLIENT_HEADER_FMT.format(num_clients=num_clients,
	496	num_mounts=num_mounts,
	497	num_kclients=num_kclients,
	498	num_libs=num_libs), hlen)
	499	self.header.refresh()
f67539c2 TL	500	return True
f67539c2 TL	501
b3b6e05e	502	def run_display(self):
f67539c2	503	while not self.exit_ev.is_set():
b3b6e05e TL	504	# use stdscr.clear() instead of clearing each window
	505	# to avoid screen blinking.
	506	self.stdscr.clear()
	507	self.refresh_window_size()
	508	if self.width <= 2 or self.width <= 2:
	509	self.exit_ev.wait(timeout=self.refresh_interval_secs)
	510	continue
	511
	512	# coordinate constants for windowing -- (height, width, y, x)
	513	# NOTE: requires initscr() call before accessing COLS, LINES.
	514	try:
	515	HEADER_WINDOW_COORD = (2, self.width - 1, 0, 0)
	516	self.header = curses.newwin(*HEADER_WINDOW_COORD)
	517	if self.height >= 3:
	518	TOPLINE_WINDOW_COORD = (1, self.width - 1, 3, 0)
	519	self.topl = curses.newwin(*TOPLINE_WINDOW_COORD)
	520	else:
	521	self.topl = None
	522	if self.height >= 5:
	523	MAIN_WINDOW_COORD = (self.height - 4, self.width - 1, 4, 0)
	524	self.mainw = curses.newwin(*MAIN_WINDOW_COORD)
	525	else:
	526	self.mainw = None
	527	except curses.error:
	528	# this may happen when creating the sub windows the
	529	# terminal window size changed, just retry it
	530	continue
	531
f67539c2	532	stats_json = self.perf_stats_query()
b3b6e05e TL	533	try:
	534	if self.refresh_header(stats_json):
	535	x_coord_map = self.refresh_top_line_and_build_coord()
	536	self.refresh_main_window(x_coord_map, stats_json)
	537	self.exit_ev.wait(timeout=self.refresh_interval_secs)
	538	except curses.error:
	539	# this may happen when addstr the terminal window
	540	# size changed, just retry it
	541	pass
f67539c2 TL	542
	543
	544	if __name__ == '__main__':
	545	def float_greater_than(x):
	546	value = float(x)
	547	if value < MIN_REFRESH_INTERVAL:
20effc67 TL	548	raise argparse.ArgumentTypeError(
20effc67 TL	549	f'Refresh interval should be greater than or equal to {MIN_REFRESH_INTERVAL}')
f67539c2 TL	550	return value
	551
	552	parser = argparse.ArgumentParser(description='Ceph Filesystem top utility')
	553	parser.add_argument('--cluster', nargs='?', const='ceph', default='ceph',
20effc67	554	help='Ceph cluster to connect (default: ceph)')
f67539c2 TL	555	parser.add_argument('--id', nargs='?', const='fstop', default='fstop',
	556	help='Ceph user to use to connection (default: fstop)')
	557	parser.add_argument('--conffile', nargs='?', default=None,
	558	help='Path to cluster configuration file')
	559	parser.add_argument('--selftest', dest='selftest', action='store_true',
20effc67	560	help='Run in selftest mode')
f67539c2 TL	561	parser.add_argument('-d', '--delay', nargs='?', default=DEFAULT_REFRESH_INTERVAL,
	562	type=float_greater_than, help='Interval to refresh data '
	563	f'(default: {DEFAULT_REFRESH_INTERVAL})')
	564
	565	args = parser.parse_args()
	566	err = False
	567	ft = FSTop(args)
	568	try:
	569	ft.init()
	570	if args.selftest:
	571	ft.selftest()
	572	sys.stdout.write("selftest ok\n")
	573	else:
b3b6e05e	574	curses.wrapper(ft.setup_curses)
f67539c2 TL	575	except FSTopException as fst:
	576	err = True
	577	sys.stderr.write(f'{fst.get_error_msg()}\n')
	578	except Exception as e:
	579	err = True
	580	sys.stderr.write(f'exception: {e}\n')
	581	finally:
	582	ft.fini()
	583	sys.exit(0 if not err else -1)