"""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: if os.name == 'posix': os.killpg(process.pid, signal.SIGKILL) else: process.kill() 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