fission-console/CRON/PYTHON_CRON_FUNCTION.md

Python cron function template

Эту функцию можно создать вручную в console UI и привязать к TimeTrigger.

Она собирает базовые метрики системы, отдает JSON и сразу отправляет snapshot в страницу cron dashboard.

Что нужно задать в UI

• Route: например /system-metrics
• Methods: GET,POST
• Cron: например */5 * * * *
• Entrypoint: main.main

Код

import json
import os
import socket
import urllib.request
from datetime import datetime, timezone


def read_meminfo():
    mem = {}
    try:
        with open('/proc/meminfo', 'r', encoding='utf-8') as fh:
            for line in fh:
                if ':' not in line:
                    continue
                key, value = line.split(':', 1)
                parts = value.strip().split()
                if not parts:
                    continue
                try:
                    mem[key] = float(parts[0]) / 1024.0  # kB -> MB
                except ValueError:
                    continue
    except OSError:
        pass
    return mem


def read_uptime_sec():
    try:
        with open('/proc/uptime', 'r', encoding='utf-8') as fh:
            return float(fh.read().split()[0])
    except Exception:
        return 0.0


def read_disk_gb(path='/'):
    st = os.statvfs(path)
    total = (st.f_blocks * st.f_frsize) / (1024 ** 3)
    free = (st.f_bavail * st.f_frsize) / (1024 ** 3)
    used = total - free
    return total, free, used


def payload():
    mem = read_meminfo()
    total = mem.get('MemTotal', 0.0)
    free = mem.get('MemFree', 0.0)
    available = mem.get('MemAvailable', free)
    used = max(0.0, total - free) if total else 0.0
    mem_percent = (used / total) * 100.0 if total else 0.0

    try:
        load1, load5, load15 = os.getloadavg()
    except (AttributeError, OSError):
        load1 = load5 = load15 = 0.0

    disk_total, disk_free, disk_used = read_disk_gb('/')
    return {
        'timestamp': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'),
        'source': os.getenv('CRON_SOURCE', 'python-cron'),
        'hostname': socket.gethostname(),
        'status': 'ok',
        'memory_total_mb': round(total, 2),
        'memory_free_mb': round(free, 2),
        'memory_available_mb': round(available, 2),
        'memory_used_mb': round(used, 2),
        'memory_percent': round(mem_percent, 2),
        'cpu_load_1m': round(load1, 2),
        'cpu_load_5m': round(load5, 2),
        'cpu_load_15m': round(load15, 2),
        'disk_total_gb': round(disk_total, 2),
        'disk_free_gb': round(disk_free, 2),
        'disk_used_gb': round(disk_used, 2),
        'uptime_sec': round(read_uptime_sec(), 2),
    }


def push_snapshot(data):
    target = os.getenv('CRON_TARGET_URL', 'http://fission-console.fission.svc.cluster.local/cron/api/metrics')
    req = urllib.request.Request(
        target,
        data=json.dumps(data).encode('utf-8'),
        method='POST',
        headers={'Content-Type': 'application/json'},
    )
    token = os.getenv('CRON_TOKEN', '').strip()
    if token:
        req.add_header('X-Cron-Token', token)
    with urllib.request.urlopen(req, timeout=10) as resp:
        return resp.read().decode('utf-8', errors='ignore')


def main():
    data = payload()
    try:
        push_snapshot(data)
    except Exception as exc:
        data['push_error'] = str(exc)
    return {
        'status': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps(data, ensure_ascii=False),
    }

Что важно

• CRON_TARGET_URL можно оставить по умолчанию, если console доступна по публичному URL.
• Если захочешь защитить ingest, задай CRON_TOKEN в функции и такой же X-Cron-Token на стороне console.
• Функция не требует внешних библиотек.