Hands-on 08-ups-monitoring-and-safe-shutdown: Lab

scope

All files created or changed by this lab remain under /opt/lab-classroom/class77/. The exercise uses simulated data and must not be connected to a production shutdown hook.

safety_boundary

Do not unplug a UPS or utility feed for this exercise.
Do not run an operating-system poweroff command.
Do not run a UPS output-removal command.
Do not install or edit system services, scheduled tasks, or production NUT configuration.
Run the commands as a user already authorized to write only to the classroom directory.

steps

step

1

title

Enter and confirm the classroom workspace

commands

cd /opt/lab-classroom/class77/
pwd

notes

The printed path must be exactly /opt/lab-classroom/class77 before creating any artifacts.

step

2

title

Create deterministic simulated UPS samples

commands

cat > samples.jsonl <<'EOF'
{"age_seconds":1,"battery_charge":100,"runtime_seconds":1800,"status":"OL"}
{"age_seconds":2,"battery_charge":78,"runtime_seconds":900,"status":"OB"}
{"age_seconds":2,"battery_charge":24,"runtime_seconds":280,"status":"OB LB"}
{"age_seconds":3,"battery_charge":22,"runtime_seconds":260,"status":"OB LB"}
{"age_seconds":4,"battery_charge":20,"runtime_seconds":240,"status":"OB LB"}
{"age_seconds":1,"battery_charge":26,"runtime_seconds":700,"status":"OL CHRG"}
{"age_seconds":2,"battery_charge":18,"runtime_seconds":220,"status":"OB LB"}
EOF

notes

The sequence contains online operation, an on-battery event, three consecutive critical samples, recovery, and one new critical sample after recovery.

step

3

title

Create the simulated shutdown decision engine

commands

cat > ups_guard.py <<'PY'
import json
import pathlib
import sys

ROOT = pathlib.Path('/opt/lab-classroom/class77').resolve()
STATE_PATH = ROOT / 'state.json'
EVENT_PATH = ROOT / 'events.jsonl'
MAX_AGE_SECONDS = 30
RUNTIME_THRESHOLD_SECONDS = 300
REQUIRED_CRITICAL_SAMPLES = 3

if pathlib.Path.cwd().resolve() != ROOT:
raise SystemExit('Refusing to run outside the class77 workspace')

if STATE_PATH.exists():
state = json.loads(STATE_PATH.read_text())
else:
state = {'low_count': 0, 'latched': False}

for sequence, line in enumerate(sys.stdin, start=1):
sample = json.loads(line)
age = sample.get('age_seconds')
runtime = sample.get('runtime_seconds')
tokens = set(str(sample.get('status', '')).split())
action = 'NO_ACTION'
reason = 'No policy condition matched'

if not isinstance(age, (int, float)) or age > MAX_AGE_SECONDS:
    state['low_count'] = 0
    action = 'DATA_STALE'
    reason = 'Telemetry is missing a valid age or is older than the permitted age'
elif 'OL' in tokens:
    state = {'low_count': 0, 'latched': False}
    action = 'ONLINE'
    reason = 'Utility power is reported available; incident latch cleared'
elif 'OB' in tokens:
    critical = 'LB' in tokens or (
        isinstance(runtime, (int, float)) and
        runtime <= RUNTIME_THRESHOLD_SECONDS
    )
    if critical:
        state['low_count'] += 1
        if state['low_count'] >= REQUIRED_CRITICAL_SAMPLES:
            if state['latched']:
                action = 'ALREADY_LATCHED'
                reason = 'A shutdown request was already issued for this incident'
            else:
                state['latched'] = True
                action = 'SHUTDOWN_REQUESTED'
                reason = 'Critical battery state was confirmed by consecutive samples'
        else:
            action = 'LOW_BATTERY_CONFIRMING'
            reason = 'Critical state observed but confirmation count is incomplete'
    else:
        state['low_count'] = 0
        action = 'ON_BATTERY_WAIT'
        reason = 'On battery without a confirmed critical condition'
else:
    state['low_count'] = 0
    action = 'UNKNOWN_STATUS'
    reason = 'Neither online nor on-battery status was reported'

event = {
    'action': action,
    'latched': state['latched'],
    'low_count': state['low_count'],
    'reason': reason,
    'sequence': sequence,
    'status_tokens': sorted(tokens)
}
with EVENT_PATH.open('a', encoding='utf-8') as event_file:
    event_file.write(json.dumps(event, sort_keys=True) + '\n')
STATE_PATH.write_text(json.dumps(state, sort_keys=True) + '\n', encoding='utf-8')
print(json.dumps(event, sort_keys=True))

PY

notes

The script checks its working directory before writing. Its decision output is a simulation label only; it contains no host shutdown operation.

step

4

title

Initialize state and run the simulation

commands

printf '%s\n' '{"latched": false, "low_count": 0}' > state.json
printf '' > events.jsonl
python3 ups_guard.py < samples.jsonl > actual-events.jsonl
cat actual-events.jsonl

notes

The fifth sample should produce SHUTDOWN_REQUESTED after three consecutive critical samples. Recovery clears the latch, and the final sample begins a new confirmation sequence without requesting shutdown.

step

5

title

Test stale telemetry handling

commands

printf '%s\n' '{"age_seconds":90,"battery_charge":10,"runtime_seconds":120,"status":"OB LB"}' > stale-sample.jsonl
printf '%s\n' '{"latched": false, "low_count": 0}' > state.json
printf '' > events.jsonl
python3 ups_guard.py < stale-sample.jsonl > stale-result.jsonl
cat stale-result.jsonl

notes

The action must be DATA_STALE rather than SHUTDOWN_REQUESTED because the simulated observation is older than the accepted age.