mgr/iostat: implement 'ceph iostat' as a mgr plugin

doc/mgr/plugins.rst

@@ -51,13 +51,16 @@ like this::
         {
             "cmd": "foobar name=myarg,type=CephString",
             "desc": "Do something awesome",
-            "perm": "rw"
+            "perm": "rw",
+            # optional:
+            "poll": "true"
         }
     ]
 
 The ``cmd`` part of each entry is parsed in the same way as internal
 Ceph mon and admin socket commands (see mon/MonCommands.h in
-the Ceph source for examples)
+the Ceph source for examples). Note that the "poll" field is optional,
+and is set to False by default.
 
 Config settings
 ---------------

qa/tasks/mgr/test_module_selftest.py

@@ -42,6 +42,9 @@ def test_prometheus(self):
     def test_influx(self):
         self._selftest_plugin("influx")
 
+    def test_iostat(self):
+        self._selftest_plugin("iostat")
+
     def test_selftest_run(self):
         self._load_module("selftest")
         self.mgr_cluster.mon_manager.raw_cluster_cmd("mgr", "self-test", "run")

src/ceph.in

@@ -20,6 +20,7 @@ Foundation.  See file COPYING.
 """
 
 from __future__ import print_function
+from time import sleep
 import codecs
 import os
 import sys
@@ -42,6 +43,7 @@ CEPH_RELEASE_TYPE = "@CEPH_RELEASE_TYPE@"
 FLAG_NOFORWARD = (1 << 0)
 FLAG_OBSOLETE = (1 << 1)
 FLAG_DEPRECATED = (1 << 2)
+FLAG_POLL = (1 << 4)
 
 # priorities from src/common/perf_counters.h
 PRIO_CRITICAL = 10
@@ -317,6 +319,9 @@ def parse_cmdargs(args=None, target=''):
 
     parser.add_argument('--block', action='store_true',
                         help='block until completion (scrub and deep-scrub only)')
+    parser.add_argument('--period', '-p', default=1, type=float,
+                        help='polling period, default 1.0 second (for ' \
+                        'polling commands only)')
 
     # returns a Namespace with the parsed args, and a list of all extras
     parsed_args, extras = parser.parse_known_args(args)
@@ -515,6 +520,41 @@ else:
                 return line
 
 
+def do_command(parsed_args, target, cmdargs, sigdict, inbuf, verbose):
+    ''' Validate a command, and handle the polling flag '''
+
+    valid_dict = validate_command(sigdict, cmdargs, verbose)
+    # Validate input args against list of sigs
+    if valid_dict:
+        if parsed_args.output_format:
+            valid_dict['format'] = parsed_args.output_format
+        if verbose:
+            print("Submitting command: ", valid_dict, file=sys.stderr)
+    else:
+        return -errno.EINVAL, '', 'invalid command'
+
+    while True:
+        ret, outbuf, outs = json_command(cluster_handle, target=target, argdict=valid_dict,
+                inbuf=inbuf)
+        if 'poll' not in valid_dict or not valid_dict['poll']:
+            # Don't print here if it's not a polling command
+            break
+        if ret:
+            ret = abs(ret)
+            print('Error: {0} {1}'.format(ret, errno.errorcode.get(ret, 'Unknown')),
+                    file=sys.stderr)
+            break
+        if outbuf:
+            print(outbuf.decode('utf-8'))
+        if outs:
+            print(outs, file=sys.stderr)
+        if parsed_args.period <= 0:
+            break
+        sleep(parsed_args.period)
+
+    return ret, outbuf, outs
+
+
 def new_style_command(parsed_args, cmdargs, target, sigdict, inbuf, verbose):
     """
     Do new-style command dance.
@@ -531,14 +571,10 @@ def new_style_command(parsed_args, cmdargs, target, sigdict, inbuf, verbose):
 
     if True:
         if cmdargs:
-            # Validate input args against list of sigs
-            valid_dict = validate_command(sigdict, cmdargs, verbose)
-            if valid_dict:
-                if parsed_args.output_format:
-                    valid_dict['format'] = parsed_args.output_format
-            else:
-                return -errno.EINVAL, '', 'invalid command'
+            # Non interactive mode
+            ret, outbuf, outs = do_command(parsed_args, target, cmdargs, sigdict, inbuf, verbose)
         else:
+            # Interactive mode (ceph cli)
             if sys.stdin.isatty():
                 # do the command-interpreter looping
                 # for input to do readline cmd editing
@@ -563,30 +599,9 @@ def new_style_command(parsed_args, cmdargs, target, sigdict, inbuf, verbose):
                     print('Can not use \'tell\' in interactive mode.',
                           file=sys.stderr)
                     continue
-                valid_dict = validate_command(sigdict, cmdargs, verbose)
-                if valid_dict:
-                    if parsed_args.output_format:
-                        valid_dict['format'] = parsed_args.output_format
-                    if verbose:
-                        print("Submitting command: ", valid_dict, file=sys.stderr)
-                    ret, outbuf, outs = json_command(cluster_handle,
-                                                     target=target,
-                                                     argdict=valid_dict)
-                    if ret:
-                        ret = abs(ret)
-                        print('Error: {0} {1}'.format(ret, errno.errorcode.get(ret, 'Unknown')),
-                              file=sys.stderr)
-                    if outbuf:
-                        print(outbuf)
-                    if outs:
-                        print('Status:\n', outs, file=sys.stderr)
-                else:
-                    print("Invalid command", file=sys.stderr)
+                do_command(parsed_args, target, cmdargs, sigdict, inbuf, verbose)
 
-    if verbose:
-        print("Submitting command: ", valid_dict, file=sys.stderr)
-    return json_command(cluster_handle, target=target, argdict=valid_dict,
-                        inbuf=inbuf)
+    return ret, outbuf, outs
 
 
 def complete(sigdict, args, target):

src/mgr/ActivePyModules.cc

@@ -282,6 +282,14 @@ PyObject *ActivePyModules::get_python(const std::string &what)
         }
     );
     return f.get();
+  } else if (what == "io_rate") {
+    PyFormatter f;
+    cluster_state.with_pgmap(
+      [&f](const PGMap &pg_map) {
+        pg_map.dump_delta(&f);
+      }
+    );
+    return f.get();
   } else if (what == "df") {
     PyFormatter f;
 

src/mgr/PyModule.cc

@@ -26,6 +26,7 @@
 
 // Courtesy of http://stackoverflow.com/questions/1418015/how-to-get-python-exception-text
 #include <boost/python.hpp>
+#include <boost/algorithm/string/predicate.hpp>
 #include "include/assert.h"  // boost clobbers this
 // decode a Python exception into a string
 std::string handle_pyerror()
@@ -384,6 +385,15 @@ int PyModule::load_commands()
     assert(pPerm != nullptr);
     item.perm = PyString_AsString(pPerm);
 
+    item.polling = false;
+    PyObject *pPoll = PyDict_GetItemString(command, "poll");
+    if (pPoll) {
+      std::string polling = PyString_AsString(pPoll);
+      if (boost::iequals(polling, "true")) {
+        item.polling = true;
+      }
+    }
+
     item.module_name = module_name;
 
     commands.push_back(item);

src/mgr/PyModule.h

@@ -30,6 +30,7 @@ class ModuleCommand {
   std::string cmdstring;
   std::string helpstring;
   std::string perm;
+  bool polling;
 
   // Call the ActivePyModule of this name to handle the command
   std::string module_name;

src/mgr/PyModuleRegistry.cc

@@ -316,8 +316,12 @@ std::vector<MonCommand> PyModuleRegistry::get_commands() const
   std::vector<ModuleCommand> commands = get_py_commands();
   std::vector<MonCommand> result;
   for (auto &pyc: commands) {
+    uint64_t flags = MonCommand::FLAG_MGR;
+    if (pyc.polling) {
+      flags |= MonCommand::FLAG_POLL;
+    }
     result.push_back({pyc.cmdstring, pyc.helpstring, "mgr",
-                        pyc.perm, "cli", MonCommand::FLAG_MGR});
+                        pyc.perm, "cli", flags});
   }
   return result;
 }

src/mon/MonCommand.h

@@ -30,6 +30,7 @@ struct MonCommand {
   static const uint64_t FLAG_OBSOLETE   = 1 << 1;
   static const uint64_t FLAG_DEPRECATED = 1 << 2;
   static const uint64_t FLAG_MGR        = 1 << 3;
+  static const uint64_t FLAG_POLL       = 1 << 4;
 
   bool has_flag(uint64_t flag) const { return (flags & flag) != 0; }
   void set_flag(uint64_t flag) { flags |= flag; }

src/mon/MonCommands.h

@@ -112,6 +112,8 @@
  *  OBSOLETE  - command is considered obsolete
  *  DEPRECATED - command is considered deprecated
  *  MGR       - command goes to ceph-mgr (for luminous+)
+ *  POLL      - command is intended to be called periodically by the
+ *              client (see iostat)
  *
  * A command should always be first considered DEPRECATED before being
  * considered OBSOLETE, giving due consideration to users and conforming

src/mon/PGMap.cc

@@ -1395,6 +1395,7 @@ void PGMap::dump_delta(Formatter *f) const
 {
   f->open_object_section("pg_stats_delta");
   pg_sum_delta.dump(f);
+  f->dump_stream("stamp_delta") << stamp_delta;
   f->close_section();
 }
 

src/pybind/ceph_argparse.py

@@ -23,8 +23,9 @@
 import uuid
 
 
+# Flags are from MonCommand.h
 FLAG_MGR = 8   # command is intended for mgr
-
+FLAG_POLL = 16 # command is intended to be ran continuously by the client
 
 try:
     basestring
@@ -999,6 +1000,9 @@ def validate(args, signature, flags=0, partial=False):
     if flags & FLAG_MGR:
         d['target'] = ('mgr','')
 
+    if flags & FLAG_POLL:
+        d['poll'] = True
+
     # Finally, success
     return d
 
@@ -1239,7 +1243,7 @@ def send_command(cluster, target=('mon', ''), cmd=None, inbuf=b'', timeout=0,
                  verbose=False):
     """
     Send a command to a daemon using librados's
-    mon_command, osd_command, or pg_command.  Any bulk input data
+    mon_command, osd_command, mgr_command, or pg_command.  Any bulk input data
     comes in inbuf.
 
     Returns (ret, outbuf, outs); ret is the return code, outbuf is
@@ -1351,7 +1355,6 @@ def json_command(cluster, target=('mon', ''), prefix=None, argdict=None,
             except:
                 # use the target we were originally given
                 pass
-
         ret, outbuf, outs = send_command_retry(cluster,
                                                target, [json.dumps(cmddict)],
                                                inbuf, timeout, verbose)

src/pybind/mgr/iostat/__init__.py

@@ -0,0 +1 @@
+from .module import Module

src/pybind/mgr/iostat/module.py

@@ -0,0 +1,53 @@
+
+from mgr_module import MgrModule
+
+
+class Module(MgrModule):
+    COMMANDS = [
+        {
+            "cmd": "iostat",
+            "desc": "Get IO rates",
+            "perm": "r",
+            "poll": "true"
+        },
+        {
+            "cmd": "iostat self-test",
+            "desc": "Run a self test the iostat module",
+            "perm": "r"
+        }
+    ]
+
+
+    def __init__(self, *args, **kwargs):
+        super(Module, self).__init__(*args, **kwargs)
+
+
+    def handle_command(self, command):
+        rd = 0
+        wr = 0
+        ops = 0
+        ret = ''
+
+        if command['prefix'] == 'iostat':
+            r = self.get('io_rate')
+
+            stamp_delta = float(r['pg_stats_delta']['stamp_delta'])
+            if (stamp_delta > 0):
+                rd = int(r['pg_stats_delta']['stat_sum']['num_read_kb']) / stamp_delta
+                wr = int(r['pg_stats_delta']['stat_sum']['num_write_kb']) / stamp_delta
+                ops = ( int(r['pg_stats_delta']['stat_sum']['num_write']) + int(r['pg_stats_delta']['stat_sum']['num_read']) ) / stamp_delta
+
+            ret = "wr: {0} kB/s, rd: {1} kB/s, iops: {2}".format(int(wr), int(rd), int(ops))
+
+        elif command['prefix'] == 'iostat self-test':
+            r = self.get('io_rate')
+            assert('pg_stats_delta' in r)
+            assert('stamp_delta' in r['pg_stats_delta'])
+            assert('stat_sum' in r['pg_stats_delta'])
+            assert('num_read_kb' in r['pg_stats_delta']['stat_sum'])
+            assert('num_write_kb' in r['pg_stats_delta']['stat_sum'])
+            assert('num_write' in r['pg_stats_delta']['stat_sum'])
+            assert('num_read' in r['pg_stats_delta']['stat_sum'])
+            ret = 'iostat self-test OK'
+
+        return 0, '', ret

src/pybind/mgr/mgr_module.py

@@ -314,7 +314,7 @@ def get(self, data_name):
 
         :param str data_name: Valid things to fetch are osd_crush_map_text, 
                 osd_map, osd_map_tree, osd_map_crush, config, mon_map, fs_map,
-                osd_metadata, pg_summary, df, osd_stats, health, mon_status.
+                osd_metadata, pg_summary, io_rate, pg_dump, df, osd_stats, health, mon_status.
 
         Note:
             All these structures have their own JSON representations: experiment