src/collectors/python.d.plugin/ceph/ceph.chart.py
# -*- coding: utf-8 -*-
# Description: ceph netdata python.d module
# Author: Luis Eduardo (lets00)
# SPDX-License-Identifier: GPL-3.0-or-later
try:
import rados
CEPH = True
except ImportError:
CEPH = False
import json
import os
from bases.FrameworkServices.SimpleService import SimpleService
# default module values (can be overridden per job in `config`)
update_every = 10
ORDER = [
'general_usage',
'general_objects',
'general_bytes',
'general_operations',
'general_latency',
'pool_usage',
'pool_objects',
'pool_read_bytes',
'pool_write_bytes',
'pool_read_operations',
'pool_write_operations',
'osd_usage',
'osd_size',
'osd_apply_latency',
'osd_commit_latency'
]
CHARTS = {
'general_usage': {
'options': [None, 'Ceph General Space', 'KiB', 'general', 'ceph.general_usage', 'stacked'],
'lines': [
['general_available', 'avail', 'absolute'],
['general_usage', 'used', 'absolute']
]
},
'general_objects': {
'options': [None, 'Ceph General Objects', 'objects', 'general', 'ceph.general_objects', 'area'],
'lines': [
['general_objects', 'cluster', 'absolute']
]
},
'general_bytes': {
'options': [None, 'Ceph General Read/Write Data/s', 'KiB/s', 'general', 'ceph.general_bytes',
'area'],
'lines': [
['general_read_bytes', 'read', 'absolute', 1, 1024],
['general_write_bytes', 'write', 'absolute', -1, 1024]
]
},
'general_operations': {
'options': [None, 'Ceph General Read/Write Operations/s', 'operations', 'general', 'ceph.general_operations',
'area'],
'lines': [
['general_read_operations', 'read', 'absolute', 1],
['general_write_operations', 'write', 'absolute', -1]
]
},
'general_latency': {
'options': [None, 'Ceph General Apply/Commit latency', 'milliseconds', 'general', 'ceph.general_latency',
'area'],
'lines': [
['general_apply_latency', 'apply', 'absolute'],
['general_commit_latency', 'commit', 'absolute']
]
},
'pool_usage': {
'options': [None, 'Ceph Pools', 'KiB', 'pool', 'ceph.pool_usage', 'line'],
'lines': []
},
'pool_objects': {
'options': [None, 'Ceph Pools', 'objects', 'pool', 'ceph.pool_objects', 'line'],
'lines': []
},
'pool_read_bytes': {
'options': [None, 'Ceph Read Pool Data/s', 'KiB/s', 'pool', 'ceph.pool_read_bytes', 'area'],
'lines': []
},
'pool_write_bytes': {
'options': [None, 'Ceph Write Pool Data/s', 'KiB/s', 'pool', 'ceph.pool_write_bytes', 'area'],
'lines': []
},
'pool_read_operations': {
'options': [None, 'Ceph Read Pool Operations/s', 'operations', 'pool', 'ceph.pool_read_operations', 'area'],
'lines': []
},
'pool_write_operations': {
'options': [None, 'Ceph Write Pool Operations/s', 'operations', 'pool', 'ceph.pool_write_operations', 'area'],
'lines': []
},
'osd_usage': {
'options': [None, 'Ceph OSDs', 'KiB', 'osd', 'ceph.osd_usage', 'line'],
'lines': []
},
'osd_size': {
'options': [None, 'Ceph OSDs size', 'KiB', 'osd', 'ceph.osd_size', 'line'],
'lines': []
},
'osd_apply_latency': {
'options': [None, 'Ceph OSDs apply latency', 'milliseconds', 'osd', 'ceph.apply_latency', 'line'],
'lines': []
},
'osd_commit_latency': {
'options': [None, 'Ceph OSDs commit latency', 'milliseconds', 'osd', 'ceph.commit_latency', 'line'],
'lines': []
}
}
class Service(SimpleService):
def __init__(self, configuration=None, name=None):
SimpleService.__init__(self, configuration=configuration, name=name)
self.order = ORDER
self.definitions = CHARTS
self.config_file = self.configuration.get('config_file')
self.keyring_file = self.configuration.get('keyring_file')
self.rados_id = self.configuration.get('rados_id', 'admin')
def check(self):
"""
Checks module
:return:
"""
if not CEPH:
self.error('rados module is needed to use ceph.chart.py')
return False
if not (self.config_file and self.keyring_file):
self.error('config_file and/or keyring_file is not defined')
return False
# Verify files and permissions
if not (os.access(self.config_file, os.F_OK)):
self.error('{0} does not exist'.format(self.config_file))
return False
if not (os.access(self.keyring_file, os.F_OK)):
self.error('{0} does not exist'.format(self.keyring_file))
return False
if not (os.access(self.config_file, os.R_OK)):
self.error('Ceph plugin does not read {0}, define read permission.'.format(self.config_file))
return False
if not (os.access(self.keyring_file, os.R_OK)):
self.error('Ceph plugin does not read {0}, define read permission.'.format(self.keyring_file))
return False
try:
self.cluster = rados.Rados(conffile=self.config_file,
conf=dict(keyring=self.keyring_file),
rados_id=self.rados_id)
self.cluster.connect()
except rados.Error as error:
self.error(error)
return False
self.create_definitions()
return True
def create_definitions(self):
"""
Create dynamically charts options
:return: None
"""
# Pool lines
for pool in sorted(self._get_df()['pools'], key=lambda x: sorted(x.keys())):
self.definitions['pool_usage']['lines'].append([pool['name'],
pool['name'],
'absolute'])
self.definitions['pool_objects']['lines'].append(["obj_{0}".format(pool['name']),
pool['name'],
'absolute'])
self.definitions['pool_read_bytes']['lines'].append(['read_{0}'.format(pool['name']),
pool['name'],
'absolute', 1, 1024])
self.definitions['pool_write_bytes']['lines'].append(['write_{0}'.format(pool['name']),
pool['name'],
'absolute', 1, 1024])
self.definitions['pool_read_operations']['lines'].append(['read_operations_{0}'.format(pool['name']),
pool['name'],
'absolute'])
self.definitions['pool_write_operations']['lines'].append(['write_operations_{0}'.format(pool['name']),
pool['name'],
'absolute'])
# OSD lines
for osd in sorted(self._get_osd_df()['nodes'], key=lambda x: sorted(x.keys())):
self.definitions['osd_usage']['lines'].append([osd['name'],
osd['name'],
'absolute'])
self.definitions['osd_size']['lines'].append(['size_{0}'.format(osd['name']),
osd['name'],
'absolute'])
self.definitions['osd_apply_latency']['lines'].append(['apply_latency_{0}'.format(osd['name']),
osd['name'],
'absolute'])
self.definitions['osd_commit_latency']['lines'].append(['commit_latency_{0}'.format(osd['name']),
osd['name'],
'absolute'])
def get_data(self):
"""
Catch all ceph data
:return: dict
"""
try:
data = {}
df = self._get_df()
osd_df = self._get_osd_df()
osd_perf = self._get_osd_perf()
osd_perf_infos = get_osd_perf_infos(osd_perf)
pool_stats = self._get_osd_pool_stats()
data.update(self._get_general(osd_perf_infos, pool_stats))
for pool in df['pools']:
data.update(self._get_pool_usage(pool))
data.update(self._get_pool_objects(pool))
for pool_io in pool_stats:
data.update(self._get_pool_rw(pool_io))
for osd in osd_df['nodes']:
data.update(self._get_osd_usage(osd))
data.update(self._get_osd_size(osd))
for osd_apply_commit in osd_perf_infos:
data.update(self._get_osd_latency(osd_apply_commit))
return data
except (ValueError, AttributeError) as error:
self.error(error)
return None
def _get_general(self, osd_perf_infos, pool_stats):
"""
Get ceph's general usage
:return: dict
"""
status = self.cluster.get_cluster_stats()
read_bytes_sec = 0
write_bytes_sec = 0
read_op_per_sec = 0
write_op_per_sec = 0
apply_latency = 0
commit_latency = 0
for pool_rw_io_b in pool_stats:
read_bytes_sec += pool_rw_io_b['client_io_rate'].get('read_bytes_sec', 0)
write_bytes_sec += pool_rw_io_b['client_io_rate'].get('write_bytes_sec', 0)
read_op_per_sec += pool_rw_io_b['client_io_rate'].get('read_op_per_sec', 0)
write_op_per_sec += pool_rw_io_b['client_io_rate'].get('write_op_per_sec', 0)
for perf in osd_perf_infos:
apply_latency += perf['perf_stats']['apply_latency_ms']
commit_latency += perf['perf_stats']['commit_latency_ms']
return {
'general_usage': int(status['kb_used']),
'general_available': int(status['kb_avail']),
'general_objects': int(status['num_objects']),
'general_read_bytes': read_bytes_sec,
'general_write_bytes': write_bytes_sec,
'general_read_operations': read_op_per_sec,
'general_write_operations': write_op_per_sec,
'general_apply_latency': apply_latency,
'general_commit_latency': commit_latency
}
@staticmethod
def _get_pool_usage(pool):
"""
Process raw data into pool usage dict information
:return: A pool dict with pool name's key and usage bytes' value
"""
return {pool['name']: pool['stats']['kb_used']}
@staticmethod
def _get_pool_objects(pool):
"""
Process raw data into pool usage dict information
:return: A pool dict with pool name's key and object numbers
"""
return {'obj_{0}'.format(pool['name']): pool['stats']['objects']}
@staticmethod
def _get_pool_rw(pool):
"""
Get read/write kb and operations in a pool
:return: A pool dict with both read/write bytes and operations.
"""
return {
'read_{0}'.format(pool['pool_name']): int(pool['client_io_rate'].get('read_bytes_sec', 0)),
'write_{0}'.format(pool['pool_name']): int(pool['client_io_rate'].get('write_bytes_sec', 0)),
'read_operations_{0}'.format(pool['pool_name']): int(pool['client_io_rate'].get('read_op_per_sec', 0)),
'write_operations_{0}'.format(pool['pool_name']): int(pool['client_io_rate'].get('write_op_per_sec', 0))
}
@staticmethod
def _get_osd_usage(osd):
"""
Process raw data into osd dict information to get osd usage
:return: A osd dict with osd name's key and usage bytes' value
"""
return {osd['name']: float(osd['kb_used'])}
@staticmethod
def _get_osd_size(osd):
"""
Process raw data into osd dict information to get osd size (kb)
:return: A osd dict with osd name's key and size bytes' value
"""
return {'size_{0}'.format(osd['name']): float(osd['kb'])}
@staticmethod
def _get_osd_latency(osd):
"""
Get ceph osd apply and commit latency
:return: A osd dict with osd name's key with both apply and commit latency values
"""
return {
'apply_latency_osd.{0}'.format(osd['id']): osd['perf_stats']['apply_latency_ms'],
'commit_latency_osd.{0}'.format(osd['id']): osd['perf_stats']['commit_latency_ms']
}
def _get_df(self):
"""
Get ceph df output
:return: ceph df --format json
"""
return json.loads(self.cluster.mon_command(json.dumps({
'prefix': 'df',
'format': 'json'
}), b'')[1].decode('utf-8'))
def _get_osd_df(self):
"""
Get ceph osd df output
:return: ceph osd df --format json
"""
return json.loads(self.cluster.mon_command(json.dumps({
'prefix': 'osd df',
'format': 'json'
}), b'')[1].decode('utf-8').replace('-nan', '"-nan"'))
def _get_osd_perf(self):
"""
Get ceph osd performance
:return: ceph osd perf --format json
"""
return json.loads(self.cluster.mon_command(json.dumps({
'prefix': 'osd perf',
'format': 'json'
}), b'')[1].decode('utf-8'))
def _get_osd_pool_stats(self):
"""
Get ceph osd pool status.
This command is used to get information about both
read/write operation and bytes per second on each pool
:return: ceph osd pool stats --format json
"""
return json.loads(self.cluster.mon_command(json.dumps({
'prefix': 'osd pool stats',
'format': 'json'
}), b'')[1].decode('utf-8'))
def get_osd_perf_infos(osd_perf):
# https://github.com/netdata/netdata/issues/8247
# module uses 'osd_perf_infos' data, its been moved under 'osdstats` since Ceph v14.2
if 'osd_perf_infos' in osd_perf:
return osd_perf['osd_perf_infos']
return osd_perf['osdstats']['osd_perf_infos']