vps/ansible/roles/healthcheck/templates/health-common.sh.j2

#!/usr/bin/env bash
# Shared contract: one JSON object per run in /var/lib/vps-health/latest.json.
# Status values: ok, warning, critical, unknown. Exit codes: 0 healthy/warning,
# 2 critical, 3 unknown. Check detail must never include secret values.
set -uo pipefail

RESULT_DIR='{{ healthcheck_state_dir }}'
LOG_DIR='{{ healthcheck_log_dir }}'
HOST_NAME="$(hostname -f 2>/dev/null || hostname)"
CHECK_NAME="$(basename "$0" .sh)"
STATUS=ok
EXIT_CODE=0
DETAILS=()

record() {
  local severity="$1" message="$2"
  DETAILS+=("${severity}:${message}")
  case "$severity" in
    critical) STATUS=critical; EXIT_CODE=2 ;;
    unknown) [[ "$STATUS" != critical ]] && STATUS=unknown; [[ "$EXIT_CODE" -eq 0 ]] && EXIT_CODE=3 ;;
    warning) [[ "$STATUS" == ok ]] && STATUS=warning ;;
  esac
}

require_command() {
  command -v "$1" >/dev/null 2>&1 || record unknown "missing_command:$1"
}

compose_ps() {
  docker compose --project-directory '{{ compose_project_dir }}' ps --all 2>&1
}

check_compose() {
  local output services bad
  # Only flag containers of *active* services (config --services excludes
  # debug/profile-gated services such as vaultwarden's pgweb, which is
  # intentionally stopped unless started with --profile debug).
  services="$(docker compose --project-directory '{{ compose_project_dir }}' config --services 2>/dev/null)" || { record critical 'compose_ps_failed'; return; }
  output="$(docker compose --project-directory '{{ compose_project_dir }}' ps --all --format json 2>&1)" || { record critical 'compose_ps_failed'; return; }
  bad="$(printf '%s\n' "$output" | python3 -c '
import json, sys
services = set(sys.argv[1].split())
for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    try:
        c = json.loads(line)
    except Exception:
        continue
    if c.get("Service") in services and c.get("State") in ("exited", "restarting", "dead"):
        print(c.get("Service"))
' "$services")"
  if [[ -n "$bad" ]]; then
    record critical 'compose_unhealthy_container'
  else
    record ok 'compose_ok'
  fi
}

check_backup_freshness() {
  local pattern="$1" newest now age
  newest="$(find {{ compose_project_dir | quote }} -type f -path "$pattern" -printf '%T@\n' 2>/dev/null | sort -nr | head -n1)"
  if [[ -z "$newest" ]]; then
    record warning 'backup_not_found'
    return
  fi
  now="$(date +%s)"
  age="$(( now - ${newest%.*} ))"
  if (( age > {{ healthcheck_backup_max_age_hours }} * 3600 )); then
    record critical 'backup_stale'
  else
    record ok 'backup_fresh'
  fi
}

check_https() {
  local url="$1" expected="$2" code
  code="$(curl --silent --show-error --output /dev/null --write-out '%{http_code}' --max-time 20 "$url" 2>/dev/null)" || {
    record critical 'https_request_failed'; return;
  }
  [[ "$code" =~ $expected ]] && record ok "https_${code}" || record critical "https_${code}"
}

check_tls_days() {
  local host="$1" port="$2" expires epoch remaining
  expires="$(timeout 20 openssl s_client -connect "${host}:${port}" -servername "$host" </dev/null 2>/dev/null | openssl x509 -noout -enddate 2>/dev/null | cut -d= -f2-)" || {
    record unknown 'tls_read_failed'; return;
  }
  epoch="$(date -d "$expires" +%s 2>/dev/null)" || { record unknown 'tls_date_parse_failed'; return; }
  remaining="$(( (epoch - $(date +%s)) / 86400 ))"
  if (( remaining < 0 )); then record critical 'tls_expired'
  elif (( remaining < {{ healthcheck_tls_warn_days }} )); then record warning 'tls_near_expiry'
  else record ok 'tls_valid'; fi
}

emit_result() {
  # Per-check JSON at latest-<check>.json. The dispatcher merges these into
  # latest.json so multiple profiles on one host do not overwrite each other.
  local tmp path detail_json
  path="${RESULT_DIR}/latest-${CHECK_NAME}.json"
  tmp="$(mktemp "${RESULT_DIR}/.latest-${CHECK_NAME}.XXXXXX")"
  detail_json="$(printf '%s\n' "${DETAILS[@]:-unknown:no_details}" | python3 -c 'import json,sys; print(json.dumps([line.rstrip() for line in sys.stdin if line.strip()]))')"
  python3 - "$tmp" "$HOST_NAME" "$CHECK_NAME" "$STATUS" "$EXIT_CODE" "$detail_json" <<'PY'
import json, sys
path, host, check, status, code, details = sys.argv[1:]
with open(path, 'w', encoding='utf-8') as f:
    json.dump({'schema': 1, 'host': host, 'check': check, 'status': status,
               'exit_code': int(code), 'details': json.loads(details)}, f,
              sort_keys=True, separators=(',', ':'))
    f.write('\n')
PY
  chmod 0640 "$tmp"
  mv "$tmp" "$path"
  cat "$path"
}

aggregate_result() {
  # Merge the just-run per-check files into latest.json. With a single complete
  # check this is a verbatim copy, preserving the historical one-object shape.
  # With several checks it emits one object whose status is the worst of all
  # checks; each check's own status/details are retained under `checks`. An
  # expected check with no fresh result file (profile crashed before writing)
  # is aggregated as `unknown`, so latest.json can never go stale while the
  # dispatcher reports a failure.
  case "$#" in
    0) return 0 ;;
    1) if [[ -f "${RESULT_DIR}/latest-$1.json" ]]; then
         cp -f "${RESULT_DIR}/latest-$1.json" "${RESULT_DIR}/latest.json"
       else
         python3 - "$RESULT_DIR" "$HOST_NAME" "$@" <<'PY'
import json, os, sys
rdir, host = sys.argv[1], sys.argv[2]
checks = sys.argv[3:]
levels = {'ok': 0, 'warning': 1, 'unknown': 2, 'critical': 3}
worst, worst_code = 'ok', 0
items = []
for c in checks:
    p = os.path.join(rdir, 'latest-%s.json' % c)
    if os.path.exists(p):
        d = json.load(open(p))
        st, code = d['status'], d['exit_code']
        items.append({'check': d['check'], 'status': st,
                      'exit_code': code, 'details': d['details']})
    else:
        st, code = 'unknown', 3
        items.append({'check': c, 'status': st, 'exit_code': code,
                      'details': ['unknown:check_did_not_complete']})
    if levels[st] > levels[worst]:
        worst, worst_code = st, code
out = {'schema': 1, 'host': host, 'check': 'aggregate', 'status': worst,
       'exit_code': worst_code, 'checks': items}
open(os.path.join(rdir, 'latest.json'), 'w').write(
    json.dumps(out, sort_keys=True, separators=(',', ':')) + '\n')
PY
       fi ;;
    *) python3 - "$RESULT_DIR" "$HOST_NAME" "$@" <<'PY'
import json, os, sys
rdir, host = sys.argv[1], sys.argv[2]
checks = sys.argv[3:]
levels = {'ok': 0, 'warning': 1, 'unknown': 2, 'critical': 3}
worst, worst_code = 'ok', 0
items = []
for c in checks:
    p = os.path.join(rdir, 'latest-%s.json' % c)
    if os.path.exists(p):
        d = json.load(open(p))
        st, code = d['status'], d['exit_code']
        items.append({'check': d['check'], 'status': st,
                      'exit_code': code, 'details': d['details']})
    else:
        st, code = 'unknown', 3
        items.append({'check': c, 'status': st, 'exit_code': code,
                      'details': ['unknown:check_did_not_complete']})
    if levels[st] > levels[worst]:
        worst, worst_code = st, code
out = {'schema': 1, 'host': host, 'check': 'aggregate', 'status': worst,
       'exit_code': worst_code, 'checks': items}
open(os.path.join(rdir, 'latest.json'), 'w').write(
    json.dumps(out, sort_keys=True, separators=(',', ':')) + '\n')
PY
  esac
  chmod 0640 "${RESULT_DIR}/latest.json"
  cat "${RESULT_DIR}/latest.json"
}