Edit

[EDIT] FILE: dbsaver_x60.py

# 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

import logging
import sys
import time

from sqlalchemy import insert
from sqlalchemy.exc import OperationalError
from sqlalchemy.sql import select
from sqlalchemy.sql.expression import func

from lvestats.core.plugin import LveStatsPlugin, LveStatsPluginTerminated
from lvestats.lib import dbengine
from lvestats.lib.commons.func import get_chunks, reboot_lock
from lvestats.lib.commons.progress import ProgressCallback
from lvestats.lib.lveinfolib import FIELD_TO_TABLE_COLUMN, HistoryShow
from lvestats.orm.history import history, history_x60

AGGREGATE_PERIOD = 60 * 60  # one hour

class DBSaverX60(LveStatsPlugin):
    FIELDS = [
        'ID',
        'aCPU',
        'lCPU',
        'CPUf',
        'aEP',
        'lEP',
        'EPf',
        'aVMem',
        'lVMem',
        'VMemF',
        'aPMem',
        'lPMem',
        'PMemF',
        'aNproc',
        'lNproc',
        'NprocF',
        'aIO',
        'lIO',
        'IOf',
        'IOPSf',
        'lIOPS',
        'aIOPS',
    ]

def __init__(self):
        self.log = logging.getLogger('plugin.DBSaverX60')
        self.now = 0  # This changes in MainLoop
        self.config = None
        self.period = AGGREGATE_PERIOD  # once an hour
        self.server_id = 'localhost'
        self.verbose_ = False
        self.execute_timeout = 20
        self.progress = ProgressCallback(time_change_step=0.1)

def set_config(self, _config):
        self.config = _config

def aggregate_db_data_by_hours(self, utc_from, utc_to):
        if self.verbose_:
            self.progress(0)
        utc_from_ = int(utc_from) // int(AGGREGATE_PERIOD) * int(AGGREGATE_PERIOD)
        if utc_from_ != utc_from:
            utc_from_ += AGGREGATE_PERIOD

if utc_to - utc_from_ > AGGREGATE_PERIOD:
            self.log.debug("Data aggregation from %s to %s started.", utc_from_, utc_to)
            for aggr_cycle, hour_from in enumerate(range(utc_from_, int(utc_to), AGGREGATE_PERIOD), 1):
                hour_to = hour_from + AGGREGATE_PERIOD
                if hour_to > utc_to:
                    break
                history_show = HistoryShow(
                    dbengine=self.engine,
                    period_from=hour_from + 1,  # "+1" for exclude "hour_from" timestamp
                    period_to=hour_to,
                    show_columns=self.FIELDS,
                    server_id=self.server_id,
                    log=self.log,
                )
                rows = history_show.proceed()
                mass_insert_values = []
                for row in rows:
                    one_insert_values = {
                        'server_id': self.server_id,
                        'time': history_show.time_count,
                        'created': hour_to,
                    }
                    one_insert_values.update(
                        {FIELD_TO_TABLE_COLUMN[field_]: row[indx_] for indx_, field_ in enumerate(self.FIELDS)}
                    )
                    # correct cpu
                    one_insert_values['cpu'] *= 100
                    one_insert_values['cpu_limit'] *= 100
                    mass_insert_values.append(one_insert_values)

with reboot_lock():
                    conn_ = self.engine.connect()
                    trans = conn_.begin()
                    try:
                        if mass_insert_values:
                            for chunk in get_chunks(mass_insert_values):
                                conn_.execute(insert(history_x60), chunk)
                        else:
                            conn_.execute(
                                insert(history_x60), {'server_id': self.server_id, 'created': hour_to, 'id': 0}
                            )
                        if self.verbose_:  # show progress
                            progress_ = 100 - (utc_to - hour_from) * 100 // (utc_to - utc_from_)
                            self.progress(progress_)
                    except OperationalError as oe:
                        trans.rollback()
                        self.log.error('Can not write aggregated data to database: %s', str(oe))
                        if self.verbose_:
                            print(f' Warning: {str(oe)[:255]}...')
                    except LveStatsPluginTerminated as exc:
                        trans.rollback()
                        conn_.close()
                        self.log.debug("Plugin is terminated.")
                        raise LveStatsPluginTerminated() from exc
                    except TypeError as oe:
                        trans.rollback()
                        conn_.close()
                        msg = str(oe)
                        # try fix incorrect database records after migration
                        if "*=: 'NoneType' and 'int'" not in msg:
                            raise
                        self.log.warning(msg)
                        from lvestats.lib.dbengine import fix_db  # pylint: disable=import-outside-toplevel

fix_db(self.engine, self.config, from_timestmp=hour_from, to_timestamp=hour_to, log_=self.log)
                        return
                    else:
                        trans.commit()
                        conn_.close()

work_time = time.time() - self.now  # calculate plugin working time
                if (
                    self.execute_timeout > 0 and self.execute_timeout - work_time < work_time / aggr_cycle + 1
                ):  # continue if can't finish in time
                    self.log.debug('Data was aggregated from %s to %s in %s seconds', utc_from_, hour_to, work_time)
                    return
            self.log.debug("Data aggregation done.")
        else:
            self.log.debug("Nothing to aggregate.")
        if self.verbose_:
            self.progress(100, force=True)
            self.progress.stop()

def execute(self, lve_data):
        self.log.debug("Plugin executed")
        self.server_id = self.config.get('server_id', 'localhost')
        dt_x1_max, dt_x1_min, dt_x60_max, dt_x60_min = self.get_max_min_ts()

self.aggregate_new(dt_x1_max, dt_x1_min, dt_x60_max)
        self.aggregate_migrated(dt_x1_min, dt_x60_min)

def aggregate_migrated(self, dt_x1_min, dt_x60_min):
        self.log.debug("Aggregating migrated")
        if dt_x1_min is not None and dt_x60_min is not None and dt_x1_min + AGGREGATE_PERIOD < dt_x60_min:
            self.aggregate_db_data_by_hours(dt_x60_min - AGGREGATE_PERIOD, dt_x60_min)

def aggregate_new(self, dt_x1_max, dt_x1_min, dt_x60_max):
        if dt_x1_max:
            if not dt_x60_max:
                dt_x60_max = dt_x1_min  # from minimum timestamp
            if self.verbose_:
                print(
                    f'[lve-stats]: Start aggregating data from "{dt_x60_max}" '
                    f'to "{dt_x1_max}"; to skip press Ctrl+C ...'
                )
            # correct execute period
            if dt_x1_max - dt_x60_max > AGGREGATE_PERIOD * 2:  # run plugin often if need aggregate more than one hour
                self.period = 60  # FIXME: This looks like mistake?
            else:
                self.period = AGGREGATE_PERIOD
            self.aggregate_db_data_by_hours(dt_x60_max, dt_x1_max)

def get_max_min_ts(self):
        conn_ = self.engine.connect()
        try:
            # get last created
            query = select([func.max(history.created), func.min(history.created)]).where(
                self.server_id == history.server_id
            )
            time_start = time.time()
            row = conn_.execute(query)
            dt_x1_max, dt_x1_min = row.fetchone()
            self.log.debug(
                '%s; query time: %s',
                str(query.compile(compile_kwargs={"literal_binds": True})).replace('\n', ' '),
                time.time() - time_start,
            )

# get last created_max
            query = select([func.max(history_x60.created), func.min(history_x60.created)]).where(
                self.server_id == history_x60.server_id
            )
            row = conn_.execute(query)
            dt_x60_max, dt_x60_min = row.fetchone()
            self.log.debug(
                '%s; query time: %s',
                str(query.compile(compile_kwargs={"literal_binds": True})).replace('\n', ' '),
                time.time() - time_start,
            )
        except LveStatsPluginTerminated as exc:
            conn_.close()
            self.log.debug("Plugin is terminated.")
            raise LveStatsPluginTerminated() from exc
        else:
            conn_.close()

return dt_x1_max, dt_x1_min, dt_x60_max, dt_x60_min

def main(argv_, config_, db_engine=None):
    lve_data = {}
    verbose_ = '--verbose' in argv_
    if db_engine is None:
        db_engine = dbengine.make_db_engine(config_, debug=False)
    if 'init' in argv_:
        if verbose_:
            print(f'[lve-stats]: creating table "{history_x60.__tablename__}"...')
        with db_engine.begin() as conn_:
            history_x60.__table__.create(bind=conn_, checkfirst=True)  # create table if exist
    if 'aggregate' in argv_:
        # configure plugin
        plugin_instance = DBSaverX60()
        plugin_instance.config = config_  # use specific config
        plugin_instance.set_db_engine(db_engine)
        plugin_instance.execute_timeout = 2 * 60 * 60  # two hours
        plugin_instance.verbose_ = verbose_
        plugin_instance.now = time.time()

plugin_instance.execute(lve_data)  # generate aggregate data

# for initial from command line
# create table if exist and generate aggregate data
if __name__ == '__main__' and len(sys.argv) >= 2:
    from lvestats.lib import config

try:
        cnf = config.read_plugin_config(config.read_config(), plugin=DBSaverX60.__name__)
        main(argv_=sys.argv[1:], config_=cnf)
    except KeyboardInterrupt:
        print('Stopped')
    except dbengine.MakeDbException as e:
        print(str(e))
    except config.ConfigError as ce:
        ce.log_and_exit()

CANCEL