import socket
+from urllib.parse import ParseResult
+from typing import Any, Dict, Optional, Tuple, Union
class BaseSocket(object):
'udp6': (socket.AF_INET6, socket.SOCK_DGRAM),
}
- def __init__(self, url):
+ def __init__(self, url: ParseResult) -> None:
self.url = url
try:
self.sock = socket.socket(family=socket_family, type=socket_type)
if self.sock.family == socket.AF_UNIX:
- self.address = self.url.path
+ self.address: Union[str, Tuple[str, int]] = self.url.path
else:
+ assert self.url.hostname
+ assert self.url.port
self.address = (self.url.hostname, self.url.port)
- def connect(self):
+ def connect(self) -> None:
return self.sock.connect(self.address)
- def close(self):
+ def close(self) -> None:
self.sock.close()
- def send(self, data, flags=0):
+ def send(self, data: str, flags: int = 0) -> int:
return self.sock.send(data.encode('utf-8') + b'\n', flags)
- def __del__(self):
+ def __del__(self) -> None:
self.sock.close()
- def __enter__(self):
+ def __enter__(self) -> 'BaseSocket':
self.connect()
return self
- def __exit__(self, exc_type, exc_val, exc_tb):
+ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
self.close()
from telegraf.basesocket import BaseSocket
from telegraf.protocol import Line
-from mgr_module import CLICommand, CLIReadCommand, MgrModule, Option, PG_STATES
+from mgr_module import CLICommand, CLIReadCommand, MgrModule, Option, OptionValue, PG_STATES
-from typing import Tuple
+from typing import cast, Any, Dict, Iterable, Optional, Tuple
from urllib.parse import urlparse
ceph_health_mapping = {'HEALTH_OK': 0, 'HEALTH_WARN': 1, 'HEALTH_ERR': 2}
@property
- def config_keys(self):
+ def config_keys(self) -> Dict[str, OptionValue]:
return dict((o['name'], o.get('default', None)) for o in self.MODULE_OPTIONS)
- def __init__(self, *args, **kwargs):
+ def __init__(self, *args: Any, **kwargs: Any) -> None:
super(Module, self).__init__(*args, **kwargs)
self.event = Event()
self.run = True
- self.fsid = None
- self.config = dict()
+ self.fsid: Optional[str] = None
+ self.config: Dict[str, OptionValue] = dict()
- def get_fsid(self):
+ def get_fsid(self) -> str:
if not self.fsid:
self.fsid = self.get('mon_map')['fsid']
-
+ assert self.fsid is not None
return self.fsid
- def get_pool_stats(self):
+ def get_pool_stats(self) -> Iterable[Dict[str, Any]]:
df = self.get('df')
df_types = [
'value': pool['stats'][df_type],
}
- def get_daemon_stats(self):
+ def get_daemon_stats(self) -> Iterable[Dict[str, Any]]:
for daemon, counters in self.get_all_perf_counters().items():
svc_type, svc_id = daemon.split('.', 1)
metadata = self.get_metadata(svc_type, svc_id)
'value': counter_info['value']
}
- def get_pg_stats(self):
+ def get_pg_stats(self) -> Dict[str, int]:
stats = dict()
pg_status = self.get('pg_status')
return stats
- def get_cluster_stats(self):
+ def get_cluster_stats(self) -> Iterable[Dict[str, Any]]:
stats = dict()
health = json.loads(self.get('health')['json'])
num_mds_up += len(fs['mdsmap']['up'])
stats['num_mds_up'] = num_mds_up
- stats['num_mds'] = num_mds_up + stats['num_mds_standby']
+ stats['num_mds'] = num_mds_up + cast(int, stats['num_mds_standby'])
stats.update(self.get_pg_stats())
for key, value in stats.items():
+ assert value is not None
yield {
'measurement': 'ceph_cluster_stats',
'tags': {
'value': int(value)
}
- def set_config_option(self, option, value):
+ def set_config_option(self, option: str, value: str) -> None:
if option not in self.config_keys.keys():
raise RuntimeError('{0} is a unknown configuration '
'option'.format(option))
- if option in ['interval']:
+ if option == 'interval':
try:
- value = int(value)
+ interval = int(value)
except (ValueError, TypeError):
raise RuntimeError('invalid {0} configured. Please specify '
'a valid integer'.format(option))
+ if interval < 5:
+ raise RuntimeError('interval should be set to at least 5 seconds')
+ self.config[option] = interval
+ else:
+ self.config[option] = value
- if option == 'interval' and value < 5:
- raise RuntimeError('interval should be set to at least 5 seconds')
-
- self.config[option] = value
-
- def init_module_config(self):
+ def init_module_config(self) -> None:
self.config['address'] = \
self.get_module_option("address", default=self.config_keys['address'])
- self.config['interval'] = \
- int(self.get_module_option("interval",
- default=self.config_keys['interval']))
+ interval = self.get_module_option("interval",
+ default=self.config_keys['interval'])
+ assert interval
+ self.config['interval'] = int(interval)
- def now(self):
+ def now(self) -> int:
return int(round(time.time() * 1000000000))
- def gather_measurements(self):
+ def gather_measurements(self) -> Iterable[Dict[str, Any]]:
return itertools.chain(
self.get_pool_stats(),
self.get_daemon_stats(),
self.get_cluster_stats()
)
- def send_to_telegraf(self):
- url = urlparse(self.config['address'])
+ def send_to_telegraf(self) -> None:
+ url = urlparse(cast(str, self.config['address']))
sock = BaseSocket(url)
self.log.debug('Sending data to Telegraf at %s', sock.address)
except FileNotFoundError:
self.log.exception('Failed to open Telegraf at: %s', url.geturl())
- def shutdown(self):
+ def shutdown(self) -> None:
self.log.info('Stopping Telegraf module')
self.run = False
self.event.set()
self.send_to_telegraf()
return 0, 'Sending data to Telegraf', ''
- def self_test(self):
+ def self_test(self) -> None:
measurements = list(self.gather_measurements())
if len(measurements) == 0:
raise RuntimeError('No measurements found')
- def serve(self):
+ def serve(self) -> None:
self.log.info('Starting Telegraf module')
self.init_module_config()
self.run = True
runtime = (self.now() - start) / 1000000
self.log.debug('Sending data to Telegraf took %d ms', runtime)
self.log.debug("Sleeping for %d seconds", self.config['interval'])
- self.event.wait(self.config['interval'])
+ self.event.wait(cast(int, self.config['interval']))
-from telegraf.utils import format_string, format_value
+from typing import Dict, Optional, Union
+
+from telegraf.utils import format_string, format_value, ValueType
class Line(object):
- def __init__(self, measurement, values, tags=None, timestamp=None):
+ def __init__(self,
+ measurement: ValueType,
+ values: Union[Dict[str, ValueType], ValueType],
+ tags: Optional[Dict[str, str]] = None,
+ timestamp: Optional[int] = None) -> None:
self.measurement = measurement
self.values = values
self.tags = tags
self.timestamp = timestamp
- def get_output_measurement(self):
+ def get_output_measurement(self) -> str:
return format_string(self.measurement)
- def get_output_values(self):
+ def get_output_values(self) -> str:
if not isinstance(self.values, dict):
metric_values = {'value': self.values}
else:
return ','.join('{0}={1}'.format(format_string(k), format_value(v)) for k, v in sorted_values)
- def get_output_tags(self):
+ def get_output_tags(self) -> str:
if not self.tags:
self.tags = dict()
return ','.join('{0}={1}'.format(format_string(k), format_string(v)) for k, v in sorted_tags)
- def get_output_timestamp(self):
+ def get_output_timestamp(self) -> str:
return ' {0}'.format(self.timestamp) if self.timestamp else ''
- def to_line_protocol(self):
+ def to_line_protocol(self) -> str:
tags = self.get_output_tags()
return '{0}{1} {2}{3}'.format(
projects / ceph.git / commitdiff