278 lines
14 KiB
Python
278 lines
14 KiB
Python
"""Read-only Linux metrics with explicit site configuration.
|
|
|
|
Configuration is trusted operator input loaded from a local runtime file, NEVER
|
|
HTTP request data. Paths select installed scheduler wrappers, not shell text.
|
|
A wrapper accepts {slurm|lsf} followed by the fixed collector argv. It must be
|
|
owned and reviewed by the operator. Status means collection, not system health.
|
|
"""
|
|
import concurrent.futures
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
import math
|
|
import os
|
|
import re
|
|
import shlex
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
|
|
COMMANDS = {
|
|
'node': ['/bin/sh', '-c', 'cat /proc/meminfo && echo __LOAD__ && cat /proc/loadavg'],
|
|
'nfs': ['df', '-B1', '--output=source,fstype,size,used,avail,pcent,target', '-t', 'nfs', '-t', 'nfs4'],
|
|
'slurm_nodes': ['sinfo', '-N', '-h', '-o', '%N|%T|%c|%m'],
|
|
'slurm_jobs': ['squeue', '-h', '-o', '%T'],
|
|
'lsf_hosts': ['bhosts', '-w'],
|
|
'lsf_queues': ['bqueues', '-w'],
|
|
'lsf_jobs': ['bjobs', '-u', 'all', '-noheader', '-o', 'stat'],
|
|
}
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z')
|
|
|
|
|
|
def _nonnegative(value):
|
|
result = int(value)
|
|
if result < 0:
|
|
raise ValueError('negative counter')
|
|
return result
|
|
|
|
|
|
def parse_node(text):
|
|
memory, load = text.split('__LOAD__', 1)
|
|
values = {}
|
|
for line in memory.splitlines():
|
|
fields = line.split()
|
|
if fields and fields[0] in ('MemTotal:', 'MemAvailable:'):
|
|
if len(fields) != 3 or fields[2] != 'kB':
|
|
raise ValueError('invalid memory unit')
|
|
values[fields[0][:-1]] = _nonnegative(fields[1]) * 1024
|
|
total, available = values['MemTotal'], values['MemAvailable']
|
|
loads = [float(v) for v in load.split()[:3]]
|
|
if total <= 0 or available > total or len(loads) != 3 or any(not math.isfinite(v) or v < 0 for v in loads):
|
|
raise ValueError('invalid memory/load sample')
|
|
return {'memory_total_bytes': total, 'memory_available_bytes': available,
|
|
'memory_used_percent': round(100 * (total - available) / total, 2),
|
|
'load_1m': loads[0], 'load_5m': loads[1], 'load_15m': loads[2]}
|
|
|
|
|
|
def parse_nfs(text):
|
|
lines = text.strip().splitlines()
|
|
if not lines or not lines[0].startswith('Filesystem'):
|
|
raise ValueError('missing df header')
|
|
rows = []
|
|
for line in lines[1:]:
|
|
source, fstype, size, used, avail, percent, target = line.split(None, 6)
|
|
size, used, avail = map(_nonnegative, (size, used, avail))
|
|
percentage = _nonnegative(percent.rstrip('%'))
|
|
if fstype not in ('nfs', 'nfs4') or size <= 0 or not percent.endswith('%') or percentage > 100:
|
|
raise ValueError('invalid NFS sample')
|
|
rows.append({'mount_source': source, 'mount_target': target, 'fstype': fstype,
|
|
'total_bytes': size, 'used_bytes': used, 'available_bytes': avail,
|
|
'used_percent': percentage})
|
|
if not rows:
|
|
raise ValueError('no NFS mounts reported')
|
|
return rows
|
|
|
|
|
|
def parse_slurm_nodes(text):
|
|
nodes = {}
|
|
for line in text.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
name, state, cpus, memory = line.strip().split('|')
|
|
row = {'name': name, 'state': state, 'cpus': _nonnegative(cpus), 'memory_mb': _nonnegative(memory)}
|
|
if name in nodes and nodes[name] != row:
|
|
raise ValueError('inconsistent duplicate Slurm node')
|
|
nodes[name] = row
|
|
return {'count': len(nodes), 'by_state': dict(Counter(n['state'] for n in nodes.values())), 'nodes': list(nodes.values())}
|
|
|
|
|
|
def parse_jobs(text):
|
|
states = [line.strip() for line in text.splitlines() if line.strip()]
|
|
allowed = {'PENDING', 'RUNNING', 'SUSPENDED', 'COMPLETING', 'CONFIGURING', 'COMPLETED', 'CANCELLED', 'FAILED', 'TIMEOUT', 'NODE_FAIL', 'PREEMPTED', 'BOOT_FAIL', 'DEADLINE', 'OUT_OF_MEMORY', 'REQUEUED', 'REQUEUE_FED', 'REQUEUE_HOLD', 'RESIZING', 'REVOKED', 'SIGNALING', 'SPECIAL_EXIT', 'STAGE_OUT', 'STOPPED', 'PEND', 'RUN', 'PSUSP', 'USUSP', 'SSUSP', 'DONE', 'EXIT', 'UNKWN', 'ZOMBI', 'WAIT', 'PROV'}
|
|
if any(s not in allowed for s in states):
|
|
raise ValueError('unexpected scheduler state output')
|
|
return {'count': len(states), 'by_state': dict(Counter(states)), 'scope': 'visible jobs; no user/job details collected'}
|
|
|
|
|
|
def parse_lsf_table(text, kind):
|
|
lines = [line.split() for line in text.splitlines() if line.strip()]
|
|
first = 'HOST_NAME' if kind == 'hosts' else 'QUEUE_NAME'
|
|
required = ('STATUS', 'NJOBS', 'RUN') if kind == 'hosts' else ('STATUS', 'NJOBS', 'PEND', 'RUN', 'SUSP')
|
|
if not lines or lines[0][0] != first or any(k not in lines[0] for k in required):
|
|
raise ValueError('missing LSF table header')
|
|
header, rows = lines[0], []
|
|
for fields in lines[1:]:
|
|
if len(fields) != len(header):
|
|
raise ValueError('invalid LSF table row')
|
|
data = dict(zip(header, fields))
|
|
row = {'name': data[first], 'state': data['STATUS']}
|
|
for key in required:
|
|
if key != 'STATUS':
|
|
row[key.lower()] = _nonnegative(data[key])
|
|
rows.append(row)
|
|
return {'count': len(rows), 'by_state': dict(Counter(r['state'] for r in rows)), kind: rows,
|
|
'counter_note': 'LSF NJOBS/RUN table counters may count slots; jobs.count counts job records'}
|
|
|
|
|
|
PARSERS = {'node': parse_node, 'nfs': parse_nfs, 'slurm_nodes': parse_slurm_nodes,
|
|
'slurm_jobs': parse_jobs, 'lsf_jobs': parse_jobs,
|
|
'lsf_hosts': lambda s: parse_lsf_table(s, 'hosts'),
|
|
'lsf_queues': lambda s: parse_lsf_table(s, 'queues')}
|
|
|
|
|
|
|
|
def _path(value):
|
|
if not isinstance(value, str) or not re.fullmatch(r'/[A-Za-z0-9_./-]+', value) or '..' in value.split('/'):
|
|
raise ValueError('expected an absolute executable/identity path without shell syntax')
|
|
return value
|
|
|
|
|
|
def _host(value):
|
|
if not isinstance(value, str) or len(value) > 253 or not re.fullmatch(r'[A-Za-z0-9](?:[A-Za-z0-9.-]*[A-Za-z0-9])?', value):
|
|
raise ValueError('invalid hostname')
|
|
if any(not label or len(label) > 63 or label.startswith('-') or label.endswith('-') for label in value.split('.')):
|
|
raise ValueError('invalid hostname')
|
|
return value
|
|
|
|
|
|
def _settings(config):
|
|
keys = {'hosts', 'nodes', 'authorized_hosts', 'entry_hostname', 'storage_hosts', 'schedulers',
|
|
'wrapper_paths', 'ssh_user', 'ssh_identity_file', 'command_timeout_seconds', 'total_timeout_seconds'}
|
|
if not isinstance(config, dict) or set(config) - keys or ('hosts' in config and 'nodes' in config):
|
|
raise ValueError('unsupported or conflicting metrics configuration keys')
|
|
entry = _host(config.get('entry_hostname', 'localhost'))
|
|
allowed = config.get('authorized_hosts', [])
|
|
nodes = config.get('hosts', config.get('nodes', [entry]))
|
|
storage = config.get('storage_hosts', [])
|
|
schedulers = config.get('schedulers', ['slurm', 'lsf'])
|
|
for values in (allowed, nodes, storage, schedulers):
|
|
if not isinstance(values, list) or len(values) > 32 or any(not isinstance(v, str) for v in values) or len(values) != len(set(values)):
|
|
raise ValueError('invalid or duplicate collector targets')
|
|
for host in allowed + nodes + storage:
|
|
_host(host)
|
|
if any(h not in allowed and h != entry for h in nodes + storage):
|
|
raise ValueError('remote target not in authorized_hosts')
|
|
if any(s not in ('slurm', 'lsf') for s in schedulers):
|
|
raise ValueError('unsupported scheduler')
|
|
wrappers = config.get('wrapper_paths', {})
|
|
if not isinstance(wrappers, dict) or set(wrappers) - {'slurm', 'lsf'}:
|
|
raise ValueError('invalid scheduler wrapper mapping')
|
|
wrappers = {k: _path(v) for k, v in wrappers.items()}
|
|
user = config.get('ssh_user')
|
|
if user is not None and (not isinstance(user, str) or not re.fullmatch(r'[a-zA-Z_][a-zA-Z0-9_-]{0,63}', user)):
|
|
raise ValueError('invalid ssh_user')
|
|
if any(h != entry for h in nodes + storage) and not user:
|
|
raise ValueError('remote collection requires explicit ssh_user')
|
|
identity = config.get('ssh_identity_file')
|
|
if identity is not None:
|
|
_path(identity)
|
|
timeout = float(config.get('command_timeout_seconds', 5))
|
|
total = float(config.get('total_timeout_seconds', 20))
|
|
if not math.isfinite(timeout) or not 0.1 <= timeout <= 8 or not math.isfinite(total) or not 0.1 <= total <= 30:
|
|
raise ValueError('invalid collection timeout')
|
|
return dict(entry=entry, nodes=nodes, storage=storage, schedulers=schedulers,
|
|
wrappers=wrappers, user=user, identity=identity, timeout=timeout, total=total)
|
|
|
|
|
|
def _command(collector, settings):
|
|
command = list(COMMANDS[collector])
|
|
if '_' in collector:
|
|
scheduler = collector.split('_')[0]
|
|
wrapper = settings['wrappers'].get(scheduler)
|
|
if not wrapper:
|
|
return None
|
|
command = [wrapper, scheduler] + command
|
|
return command
|
|
|
|
|
|
def _run(host, collector, timeout, settings):
|
|
command = _command(collector, settings)
|
|
if command is None:
|
|
raise RuntimeError('not_configured: scheduler wrapper missing')
|
|
if host != settings['entry']:
|
|
remote = 'LC_ALL=C timeout -k 1s %ss %s' % (max(1, int(timeout)), shlex.join(command))
|
|
ssh = ['ssh', '-o', 'BatchMode=yes', '-o', 'StrictHostKeyChecking=yes',
|
|
'-o', 'ConnectTimeout=3', '-o', 'ConnectionAttempts=1',
|
|
'-o', 'ServerAliveInterval=2', '-o', 'ServerAliveCountMax=1']
|
|
if settings['identity']:
|
|
ssh += ['-i', settings['identity'], '-o', 'IdentitiesOnly=yes']
|
|
command = ssh + [settings['user'] + '@' + host, remote]
|
|
process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
text=True, env=dict(os.environ, LC_ALL='C'), start_new_session=True)
|
|
try:
|
|
stdout, stderr = process.communicate(timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
process.communicate()
|
|
raise TimeoutError('collector timed out')
|
|
if process.returncode:
|
|
if collector == 'lsf_jobs' and (stdout + stderr).strip().rstrip('.') == 'No unfinished job found':
|
|
return ''
|
|
category = 'permission denied' if 'permission denied' in stderr.lower() else 'command failed'
|
|
if 'not found' in stderr.lower() or process.returncode == 127:
|
|
category = 'command or environment unavailable'
|
|
raise RuntimeError('%s (exit %s)' % (category, process.returncode))
|
|
return stdout
|
|
|
|
|
|
def _sample(host, collector, settings, deadline):
|
|
command = _command(collector, settings)
|
|
record = {'host': host, 'collector': collector,
|
|
'source': host + ':' + (shlex.join(command) if command else collector),
|
|
'collected_at': _now(), 'status': 'error', 'error': None, 'data': None}
|
|
if command is None:
|
|
record.update(status='not_configured', error='scheduler wrapper not configured')
|
|
return record
|
|
try:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TimeoutError('collection deadline exceeded')
|
|
record['data'] = PARSERS[collector](_run(host, collector, min(settings['timeout'], remaining), settings))
|
|
record['status'] = 'ok'
|
|
except (OSError, RuntimeError, TimeoutError, ValueError, KeyError, IndexError) as exc:
|
|
record['error'] = str(exc) if isinstance(exc, (RuntimeError, TimeoutError)) else 'unavailable or malformed collector output (%s)' % type(exc).__name__
|
|
record['collected_at'] = _now()
|
|
return record
|
|
|
|
|
|
def collect(config):
|
|
"""Collect from trusted local configuration; default: localhost node only.
|
|
|
|
Optional schedulers appear as not_configured until wrapper_paths is supplied.
|
|
entry_hostname labels this process's host and must describe its deployment;
|
|
it never initiates an SSH connection. Remote hosts require authorized_hosts
|
|
AND ssh_user. The service must never accept these settings from web clients.
|
|
"""
|
|
result = {'collected_at': _now(), 'status': 'error', 'nodes': [], 'storage': [], 'schedulers': {}, 'errors': []}
|
|
try:
|
|
settings = _settings(config)
|
|
except (ValueError, TypeError) as exc:
|
|
result['errors'].append({'source': 'config', 'collected_at': _now(), 'error': str(exc)})
|
|
return result
|
|
tasks = [(h, 'node') for h in settings['nodes']] + [(h, 'nfs') for h in settings['storage']]
|
|
tasks += [(settings['entry'], c) for c in COMMANDS if '_' in c and c.split('_')[0] in settings['schedulers']]
|
|
deadline = time.monotonic() + settings['total']
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
|
|
records = list(executor.map(lambda pair: _sample(pair[0], pair[1], settings, deadline), tasks))
|
|
for record in records:
|
|
collector = record['collector']
|
|
if collector == 'node':
|
|
result['nodes'].append(record)
|
|
elif collector == 'nfs':
|
|
result['storage'].append(record)
|
|
else:
|
|
scheduler, part = collector.split('_', 1)
|
|
result['schedulers'].setdefault(scheduler, {})[part] = record
|
|
if record['error']:
|
|
result['errors'].append({k: record[k] for k in ('host', 'collector', 'source', 'collected_at', 'error')})
|
|
successful = sum(r['status'] == 'ok' for r in records)
|
|
result['status'] = 'ok' if records and successful == len(records) else 'partial' if successful else 'error'
|
|
if not records:
|
|
result['errors'].append({'source': 'config', 'collected_at': _now(), 'error': 'no collectors selected'})
|
|
result['collected_at'] = _now()
|
|
return result
|