ceph/src/pybind/mgr/zabbix/module.py

   1 """
   2 Zabbix module for ceph-mgr
   3
   4 Collect statistics from Ceph cluster and every X seconds send data to a Zabbix
   5 server using the zabbix_sender executable.
   6 """
   7 import json
   8 import errno
   9 from subprocess import Popen, PIPE
  10 from threading import Event
  11 from mgr_module import MgrModule
  12
  13
  14 def avg(data):
  15     if len(data):
  16         return sum(data) / float(len(data))
  17     else:
  18         return 0
  19
  20
  21 class ZabbixSender(object):
  22     def __init__(self, sender, host, port, log):
  23         self.sender = sender
  24         self.host = host
  25         self.port = port
  26         self.log = log
  27
  28     def send(self, hostname, data):
  29         if len(data) == 0:
  30             return
  31
  32         cmd = [self.sender, '-z', self.host, '-p', str(self.port), '-s',
  33                hostname, '-vv', '-i', '-']
  34
  35         self.log.debug('Executing: %s', cmd)
  36
  37         proc = Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE)
  38
  39         for key, value in data.items():
  40             proc.stdin.write('{0} ceph.{1} {2}\n'.format(hostname, key, value))
  41
  42         stdout, stderr = proc.communicate()
  43         if proc.returncode != 0:
  44             raise RuntimeError('%s exited non-zero: %s' % (self.sender,
  45                                                            stderr))
  46
  47         self.log.debug('Zabbix Sender: %s', stdout.rstrip())
  48
  49
  50 class Module(MgrModule):
  51     run = False
  52     config = dict()
  53     ceph_health_mapping = {'HEALTH_OK': 0, 'HEALTH_WARN': 1, 'HEALTH_ERR': 2}
  54
  55     @property
  56     def config_keys(self):
  57         return dict((o['name'], o.get('default', None))
  58                 for o in self.MODULE_OPTIONS)
  59
  60     MODULE_OPTIONS = [
  61             {
  62                 'name': 'zabbix_sender',
  63                 'default': '/usr/bin/zabbix_sender'
  64             },
  65             {
  66                 'name': 'zabbix_host',
  67                 'default': None
  68             },
  69             {
  70                 'name': 'zabbix_port',
  71                 'type': 'int',
  72                 'default': 10051
  73             },
  74             {
  75                 'name': 'identifier',
  76                 'default': ""
  77             },
  78             {
  79                 'name': 'interval',
  80                 'type': 'secs',
  81                 'default': 60
  82             }
  83     ]
  84
  85     COMMANDS = [
  86         {
  87             "cmd": "zabbix config-set name=key,type=CephString "
  88                    "name=value,type=CephString",
  89             "desc": "Set a configuration value",
  90             "perm": "rw"
  91         },
  92         {
  93             "cmd": "zabbix config-show",
  94             "desc": "Show current configuration",
  95             "perm": "r"
  96         },
  97         {
  98             "cmd": "zabbix send",
  99             "desc": "Force sending data to Zabbix",
 100             "perm": "rw"
 101         },
 102     ]
 103
 104     def __init__(self, *args, **kwargs):
 105         super(Module, self).__init__(*args, **kwargs)
 106         self.event = Event()
 107
 108     def init_module_config(self):
 109         self.fsid = self.get('mon_map')['fsid']
 110         self.log.debug('Found Ceph fsid %s', self.fsid)
 111
 112         for key, default in self.config_keys.items():
 113             self.set_config_option(key, self.get_module_option(key, default))
 114
 115     def set_config_option(self, option, value):
 116         if option not in self.config_keys.keys():
 117             raise RuntimeError('{0} is a unknown configuration '
 118                                'option'.format(option))
 119
 120         if option in ['zabbix_port', 'interval']:
 121             try:
 122                 value = int(value)
 123             except (ValueError, TypeError):
 124                 raise RuntimeError('invalid {0} configured. Please specify '
 125                                    'a valid integer'.format(option))
 126
 127         if option == 'interval' and value < 10:
 128             raise RuntimeError('interval should be set to at least 10 seconds')
 129
 130         self.log.debug('Setting in-memory config option %s to: %s', option,
 131                        value)
 132         self.config[option] = value
 133         return True
 134
 135     def get_pg_stats(self):
 136         stats = dict()
 137
 138         pg_states = ['active', 'peering', 'clean', 'scrubbing', 'undersized',
 139                      'backfilling', 'recovering', 'degraded', 'inconsistent',
 140                      'remapped', 'backfill_toofull', 'wait_backfill',
 141                      'recovery_wait']
 142
 143         for state in pg_states:
 144             stats['num_pg_{0}'.format(state)] = 0
 145
 146         pg_status = self.get('pg_status')
 147
 148         stats['num_pg'] = pg_status['num_pgs']
 149
 150         for state in pg_status['pgs_by_state']:
 151             states = state['state_name'].split('+')
 152             for s in pg_states:
 153                 key = 'num_pg_{0}'.format(s)
 154                 if s in states:
 155                     stats[key] += state['count']
 156
 157         return stats
 158
 159     def get_data(self):
 160         data = dict()
 161
 162         health = json.loads(self.get('health')['json'])
 163         # 'status' is luminous+, 'overall_status' is legacy mode.
 164         data['overall_status'] = health.get('status',
 165                                             health.get('overall_status'))
 166         data['overall_status_int'] = \
 167             self.ceph_health_mapping.get(data['overall_status'])
 168
 169         mon_status = json.loads(self.get('mon_status')['json'])
 170         data['num_mon'] = len(mon_status['monmap']['mons'])
 171
 172         df = self.get('df')
 173         data['num_pools'] = len(df['pools'])
 174         data['total_used_bytes'] = df['stats']['total_used_bytes']
 175         data['total_bytes'] = df['stats']['total_bytes']
 176         data['total_avail_bytes'] = df['stats']['total_avail_bytes']
 177
 178         wr_ops = 0
 179         rd_ops = 0
 180         wr_bytes = 0
 181         rd_bytes = 0
 182
 183         for pool in df['pools']:
 184             wr_ops += pool['stats']['wr']
 185             rd_ops += pool['stats']['rd']
 186             wr_bytes += pool['stats']['wr_bytes']
 187             rd_bytes += pool['stats']['rd_bytes']
 188
 189         data['wr_ops'] = wr_ops
 190         data['rd_ops'] = rd_ops
 191         data['wr_bytes'] = wr_bytes
 192         data['rd_bytes'] = rd_bytes
 193
 194         osd_map = self.get('osd_map')
 195         data['num_osd'] = len(osd_map['osds'])
 196         data['osd_nearfull_ratio'] = osd_map['nearfull_ratio']
 197         data['osd_full_ratio'] = osd_map['full_ratio']
 198         data['osd_backfillfull_ratio'] = osd_map['backfillfull_ratio']
 199
 200         data['num_pg_temp'] = len(osd_map['pg_temp'])
 201
 202         num_up = 0
 203         num_in = 0
 204         for osd in osd_map['osds']:
 205             if osd['up'] == 1:
 206                 num_up += 1
 207
 208             if osd['in'] == 1:
 209                 num_in += 1
 210
 211         data['num_osd_up'] = num_up
 212         data['num_osd_in'] = num_in
 213
 214         osd_fill = list()
 215         osd_pgs = list()
 216         osd_apply_latency_ns = list()
 217         osd_commit_latency_ns = list()
 218
 219         osd_stats = self.get('osd_stats')
 220         for osd in osd_stats['osd_stats']:
 221             if osd['kb'] == 0:
 222                 continue
 223             osd_fill.append((float(osd['kb_used']) / float(osd['kb'])) * 100)
 224             osd_pgs.append(osd['num_pgs'])
 225             osd_apply_latency_ns.append(osd['perf_stat']['apply_latency_ns'])
 226             osd_commit_latency_ns.append(osd['perf_stat']['commit_latency_ns'])
 227
 228         try:
 229             data['osd_max_fill'] = max(osd_fill)
 230             data['osd_min_fill'] = min(osd_fill)
 231             data['osd_avg_fill'] = avg(osd_fill)
 232             data['osd_max_pgs'] = max(osd_pgs)
 233             data['osd_min_pgs'] = min(osd_pgs)
 234             data['osd_avg_pgs'] = avg(osd_pgs)
 235         except ValueError:
 236             pass
 237
 238         try:
 239             data['osd_latency_apply_max'] = max(osd_apply_latency_ns) / 1000000.0 # ns -> ms
 240             data['osd_latency_apply_min'] = min(osd_apply_latency_ns) / 1000000.0 # ns -> ms
 241             data['osd_latency_apply_avg'] = avg(osd_apply_latency_ns) / 1000000.0 # ns -> ms
 242
 243             data['osd_latency_commit_max'] = max(osd_commit_latency_ns) / 1000000.0 # ns -> ms
 244             data['osd_latency_commit_min'] = min(osd_commit_latency_ns) / 1000000.0 # ns -> ms
 245             data['osd_latency_commit_avg'] = avg(osd_commit_latency_ns) / 1000000.0 # ns -> ms
 246         except ValueError:
 247             pass
 248
 249         data.update(self.get_pg_stats())
 250
 251         return data
 252
 253     def send(self):
 254         data = self.get_data()
 255
 256         identifier = self.config['identifier']
 257         if identifier is None or len(identifier) == 0:
 258             identifier = 'ceph-{0}'.format(self.fsid)
 259
 260         if not self.config['zabbix_host']:
 261             self.log.error('Zabbix server not set, please configure using: '
 262                            'ceph zabbix config-set zabbix_host <zabbix_host>')
 263             self.set_health_checks({
 264                 'MGR_ZABBIX_NO_SERVER': {
 265                     'severity': 'warning',
 266                     'summary': 'No Zabbix server configured',
 267                     'detail': ['Configuration value zabbix_host not configured']
 268                 }
 269             })
 270             return
 271
 272         try:
 273             self.log.info(
 274                 'Sending data to Zabbix server %s as host/identifier %s',
 275                 self.config['zabbix_host'], identifier)
 276             self.log.debug(data)
 277
 278             zabbix = ZabbixSender(self.config['zabbix_sender'],
 279                                   self.config['zabbix_host'],
 280                                   self.config['zabbix_port'], self.log)
 281
 282             zabbix.send(identifier, data)
 283             self.set_health_checks(dict())
 284             return True
 285         except Exception as exc:
 286             self.log.error('Exception when sending: %s', exc)
 287             self.set_health_checks({
 288                 'MGR_ZABBIX_SEND_FAILED': {
 289                     'severity': 'warning',
 290                     'summary': 'Failed to send data to Zabbix',
 291                     'detail': [str(exc)]
 292                 }
 293             })
 294
 295         return False
 296
 297     def handle_command(self, inbuf, command):
 298         if command['prefix'] == 'zabbix config-show':
 299             return 0, json.dumps(self.config), ''
 300         elif command['prefix'] == 'zabbix config-set':
 301             key = command['key']
 302             value = command['value']
 303             if not value:
 304                 return -errno.EINVAL, '', 'Value should not be empty or None'
 305
 306             self.log.debug('Setting configuration option %s to %s', key, value)
 307             if self.set_config_option(key, value):
 308                 self.set_module_option(key, value)
 309                 return 0, 'Configuration option {0} updated'.format(key), ''
 310
 311             return 1,\
 312                 'Failed to update configuration option {0}'.format(key), ''
 313
 314         elif command['prefix'] == 'zabbix send':
 315             if self.send():
 316                 return 0, 'Sending data to Zabbix', ''
 317
 318             return 1, 'Failed to send data to Zabbix', ''
 319         else:
 320             return (-errno.EINVAL, '',
 321                     "Command not found '{0}'".format(command['prefix']))
 322
 323     def shutdown(self):
 324         self.log.info('Stopping zabbix')
 325         self.run = False
 326         self.event.set()
 327
 328     def serve(self):
 329         self.log.info('Zabbix module starting up')
 330         self.run = True
 331
 332         self.init_module_config()
 333
 334         while self.run:
 335             self.log.debug('Waking up for new iteration')
 336
 337             try:
 338                 self.send()
 339             except Exception as exc:
 340                 # Shouldn't happen, but let's log it and retry next interval,
 341                 # rather than dying completely.
 342                 self.log.exception("Unexpected error during send():")
 343
 344             interval = self.config['interval']
 345             self.log.debug('Sleeping for %d seconds', interval)
 346             self.event.wait(interval)
 347
 348     def self_test(self):
 349         data = self.get_data()
 350
 351         if data['overall_status'] not in self.ceph_health_mapping:
 352             raise RuntimeError('No valid overall_status found in data')
 353
 354         int(data['overall_status_int'])
 355
 356         if data['num_mon'] < 1:
 357             raise RuntimeError('num_mon is smaller than 1')