昨天夜里 2:17，我站在葵涌机房的第 7 排机架。风冷呼呼直吹，机柜 42U 的门缝里漏出一点硬盘指示灯的绿光。刚坐下，NOC 电话进来：“某台业务服务器偶发 502，延迟上去了。”我盯着监控屏，R740xd 的 iDRAC 没告警，传统探针更像“看天气”。我习惯的第一步不是开 RDP 点点点，而是让 Windows 事件日志（Event Log） 说话，再用 PowerShell 把“说话”变成自动化告警与现场证据。这篇就是我在香港托管环境里，把 Windows Server 2016 的事件能力“榨干”的全过程与可复制方案。

一、现场环境与边界条件

1）硬件/软件基线

经验：在香港 IDC，出站 25 端口常被封，邮件告警务必走 587/465 的 SMTP 中继（SendGrid、企业邮或自建中继）。WinRM 只走管理 VLAN 的 5985/5986，避免暴露公网。

二、我如何设计“事件驱动”的监控与告警

先不是上来就“全量转发”，而是明确哪些事件一旦出现必须马上通知并保留取证。下面是我在 Windows Server 2016 上的最小可用（MVP）事件策略清单：

原则：少而准，并对易“风暴”的事件做聚合/去重。一开始把这 8 类打通，覆盖 80% 的事故。

三、事件汇聚：Windows Event Forwarding（WEF）落地

1）Collector 侧（统一汇聚）

# 以管理员在 Collector 上执行
wecutil qc /q      # 配置 Windows Event Collector 服务
sc config Wecsvc start= auto
net start Wecsvc

• 创建订阅（图形化最直观）：“事件查看器 → 订阅 → 创建订阅”
• 类型：源启动（Source-initiated）
• 事件通道：System、Security、Application
• 事件筛选：按上表的 ID 过滤（XML 见下）
• 传送设置：Minimize Latency（低延迟），批量大小 15，间隔 3060 秒

订阅筛选 XML 示例（多通道合并）（我常用一份模板）：

<QueryList>
  <Query Id="0" Path="System">
    <Select Path="System">
      *[System[(EventID=41 or EventID=6008 or EventID=129 or EventID=153 or EventID=55 or EventID=36874 or EventID=36888 or EventID=1014)]]
    </Select>
  </Query>
  <Query Id="1" Path="Security">
    <Select Path="Security">
      *[System[(EventID=4625 or EventID=4720 or EventID=4726 or EventID=4728 or EventID=4719)]]
    </Select>
  </Query>
  <Query Id="2" Path="Application">
    <Select Path="Application">
      *[System[(EventID=1309)]]
    </Select>
  </Query>
</QueryList>

提示：Collector 的目标日志默认是 ForwardedEvents。容量请调大（见后文）。

2）Source 侧（被监控主机）

# 启用 WinRM（管理 VLAN 内执行）
winrm quickconfig -q

#（可选）启用 HTTPS（推荐域内证书，或内部 CA）
# winrm create winrm/config/Listener?Address=*+Transport=HTTPS "@{Hostname='collector.example.local'; CertificateThumbprint='THUMB'}"
# 防火墙：放行 5985（HTTP）或 5986（HTTPS）

指向 Collector（工作组环境用注册表，域内建议 GPO）：

# 非域环境（或先行试点），在 Source 上设置 SubscriptionManager：
reg add HKLM\SOFTWARE\Microsoft\Windows\CurrentVersion\EventCollector\SubscriptionManager ^
 /v 1 /t REG_SZ ^
 /d "Server=http://<collectorFQDN>:5985/wsman/SubscriptionManager/WEC,Refresh=60" /f

坑 1：时间漂移会让 Collector 上事件打乱顺序，统计“近 5 分钟”会偏差。所有主机对齐 NTP（域控或内网 NTP），w32tm /query /status 自查。

坑 2：工作组环境认证困难？优先 HTTPS + 本地用户组权限（把需要的账户加入源机器的 Event Log Readers），或改成 Collector-initiated 并在凭据里指定“运行方式”。

四、统一告警脚本：我用 PowerShell 做的“胶水层”

很多人会直接在 WEF 层就配置“任务触发”，但我更偏好由 Collector 的一个脚本统一聚合、限流、分发（Slack/企业微信/邮件）。下面是我在生产里可跑的简化版（Windows PowerShell 5.1 兼容）。

核心思路：按 RecordId 增量读取 ForwardedEvents（或本地 System/Security/Application），规则匹配 + 限流，多渠道发送。

# C:\Ops\HKOps-Monitor.ps1
# 运行账户：本机管理员（建议专用 svc 账号）
# 依赖：无额外模块（5.1 原生）

$ErrorActionPreference = 'Stop'

# ==== 配置区 ====
$Config = @{
  UseForwarded = $true                        # Collector 上读 ForwardedEvents
  LogNames     = @('ForwardedEvents')         # 或 @('System','Security','Application')
  StateFile    = 'C:\ProgramData\HKOps\EventState.json'
  SMTP         = @{
    Server='smtp.sendgrid.net'; Port=587; UseTLS=$true
    User='apikey'; Password='YOUR_SENDGRID_KEY'
    From='ops@yourcorp.hk'; To=@('noc@yourcorp.hk')
  }
  Slack        = @{
    Webhook='https://hooks.slack.com/services/XXX/YYY/ZZZ'
    Enable=$true
  }
  # 规则表：可按需扩展 MessageLike/ProviderName 等
  Rules = @(
    @{ Name='Power-Kernel'; Log='System';      Id=@(41,6008); Level='Critical'; DedupMin=10 },
    @{ Name='Storage-Reset'; Log='System';     Id=@(129,153); Level='High';     DedupMin=15 },
    @{ Name='NTFS-Error';    Log='System';     Id=@(55);      Level='High';     DedupMin=60 },
    @{ Name='TLS-Fatal';     Log='System';     Id=@(36874,36888); Level='Medium'; Aggregate=5; WindowMin=5 },
    @{ Name='DNS-Timeout';   Log='System';     Id=@(1014);    Level='Warn';     Aggregate=5; WindowMin=5 },
    @{ Name='Login-Fail';    Log='Security';   Id=@(4625);    Level='Medium';   Aggregate=10; WindowMin=5 },
    @{ Name='User-Change';   Log='Security';   Id=@(4720,4726,4728,4719); Level='High'; DedupMin=10 },
    @{ Name='IIS-Error';     Log='Application';Id=@(1309);    Level='High';     DedupMin=10 }
  )
}

# ==== 工具函数 ====
function Read-State {
  if (Test-Path $Config.StateFile) {
    return Get-Content $Config.StateFile -Raw | ConvertFrom-Json
  } else {
    return [pscustomobject]@{ LastRecordIds = @{}; LastFire=@{} }
  }
}

function Save-State($state) {
  $dir = Split-Path $Config.StateFile
  if (!(Test-Path $dir)) { New-Item -ItemType Directory -Path $dir -Force | Out-Null }
  $state | ConvertTo-Json -Depth 5 | Set-Content -Path $Config.StateFile -Encoding UTF8
}

function Get-NewEvents($logName, $lastRecordId) {
  # 用 XPath 以 RecordId 递增读取
  $filter = "*[System[(EventRecordID > $lastRecordId)]]"
  try {
    return Get-WinEvent -LogName $logName -FilterXPath $filter -ErrorAction Stop
  } catch {
    # 兼容空状态
    return @()
  }
}

function Send-Mail($subject, $body) {
  try {
    if ($Config.SMTP.UseTLS) {
      [Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
    }
    Send-MailMessage -SmtpServer $Config.SMTP.Server -Port $Config.SMTP.Port `
      -UseSsl:$($Config.SMTP.UseTLS) -Credential (New-Object System.Management.Automation.PSCredential($Config.SMTP.User, (ConvertTo-SecureString $Config.SMTP.Password -AsPlainText -Force))) `
      -From $Config.SMTP.From -To $Config.SMTP.To -Subject $subject -Body $body
  } catch {
    Write-Warning "SMTP send failed: $_"
  }
}

function Send-Slack($text) {
  if (-not $Config.Slack.Enable) { return }
  try {
    $payload = @{ text = $text } | ConvertTo-Json
    Invoke-RestMethod -Method Post -Uri $Config.Slack.Webhook -Body $payload -ContentType 'application/json'
  } catch {
    Write-Warning "Slack send failed: $_"
  }
}

function Format-Event($e) {
  $time = ([DateTime]$e.TimeCreated).ToString('yyyy-MM-dd HH:mm:ss')
  $mach = $e.MachineName
  $chan = $e.LogName
  $id   = $e.Id
  $msg  = ($e.FormatDescription() -replace '\s+', ' ').Trim()
  return "[${time}] [$mach] [$chan/$id] $msg"
}

# ==== 主流程 ====
$state = Read-State
if (-not $state.LastRecordIds) { $state.LastRecordIds = @{} }
if (-not $state.LastFire)     { $state.LastFire     = @{} }

$allAlerts = @()

foreach ($log in $Config.LogNames) {
  if (-not $state.LastRecordIds.ContainsKey($log)) {
    # 初始化为当前最大 RecordId，避免第一次读出历史
    $cur = Get-WinEvent -LogName $log -MaxEvents 1 -ErrorAction SilentlyContinue
    $state.LastRecordIds[$log] = if ($cur) { $cur.RecordId } else { 0 }
    continue
  }

  $events = Get-NewEvents -logName $log -lastRecordId $state.LastRecordIds[$log] | Sort-Object RecordId
  if (-not $events) { continue }

  foreach ($e in $events) {
    # 规则匹配（按通道与 ID）
    $matched = $Config.Rules | Where-Object { ($_.Log -eq $e.Properties[0]) -or ($_.Log -eq $e.LogName) } |
                               Where-Object { $_.Id -contains $e.Id }

    foreach ($rule in $matched) {
      $key = "$($rule.Name)|$($e.MachineName)"
      $now = Get-Date

      # 限流策略
      $canFire = $true
      if ($rule.DedupMin) {
        if ($state.LastFire.ContainsKey($key)) {
          if ($now -lt ([datetime]$state.LastFire[$key]).AddMinutes($rule.DedupMin)) { $canFire = $false }
        }
      }

      # 聚合：把同类事件先放入桶，稍后统一发
      if ($rule.Aggregate) {
        $bucketKey = "AGG|$key"
        if (-not $state.LastFire.ContainsKey($bucketKey)) {
          $state.LastFire[$bucketKey] = @()
        }
        $list = [System.Collections.ArrayList]$state.LastFire[$bucketKey]
        [void]$list.Add($e)
        $state.LastFire[$bucketKey] = $list
        # 聚合窗口到期再发（这里简单：每次循环后统一处理）
      } elseif ($canFire) {
        $allAlerts += @{
          Subject = "[${($rule.Level)}][$($rule.Name)] $($e.MachineName) $($e.LogName)/$($e.Id)"
          Body    = (Format-Event $e)
        }
        $state.LastFire[$key] = $now
      }
    }
  }

  # 更新最新 RecordId
  $state.LastRecordIds[$log] = ($events | Select-Object -Last 1).RecordId
}

# 处理聚合桶（5 分钟窗口）
foreach ($rule in $Config.Rules | Where-Object { $_.Aggregate }) {
  $keys = $state.LastFire.Keys | Where-Object { $_ -like "AGG|$($rule.Name)*" }
  foreach ($k in $keys) {
    $list = [System.Collections.ArrayList]$state.LastFire[$k]
    if (-not $list -or $list.Count -eq 0) { continue }
    # 只聚合窗口内的
    $windowStart = (Get-Date).AddMinutes(-$rule.WindowMin)
    $inwin = $list | Where-Object { $_.TimeCreated -ge $windowStart }
    if ($inwin.Count -ge $rule.Aggregate) {
      $mach  = ($inwin | Select-Object -First 1).MachineName
      $title = "[${($rule.Level)}][AGG/$($rule.Name)] $mach x $($inwin.Count)"
      $body  = ($inwin | Select-Object -First 20 | ForEach-Object { Format-Event $_ }) -join "`r`n"
      $allAlerts += @{ Subject=$title; Body=$body }
      # 清空已发（简单处理）
      $state.LastFire[$k] = @()
    }
  }
}

# 发送
foreach ($a in $allAlerts) {
  Send-Slack ("*{0}*`n```{1}```" -f $a.Subject, $a.Body)
  Send-Mail  $a.Subject $a.Body
}

Save-State $state

说明与取舍：

• RecordId 增量比 StartTime 稳妥（避免时钟漂移）。
• 规则表是核心，把不同事件映射到不同限流/聚合策略。
• Slack 失败不影响邮件，反之亦然。
• 第一次运行只初始化 RecordId，避免历史“倒灌”。

让脚本“永远在线”：任务计划程序

$taskName = "HKOps-EventMonitor"
$act = New-ScheduledTaskAction -Execute 'Powershell.exe' -Argument '-NoProfile -ExecutionPolicy Bypass -File "C:\Ops\HKOps-Monitor.ps1"'
$trg = New-ScheduledTaskTrigger -Once -At (Get-Date).AddMinutes(1) -RepetitionInterval (New-TimeSpan -Minutes 1) -RepetitionDuration ([TimeSpan]::MaxValue)
$pri = New-ScheduledTaskPrincipal -UserId 'SYSTEM' -RunLevel Highest
Register-ScheduledTask -TaskName $taskName -Action $act -Trigger $trg -Principal $pri -Description 'EventLog unified alert loop'

我在 Collector 上每分钟跑一次，事件延迟基本 < 60 秒，足够“准实时”。

五、无 WEF 的“小团队版”：各主机本地脚本 + Webhook

如果你暂时没有 Collector，先在每台主机本地跑一个“瘦版”，只看 System/Security 关键事件并发 Slack/企业微信：

# 精简版：只看本地 System 的 129/153/55
$LastFile = 'C:\ProgramData\HKOps\local.state'
$last = if (Test-Path $LastFile) { Get-Content $LastFile } else { 0 }
$filter = "*[System[(EventRecordID > $last) and (EventID=129 or EventID=153 or EventID=55)]]"
$evts = Get-WinEvent -LogName 'System' -FilterXPath $filter | Sort-Object RecordId
if ($evts) {
  $txt = ($evts | Select-Object -Last 10 | ForEach-Object { "[{0}] {1}/{2} {3}" -f $_.TimeCreated,$_.LogName,$_.Id,$_.Message }) -join "`n"
  Invoke-RestMethod -Method Post -Uri 'https://hooks.slack.com/services/XXX/YYY/ZZZ' -Body (@{text=$txt} | ConvertTo-Json) -ContentType 'application/json'
  ($evts | Select-Object -Last 1).RecordId | Set-Content $LastFile
}

这版可 5 分钟跑一次，适合先把“告警线”拉起来，再过渡到 WEF 统一管理。

六、事件触发的“自动取证”：Task Scheduler 基于事件的触发

场景：System/129（存储重置）出现时，我会立刻抓一份性能计数器、Get-PhysicalDisk、Get-StorageSubSystem、Get-EventLog -Newest 的打包，落到 C:\Ops\Evidence\yyyyMMdd-HHmmss.zip，方便事后与 IDC 或存储厂商沟通。

1）触发器 XML（基于事件）

<QueryList>
  <Query Id="0" Path="System">
    <Select Path="System">*[System[(EventID=129)]]</Select>
  </Query>
</QueryList>

2）动作脚本（示例）

# C:\Ops\Capture-StorageEvidence.ps1
$ts = Get-Date -Format 'yyyyMMdd-HHmmss'
$dir = "C:\Ops\Evidence\$ts"; New-Item -ItemType Directory -Path $dir -Force | Out-Null

Get-PhysicalDisk | Format-List * | Out-File "$dir\physicaldisk.txt"
Get-StorageSubSystem | Format-List * | Out-File "$dir\storagesubsystem.txt"
typeperf "\PhysicalDisk(*)\Avg. Disk sec/Read" -sc 10 | Out-File "$dir\typeperf_read.csv"
typeperf "\PhysicalDisk(*)\Avg. Disk sec/Write" -sc 10 | Out-File "$dir\typeperf_write.csv"
Get-WinEvent -LogName System -MaxEvents 200 | Export-Clixml "$dir\system_last200.xml"

Compress-Archive -Path $dir -DestinationPath "C:\Ops\Evidence\$ts.zip" -Force

这样做，告警不仅“叫”，还会第一时间留下“指纹”，排障效率高很多。

七、安全审计：我常用的快速基线

Windows Server 2016 默认审计粒度不够。我上线时会统一下发：

# 关键审计策略（高级审计）
auditpol /set /category:"Logon/Logoff" /subcategory:"Logon" /success:enable /failure:enable
auditpol /set /category:"Account Management" /subcategory:"User Account Management" /success:enable /failure:enable
auditpol /set /category:"Policy Change" /subcategory:"Audit Policy Change" /success:enable /failure:enable
auditpol /set /category:"DS Access" /subcategory:"Directory Service Changes" /success:enable /failure:enable
# ……按需扩展

并把 Security 日志容量与保留策略调大（否则 4625 类容易滚掉）：

wevtutil sl Security /ms:134217728   # 128MB
wevtutil sl System   /ms:134217728
wevtutil sl Application /ms:134217728
# 保留策略（覆盖旧事件）
wevtutil sl Security /rt:true

八、容量与留存：别等到“日志被覆盖”才后悔

Collector 侧我每晚 02:30 备份一次日志：

$dst="D:\EventBackups\$(Get-Date -Format yyyyMMdd)"
New-Item -ItemType Directory -Path $dst -Force | Out-Null
wevtutil epl ForwardedEvents "$dst\ForwardedEvents.evtx"
wevtutil epl Security        "$dst\Security.evtx"
wevtutil epl System          "$dst\System.evtx"
wevtutil epl Application     "$dst\Application.evtx"

九、一次真实故障的小结（数据支撑）

7 月份一次延迟告警，我这边的 聚合规则在 5 分钟内抓到了 18 次 System/129，来自相同主机。事件风暴被 15 分钟去重压住，但取证脚本留了 4 份证据包。对比 typeperf 曲线后，和 IDC 确认是某条上联链路抖动，存储阵列日志里也能对上时间点。
修复后 24 小时内，Collector 的 ForwardedEvents 统计：

这就是“事件驱动 + 自动取证”的价值：不靠猜。

十、香港托管环境的常见坑与我的解决法

1. SMTP 25 被封 → 改 587，[Net.ServicePointManager]::SecurityProtocol = Tls12，使用 SendGrid/企业中继。
2. WinRM 认证失败（工作组） → 用 HTTPS + 内部 CA，或在客户端设置 TrustedHosts（管理网内使用）：winrm set winrm/config/client '@{TrustedHosts="<collectorFQDN>"}'。
3. 时间不准 → 所有主机指向同一 NTP（域控或 Stratum2），w32tm /resync。
4. Schannel 36888 风暴 → 统一 TLS1.2，关闭过时套件，校验反向代理/上游的 TLS 配置。
5. 事件日志容量太小 → wevtutil sl <Log> /ms:<bytes> 提前调大。
6. WEF 一直不来事件 → 检查 Wecsvc 状态、订阅状态（事件查看器里看“运行状况”）、源上的 EventLog-ForwardPlugin 日志。
7. Collector 自己成单点 → Collector 做成虚机对，定期导出订阅配置，或双 Collector 分摊来源。

十一、把“可复用”做成一键化（落地清单）

•  在 Collector 上：wecutil qc、创建订阅、扩容 ForwardedEvents。
•  在 Source 上：winrm quickconfig、指向 SubscriptionManager、放行 5985/5986。
•  部署 C:\Ops\HKOps-Monitor.ps1，注册计划任务每分钟运行。
•  可选：为关键事件（129/55/4625）建立 事件触发任务，绑定 Capture-StorageEvidence.ps1。
•  每晚导出 EVTX 备份；每周自检一次告警风暴（看聚合条数）。
•  文档化规则表与变更流程（谁可以改、怎么回滚）。

我不追求铺天盖地的可观测性平台，而是把 Windows Server 2016 自带的 Event Log + PowerShell 玩到“干净、可控、能取证”。在香港这种连接复杂、合规苛刻的环境里，这套做法部署快、依赖少、取证强。你可以先用“小团队版”起步，随着主机数量增长再切到 WEF 汇聚与统一告警。