22

✏️ Editing: update_utils.py

# Copyright (c) Cloud Linux Software, Inc
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT

import functools
import json
import os
import time

from . import config, constants, log_utils, utils
from .py23 import json_loads_nstr

if False:  # pragma: no cover
    from typing import Any, Callable, Dict  # noqa: F401

STATUS_CHANGE_GAP_DELAY = 5 * 60  # 5 minute

def touch_status_gap_file(filename='.kcarestatus'):
    status_filepath = os.path.join(constants.PATCH_CACHE, filename)
    utils.atomic_write(status_filepath, utils.timestamp_str())

def status_gap_passed(filename='.kcarestatus'):
    status_filepath = os.path.join(constants.PATCH_CACHE, filename)
    if os.path.isfile(status_filepath):
        with open(status_filepath, 'r') as sfile:
            try:
                timestamp = int(sfile.read())
                if int(timestamp) + config.STATUS_CHANGE_GAP + STATUS_CHANGE_GAP_DELAY > time.time():
                    return False
            except Exception:
                pass
    return True

def _check_component(component):
    # type: (str) -> None
    if component not in ('kernel', 'libcare'):
        raise ValueError('Unknown update status component: {0}'.format(component))

def _load_update_status():
    # type: () -> Dict[str, Any]
    content = utils.read_file(constants.UPDATE_STATUS_PATH)
    if content is None:
        return {}

try:
        result = json_loads_nstr(content)  # type: Dict[str, Any]
        return result
    except (ValueError, TypeError):
        log_utils.kcarelog.warning('Failed to parse update status file')
        return {}

def save_update_status(component, error):
    # type: (str, str) -> None
    _check_component(component)
    try:
        data = _load_update_status()
        data[component] = {
            'error': error,
            'timestamp': int(time.time()),
        }
        utils.atomic_write(constants.UPDATE_STATUS_PATH, json.dumps(data))
    except Exception:
        log_utils.kcarelog.warning('Failed to save update status', exc_info=True)

def _error_status(err):
    # type: (Exception) -> str
    status = getattr(err, 'status', '')
    if status:
        return status

return str(err)

def track_update_status(component):
    # type: (str) -> Callable[..., Any]
    _check_component(component)

def decorator(fn):  # type: (Callable[..., Any]) -> Callable[..., Any]
        @functools.wraps(fn)
        def inner(*args, **kwargs):  # type: (Any, Any) -> Any
            try:
                result = fn(*args, **kwargs)
            except Exception as err:
                save_update_status(component, error=_error_status(err))
                raise

save_update_status(component, error='')
            return result

return inner

return decorator

def read_update_error(component):
    # type: (str) -> str
    _check_component(component)
    try:
        data = _load_update_status()
        error = data.get(component, {}).get('error', '')  # type: str
        return error[: constants.UPDATE_ERROR_MAX_LENGTH]
    except Exception:
        log_utils.kcarelog.warning('Failed to read update status', exc_info=True)
        return ''

MR SKB