Viewing file:  aggregators.py (5.45 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
from collections import defaultdict
from lvestats.core.plugin import LveStatsPlugin
from lvestats.plugins.generic.analyzers import LVEUsage, FAULTS
class AggregatedLveUsage(LVEUsage):
__slots__ = ()
def __init__(self, lve_version=8):
LVEUsage.__init__(self, lve_version)
self.time = 0.0
def add(self, lveusage):
self.time += lveusage.time or 0
if lveusage.time:
self.cpu_usage += lveusage.cpu_usage * lveusage.time
self.mem_usage += lveusage.mem_usage * lveusage.time
self.mep += lveusage.mep * lveusage.time
self.io_usage += lveusage.io_usage * lveusage.time
self.iops += lveusage.iops * lveusage.time
self.memphy += lveusage.memphy * lveusage.time
self.nproc += lveusage.nproc * lveusage.time
self.cpu_fault += lveusage.cpu_fault
self.mem_fault += lveusage.mem_fault
self.mep_fault += lveusage.mep_fault
self.io_fault += lveusage.io_fault
# Commented due to LVES-159. See aggregate function below
# self.iops_fault += lveusage.iops_fault
self.memphy_fault += lveusage.memphy_fault
self.nproc_fault += lveusage.nproc_fault
self.lmem = max(lveusage.lmem, self.lmem)
self.lcpu = max(lveusage.lcpu, self.lcpu)
self.lep = max(lveusage.lep, self.lep)
self.io = max(lveusage.io, self.io)
self.lmemphy = max(lveusage.lmemphy, self.lmemphy)
self.lnproc = max(lveusage.lnproc, self.lnproc)
self.liops = max(lveusage.liops, self.liops)
self.has_changed_limits = self.has_changed_limits or lveusage.has_changed_limits
self.has_changed_nproc = self.has_changed_nproc or lveusage.has_changed_nproc
def aggregate(self):
if self.time > 0:
self.cpu_usage /= self.time
self.mem_usage /= self.time
self.mep /= self.time
self.io_usage /= self.time
self.iops /= self.time
self.memphy /= self.time
self.nproc /= self.time
# LVES-159: change the way iops_faults are calculated
# if aIOPS >= lIOPS, then iops_faults = 1
if self.iops >= self.liops > 0:
self.iops_fault = 1
else:
self.iops_fault = 0
class LveUsageAggregator5S(LveStatsPlugin):
def __init__(self):
self.get_data_from = 'lve_usages_5s'
self.set_data_to = 'lve_usage_5s'
self.snapshots_enabled = True
def set_config(self, config):
"""
:param dict config:
"""
self.snapshots_enabled = config.get('disable_snapshots', "false").lower() != "true"
def dict_sum_values(self, dict1, dict2):
"""
:param dict dict1:
:param dict dict2:
"""
result = {}
items = list(dict1.items()) + list(dict2.items())
for attr_name, count in items:
result[attr_name] = result.get(attr_name, 0) + count
return result
def aggregate_faults(self, lve_data, lve_usage):
"""
:type lve_data: dict
:type lve_usage: dict[int, AggregatedLveUsage]
"""
if self.snapshots_enabled:
old_faults = lve_data.get('faults', {})
for lve_id, usage in lve_usage.items():
for attr_name in FAULTS:
fault = getattr(usage, attr_name)
if fault > 0:
if lve_id not in old_faults:
old_faults[lve_id] = {}
old_faults[lve_id][attr_name] = old_faults[lve_id].get(attr_name, 0) + fault
lve_data['faults'] = old_faults
def execute(self, lve_data):
"""
:param dict lve_data:
"""
lve_usages = lve_data[self.get_data_from]
aggregated = defaultdict(AggregatedLveUsage)
for iteration_data in lve_usages:
for lve_id, lve_usage in iteration_data.items():
aggregated[lve_id].add(lve_usage)
for aggregated_lve_usage in aggregated.values():
aggregated_lve_usage.aggregate()
result = {}
for lve_id, usage in aggregated.items():
if lve_id == 0:
result[lve_id] = usage
continue
if usage.has_interesting_values() \
or usage.has_changed_limits \
or usage.has_changed_nproc:
result[lve_id] = usage
lve_data[self.set_data_to] = result
self.aggregate_faults(lve_data, result)
# FIXME(vlebedev): Dirty hack to get raw usage measurement in plugins after this one.
# lve_data[self.get_data_from] = []
class LveUsageAggregator(LveUsageAggregator5S):
def __init__(self): # pylint: disable=super-init-not-called
self.period = 60
self.get_data_from = 'lve_usages'
self.set_data_to = 'lve_usage'
def set_config(self, config):
"""
:param dict config:
"""
self.period = config.get('aggregation_period', self.period)
def aggregate_faults(self, lve_data, lve_usage):
pass
# FIXME(vlebedev): Dirty hack to get raw usage measurement in plugins after this one.
def execute(self, lve_data):
super().execute(lve_data)
lve_data[self.get_data_from] = []
|