监控OGGMA同步状态与延迟告警

#!/usr/bin/env bash
# OGG MA 21c monitor - portable Bash + curl + Python3 stdlib, no jq.
set -uo pipefail
umask 077
PATH=/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin
SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
CONFIG=${OGG_MONITOR_CONFIG:-$SCRIPT_DIR/ogg_ma_monitor.conf}
[[ -r "$CONFIG" ]] || { echo "UNKNOWN: config unreadable: $CONFIG" >&2; exit 3; }
# Configuration is a trusted shell file; owner-only permissions recommended.
# shellcheck source=/dev/null
source "$CONFIG"
: "${STATE_DIR:=$SCRIPT_DIR/.ogg_monitor_state}"
: "${WARN_LAG:=300}" "${CRIT_LAG:=900}" "${STALE_SECONDS:=900}" "${REMIND_SECONDS:=3600}" "${CURL_TIMEOUT:=20}"
: "${MAIL_TO:=}" "${MAIL_FROM:=pu.se.mes@csisolar.com}" "${SMTP_HOST:=10.0.8.27}" "${SMTP_PORT:=25}" "${SMTP_TIMEOUT:=15}"
: "${AUTH_FILE:=$HOME/.ogg_monitor_curl.conf}"
mkdir -p "$STATE_DIR" || exit 3
exec 9>"$STATE_DIR/monitor.lock"
flock -n 9 || { echo 'UNKNOWN: previous run active'; exit 3; }
for bin in curl python3 flock; do command -v "$bin" >/dev/null || { echo "UNKNOWN: missing $bin"; exit 3; }; done
[[ -r "$AUTH_FILE" ]] || { echo "UNKNOWN: auth file unreadable: $AUTH_FILE"; exit 3; }
[[ ${#ENDPOINTS[@]} -gt 0 ]] || { echo 'UNKNOWN: ENDPOINTS empty'; exit 3; }
for n in "$WARN_LAG" "$CRIT_LAG" "$STALE_SECONDS" "$REMIND_SECONDS" "$CURL_TIMEOUT"; do [[ $n =~ ^[0-9]+$ ]] || { echo 'UNKNOWN: invalid numeric setting'; exit 3; }; done
(( WARN_LAG < CRIT_LAG )) || { echo 'UNKNOWN: WARN_LAG must be below CRIT_LAG'; exit 3; }
LOG="$STATE_DIR/monitor.log"; now=$(date +%s); overall=0; summaries=()
FORCE_MAIL=0
case "${1:-}" in
  --force-mail) FORCE_MAIL=1 ;;
  --test-mail|'') ;;
  -h|--help) printf 'Usage: %s [--test-mail|--force-mail|--help]\n' "$0"; exit 0 ;;
  *) printf 'UNKNOWN: unsupported argument: %s\n' "$1" >&2; exit 3 ;;
esac
log(){ printf '%s %s\n' "$(date '+%F %T %z')" "$*" | tee -a "$LOG"; }
fetch(){
  local url=$1 temp rc
  temp=$(mktemp "$STATE_DIR/resp.XXXXXX") || return 1
  HTTP=$(curl --digest --config "$AUTH_FILE" --silent --show-error --connect-timeout 5 --max-time "$CURL_TIMEOUT" -o "$temp" -w '%{http_code}' "$url" 2>>"$LOG"); rc=$?
  BODY=$(cat "$temp"); rm -f "$temp"
  [[ $rc -eq 0 && $HTTP == 200 ]] || return 1
  printf '%s' "$BODY" | python3 -c 'import json,sys; x=json.load(sys.stdin); assert isinstance(x.get("response"),dict); assert not any(isinstance(m,dict) and m.get("severity")=="ERROR" for m in x.get("messages",[]))' 2>>"$LOG"
}
# SMTP HTML mail through the configured relay; no local Postfix/sendmail path.
# A successful SMTP response means relay accepted the message, not final inbox delivery.
send_notice(){
  local level=$1 label=$2 role=$3 name=$4 endpoint=$5 status=$6 lag=$7 seq=$8 offset=$9 reason=${10} kind=${11}
  [[ -n $MAIL_TO && -n $MAIL_FROM && -n $SMTP_HOST ]] || { log 'MAIL ERROR: incomplete SMTP configuration'; return 1; }
  local lagtext issue subject cnlevel advice msg
  if [[ $lag =~ ^[0-9]+$ ]]; then lagtext="$((lag/3600))小时$(((lag%3600)/60))分$((lag%60))秒"; else lagtext='未知'; fi
  case $level in CRITICAL) cnlevel=严重;; WARNING) cnlevel=警告;; UNKNOWN) cnlevel=监控异常;; *) cnlevel=恢复;; esac
  if [[ $kind == RECOVERY ]]; then
    issue="$name 同步状态已恢复"; advice='继续观察延迟及检查点是否稳定。'
  elif [[ $level == UNKNOWN ]]; then
    issue="$name 监控数据采集异常"; advice='检查管理接口、认证和返回格式。监控失败不代表 OGG 进程停止。'
  elif [[ $status != running ]]; then
    issue="$name 进程状态异常（$status）"; advice='检查 Administration Server 中的进程报告及错误日志。'
  elif [[ $reason == *checkpoint* ]]; then
    issue="$name 检查点长时间未推进"; advice='先确认源端有新事务，再检查 Trail 积压及 Replicat 报告。'
  else
    issue="$name 同步延迟 $lagtext"; advice='连续观察 Lag 和检查点；若持续增加，检查数据库表空间、等待事件和 OGG 报告。'
  fi
  subject="【OGG${cnlevel}】【${label}】${issue}"
  msg="进程：$name；状态：$status；延迟：$lagtext；原因：$reason；管理地址：$endpoint"
  if OGG_MAIL_TO="$MAIL_TO" OGG_MAIL_FROM="$MAIL_FROM" OGG_MAIL_SUBJECT="$subject" OGG_MAIL_TEXT="$msg" OGG_MAIL_ISSUE="$issue" OGG_MAIL_LEVEL="$cnlevel" OGG_MAIL_LAG="$lagtext" OGG_MAIL_STATUS="$status" OGG_MAIL_ENDPOINT="$endpoint" OGG_MAIL_REASON="$reason" OGG_MAIL_ADVICE="$advice" OGG_MAIL_LABEL="$label" OGG_MAIL_NAME="$name" OGG_MAIL_ROLE="$role" OGG_MAIL_SEQ="$seq" OGG_MAIL_OFFSET="$offset" OGG_SMTP_HOST="$SMTP_HOST" OGG_SMTP_PORT="$SMTP_PORT" OGG_SMTP_TIMEOUT="$SMTP_TIMEOUT" python3 - <<'PYMAIL' >>"$LOG" 2>&1
import os, smtplib, html, sys
from email.message import EmailMessage
from email.utils import formatdate, make_msgid
v=lambda k:os.environ.get('OGG_MAIL_'+k,'')
h=lambda k:html.escape(v(k),quote=True)
color={'严重':'#b42318','警告':'#b54708','恢复':'#067647','监控异常':'#6941c6'}.get(v('LEVEL'),'#344054')
subject=v('SUBJECT')
msg=EmailMessage()
msg['From']=v('FROM')
msg['To']=v('TO')
msg['Subject']=subject
msg['Date']=formatdate(localtime=True)
msg['Message-ID']=make_msgid()
msg.set_content(v('TEXT')+'\n')
body=f'''<!DOCTYPE html><html lang="zh-CN"><head><meta charset="utf-8"></head>
<body style="margin:0;background:#f4f6f8;font-family:Arial,'Microsoft YaHei',sans-serif;color:#1d2939">
<table role="presentation" cellpadding="0" cellspacing="0" width="100%" style="background:#f4f6f8"><tr><td align="center" style="padding:24px 12px">
<table role="presentation" cellpadding="0" cellspacing="0" width="640" style="max-width:640px;width:100%;background:#fff;border:1px solid #e4e7ec;border-radius:8px">
<tr><td style="padding:22px 26px;background:{color};color:#fff"><div style="font-size:13px">Oracle GoldenGate · {h('LEVEL')}告警</div><div style="font-size:22px;font-weight:bold;margin-top:8px">{h('ISSUE')}</div></td></tr>
<tr><td style="padding:26px"><div style="color:#667085;font-size:13px">当前同步延迟</div><div style="font-size:30px;font-weight:bold;color:{color};margin:8px 0 18px">{h('LAG')}</div>
<p style="line-height:1.7">{h('LABEL')} 的 {h('NAME')} 当前状态为 <b>{h('STATUS')}</b>。请结合延迟和检查点确认同步情况。</p>
<table role="presentation" cellpadding="9" cellspacing="0" width="100%" style="font-size:14px;border-collapse:collapse">
<tr style="background:#f9fafb"><td>监控环境</td><td><b>{h('LABEL')}</b></td></tr>
<tr><td>进程 / 类型</td><td>{h('NAME')} / {h('ROLE')}</td></tr>
<tr style="background:#f9fafb"><td>进程状态</td><td>{h('STATUS')}</td></tr>
<tr><td>判断依据</td><td>{h('REASON')}</td></tr>
<tr style="background:#f9fafb"><td>Trail 检查点</td><td>Seq {h('SEQ')} / Offset {h('OFFSET')}</td></tr>
<tr><td>管理地址</td><td><a href="{h('ENDPOINT')}">{h('ENDPOINT')}</a></td></tr></table>
<div style="margin-top:20px;padding:14px;background:#f9fafb;border-left:3px solid {color};line-height:1.7"><b>建议处理</b><br>{h('ADVICE')}</div>
<div style="margin-top:18px;font-size:12px;color:#667085">检查时间：{html.escape(__import__('datetime').datetime.now().astimezone().strftime('%Y-%m-%d %H:%M:%S %Z %z'))}</div>
</td></tr></table></td></tr></table></body></html>'''
msg.add_alternative(body,subtype='html')
try:
    recipients=[x.strip() for x in v('TO').replace(';',',').split(',') if x.strip()]
    if not recipients: raise ValueError('no recipients')
    with smtplib.SMTP(os.environ['OGG_SMTP_HOST'],int(os.environ['OGG_SMTP_PORT']),timeout=int(os.environ['OGG_SMTP_TIMEOUT'])) as smtp:
        refused=smtp.send_message(msg,from_addr=v('FROM'),to_addrs=recipients)
        if refused: raise RuntimeError('recipients refused: '+repr(refused))
    print('SMTP relay accepted message (final delivery unverified)')
except Exception as e:
    print('SMTP SEND ERROR:',str(e),file=sys.stderr)
    sys.exit(1)
PYMAIL
  then
    log "MAIL accepted by SMTP relay: $subject -> $MAIL_TO (inbox delivery unverified)"; return 0
  fi
  log "MAIL ERROR: SMTP submission failed: $subject"; return 1
}
if [[ ${1:-} == --test-mail ]]; then
  send_notice CRITICAL '测试环境' replicat R_TEST 'http://127.0.0.1:7810' running 3600 123 456 'SMTP HTML template test' ALERT
  exit $?
fi
# state: level|last_sent|last_checkpoint_move|seq|offset
monitor_item(){
  local label=$1 base=$2 role=$3 name=$4 liststatus=$5 key file oldlevel='' lastsent=0 lastmove=$now oldseq='' oldoffset=''
  key=$(printf '%s_%s_%s_%s' "$label" "$base" "$role" "$name" | sha256sum | cut -c1-20)
  file="$STATE_DIR/$key.state"
  if [[ -f $file ]]; then IFS='|' read -r oldlevel lastsent lastmove oldseq oldoffset < "$file" || true; fi
  [[ $lastsent =~ ^[0-9]+$ ]] || lastsent=0
  [[ $lastmove =~ ^[0-9]+$ ]] || lastmove=$now
  local status='' lag='' seq='' offset='' level=OK reason=OK fields='' httpinfo=''
  if [[ $liststatus == RECOVERED ]]; then
    level=OK; status=running; reason="list API recovered"
  elif [[ $liststatus != OK ]]; then
    level=UNKNOWN; status=unknown; reason="list API $liststatus"
  elif ! fetch "$base/services/v2/${role}s/$name/info/status"; then
    level=UNKNOWN; status=unknown; reason="status API failed HTTP ${HTTP:-000}"
  else
    if fields=$(printf '%s' "$BODY" | python3 -c '
import json,sys
r=json.load(sys.stdin).get("response")
if not isinstance(r,dict):sys.exit(2)
p=r.get("position")
if not isinstance(p,dict):p={}
for x in (r.get("status"),r.get("lag"),p.get("sequence"),p.get("offset")):
 print("" if x is None else str(x))
' 2>>"$LOG"); then
      status=$(printf '%s\n' "$fields" | sed -n '1p' | tr '[:upper:]' '[:lower:]')
      lag=$(printf '%s\n' "$fields" | sed -n '2p')
      seq=$(printf '%s\n' "$fields" | sed -n '3p')
      offset=$(printf '%s\n' "$fields" | sed -n '4p')
      [[ $lag =~ ^[0-9]+$ ]] || lag=''
      [[ $seq =~ ^[0-9]+$ ]] || seq=''
      [[ $offset =~ ^[0-9]+$ ]] || offset=''
      if [[ -z $status ]]; then level=UNKNOWN; reason='status missing';
      elif [[ $status != running ]]; then level=CRITICAL; reason="process status=$status";
      elif [[ -z $lag ]]; then level=UNKNOWN; reason='lag missing or invalid';
      elif (( lag >= CRIT_LAG )); then level=CRITICAL; reason="lag=${lag}s >= ${CRIT_LAG}s";
      elif (( lag >= WARN_LAG )); then level=WARNING; reason="lag=${lag}s >= ${WARN_LAG}s"; fi
      if [[ -n $seq && -n $offset ]]; then
        if [[ $seq != "$oldseq" || $offset != "$oldoffset" ]]; then lastmove=$now; fi
        if [[ $status == running && -n $lag ]] && (( lag >= WARN_LAG && now-lastmove >= STALE_SECONDS )); then
          level=CRITICAL; reason="checkpoint unchanged $((now-lastmove))s with lag=${lag}s (verify workload)"
        fi
      fi
    else
      level=UNKNOWN; status=unknown; reason='invalid JSON fields'
    fi
  fi
  case $level in OK) sev=0;; WARNING) sev=1;; CRITICAL) sev=2;; *) sev=3;; esac
  (( sev > overall )) && overall=$sev
  log "$label $role $name status=${status:-NA} lag=${lag:-NA}s seq=${seq:-NA} offset=${offset:-NA} $level $reason"
  summaries+=("$label $role $name | status=${status:-NA} | lag=${lag:-NA}s | $level | $reason")
  local notify=0 kind=ALERT
  if [[ $level != "$oldlevel" ]]; then
    if [[ $level != OK || ( -n $oldlevel && $oldlevel != OK ) ]]; then notify=1; fi
    [[ $level == OK ]] && kind=RECOVERY
  elif [[ $level != OK ]] && (( now-lastsent >= REMIND_SECONDS )); then notify=1; fi
  if (( FORCE_MAIL )) && [[ $level != OK ]]; then
    notify=1
    kind=ALERT
    log "MAIL forced: $label $role $name level=$level (alert history will be preserved)"
  elif (( ! notify )) && [[ $level != OK ]]; then
    local remaining=$(( REMIND_SECONDS - (now-lastsent) ))
    (( remaining < 0 )) && remaining=0
    log "MAIL skipped: duplicate $level alert for $label $name; last_sent_epoch=$lastsent; next_reminder_in=${remaining}s"
  fi
  if (( notify )); then
    if send_notice "$level" "$label" "$role" "$name" "$base" "$status" "$lag" "$seq" "$offset" "$reason" "$kind"; then
      if (( ! FORCE_MAIL )); then lastsent=$now; fi
    else
      log "MAIL retry pending: $label $name; SMTP submission failed"
      # Preserve last successfully notified level and time so next run retries.
      level=$oldlevel
    fi
  fi
  printf '%s|%s|%s|%s|%s\n' "$level" "$lastsent" "$lastmove" "$seq" "$offset" > "$file.tmp"
  mv -f "$file.tmp" "$file"
}
for entry in "${ENDPOINTS[@]}"; do
  IFS='|' read -r label base role <<< "$entry"
  if [[ -z $label || -z $base || ( $role != extract && $role != replicat ) ]]; then log "UNKNOWN: invalid ENDPOINTS entry: $entry"; overall=3; continue; fi
  base=${base%/}
  if ! fetch "$base/services/v2/${role}s"; then
    monitor_item "$label" "$base" "$role" '__API_LIST__' "HTTP_${HTTP:-000}"
    continue
  fi
  names=$(printf '%s' "$BODY" | python3 -c 'import json,sys; r=json.load(sys.stdin)["response"]; print("\n".join(x["name"] for x in r.get("items",[]) if isinstance(x,dict) and isinstance(x.get("name"),str)))' 2>>"$LOG")
  if [[ -z $names ]]; then monitor_item "$label" "$base" "$role" '__API_LIST__' 'EMPTY_LIST'; continue; fi
  # Track API-list recovery, even if the earlier list failed.
  key=$(printf '%s_%s_%s_%s' "$label" "$base" "$role" '__API_LIST__' | sha256sum | cut -c1-20)
  if [[ -f "$STATE_DIR/$key.state" ]]; then monitor_item "$label" "$base" "$role" '__API_LIST__' RECOVERED; fi
  while IFS= read -r name; do [[ -n $name ]] && monitor_item "$label" "$base" "$role" "$name" OK; done <<< "$names"
done
printf '\n=== OGG MA Monitoring Summary ===\n'; printf '%s\n' "${summaries[@]}"
exit "$overall"