Merge pull request #44756 from thatch45/mstats

Master Stats

conf/master

@@ -297,6 +297,11 @@
 #batch_safe_limit: 100
 #batch_safe_size: 8
 
+# Master stats enables stats events to be fired from the master at close
+# to the defined interval
+#master_stats: False
+#master_stats_event_iter: 60
+
 
 #####        Security settings       #####
 ##########################################

doc/ref/configuration/master.rst

@@ -868,6 +868,29 @@ what you are doing! Transports are explained in :ref:`Salt Transports
         ret_port: 4606
       zeromq: []
 
+.. conf_master:: master_stats
+
+``master_stats``
+----------------
+
+Default: False
+
+Turning on the master stats enables runtime throughput and statistics events
+to be fired from the master event bus. These events will report on what
+functions have been run on the master and how long these runs have, on
+average, taken over a given period of time.
+
+.. conf_master:: master_stats_event_iter
+
+``master_stats_event_iter``
+---------------------------
+
+Default: 60
+
+The time in seconds to fire master_stats events. This will only fire in
+conjunction with receiving a request to the master, idle masters will not
+fire these events.
+
 .. conf_master:: sock_pool_size
 
 ``sock_pool_size``

salt/config/__init__.py

@@ -165,6 +165,10 @@ VALID_OPTS = {
     # The master_pubkey_signature must also be set for this.
     'master_use_pubkey_signature': bool,
 
+    # Enable master stats eveents to be fired, these events will contain information about
+    # what commands the master is processing and what the rates are of the executions
+    'master_stats': bool,
+    'master_stats_event_iter': int,
     # The key fingerprint of the higher-level master for the syndic to verify it is talking to the
     # intended master
     'syndic_finger': str,
@@ -1515,6 +1519,8 @@ DEFAULT_MASTER_OPTS = {
     'svnfs_saltenv_whitelist': [],
     'svnfs_saltenv_blacklist': [],
     'max_event_size': 1048576,
+    'master_stats': False,
+    'master_stats_event_iter': 60,
     'minionfs_env': 'base',
     'minionfs_mountpoint': '',
     'minionfs_whitelist': [],

salt/master.py

@@ -16,6 +16,7 @@ import errno
 import signal
 import stat
 import logging
+import collections
 import multiprocessing
 import salt.serializers.msgpack
 
@@ -797,6 +798,7 @@ class MWorker(salt.utils.process.SignalHandlingMultiprocessingProcess):
         :return: Master worker
         '''
         kwargs[u'name'] = name
+        self.name = name
         super(MWorker, self).__init__(**kwargs)
         self.opts = opts
         self.req_channels = req_channels
@@ -804,6 +806,8 @@ class MWorker(salt.utils.process.SignalHandlingMultiprocessingProcess):
         self.mkey = mkey
         self.key = key
         self.k_mtime = 0
+        self.stats = collections.defaultdict(lambda: {'mean': 0, 'runs': 0})
+        self.stat_clock = time.time()
 
     # We need __setstate__ and __getstate__ to also pickle 'SMaster.secrets'.
     # Otherwise, 'SMaster.secrets' won't be copied over to the spawned process
@@ -879,6 +883,19 @@ class MWorker(salt.utils.process.SignalHandlingMultiprocessingProcess):
                u'clear': self._handle_clear}[key](load)
         raise tornado.gen.Return(ret)
 
+    def _post_stats(self, start, cmd):
+        '''
+        Calculate the master stats and fire events with stat info
+        '''
+        end = time.time()
+        duration = end - start
+        self.stats[cmd][u'mean'] = (self.stats[cmd][u'mean'] * (self.stats[cmd][u'runs'] - 1) + duration) / self.stats[cmd][u'runs']
+        if end - self.stat_clock > self.opts[u'master_stats_event_iter']:
+            # Fire the event with the stats and wipe the tracker
+            self.aes_funcs.event.fire_event({u'time': end - self.stat_clock, u'worker': self.name, u'stats': self.stats}, tagify(self.name, u'stats'))
+            self.stats = collections.defaultdict(lambda: {'mean': 0, 'runs': 0})
+            self.stat_clock = end
+
     def _handle_clear(self, load):
         '''
         Process a cleartext command
@@ -888,9 +905,16 @@ class MWorker(salt.utils.process.SignalHandlingMultiprocessingProcess):
                  the command specified in the load's 'cmd' key.
         '''
         log.trace(u'Clear payload received with command %s', load[u'cmd'])
-        if load[u'cmd'].startswith(u'__'):
+        cmd = load[u'cmd']
+        if cmd.startswith(u'__'):
             return False
-        return getattr(self.clear_funcs, load[u'cmd'])(load), {u'fun': u'send_clear'}
+        if self.opts[u'master_stats']:
+            start = time.time()
+            self.stats[cmd][u'runs'] += 1
+        ret = getattr(self.clear_funcs, cmd)(load), {u'fun': u'send_clear'}
+        if self.opts[u'master_stats']:
+            self._post_stats(start, cmd)
+        return ret
 
     def _handle_aes(self, data):
         '''
@@ -903,10 +927,17 @@ class MWorker(salt.utils.process.SignalHandlingMultiprocessingProcess):
         if u'cmd' not in data:
             log.error(u'Received malformed command %s', data)
             return {}
+        cmd = data[u'cmd']
         log.trace(u'AES payload received with command %s', data[u'cmd'])
-        if data[u'cmd'].startswith(u'__'):
+        if cmd.startswith(u'__'):
             return False
-        return self.aes_funcs.run_func(data[u'cmd'], data)
+        if self.opts[u'master_stats']:
+            start = time.time()
+            self.stats[cmd][u'runs'] += 1
+        ret = self.aes_funcs.run_func(data[u'cmd'], data)
+        if self.opts[u'master_stats']:
+            self._post_stats(start, cmd)
+        return ret
 
     def run(self):
         '''