1#@!#!123s

: / opt / cloudlinux / venv / lib / python3.11 / site-packages / lvestats / plugins / generic /
Filename : aggregators.py
back
# 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] = []