Elasticsearch在日志检索和指标分析场景中承担着核心存储角色，但数据本身不会主动告诉我们什么时候出现了异常。要想在问题发生的第一时间收到通知，就需要一套监控告警机制。Elastic官方的X-Pack Watcher功能完备，不过它要求商业授权，并且配置JSON结构相对复杂，很多小团队更愿意自己写脚本实现。这篇文章会带你用Node.js结合官方客户端自己动手做一个灵活的watcher，通过定时查询和阈值判断来触发告警，代码量不大，但足够应对大多数业务监控需求。

一、先想清楚要监控什么以及查询怎么设计

动手写代码之前，先明确监控目标。常见的监控指标包括：错误日志数量、接口平均响应时间、消息队列积压、磁盘使用率等。这些指标通常已经以文档形式写入Elasticsearch，每条文档带有时间戳、级别、服务名等字段。要判断是否异常，往往需要对最近一段时间的数据做聚合统计。

以错误日志为例，可以每隔5分钟统计一次ERROR级别日志的数量，如果超过100条就认为服务异常。在Elasticsearch中，使用count API最直接，配合bool查询过滤级别和时间范围。时间范围可以用now-5m这种相对表达式，也可以用程序计算出的具体时间戳。下面是一个查询DSL示例：

const query = {
  bool: {
    must: [
      { term: { level: 'ERROR' } },
      { range: { timestamp: { gte: 'now-5m' } } }
    ]
  }
};

如果需要同时监控多个服务，可以在query中增加service字段的条件，或者使用aggregation按服务分组统计。如果数据量较大，建议使用count而不是search再获取总数，因为count只返回文档数量，开销更小。

查询频率也需要权衡。实时性要求高的场景可以30秒跑一次，但频繁查询会给ES集群带来额外负载，尤其在数据节点较少的情况下更要谨慎。一般业务监控1分钟到5分钟一次足够。Node.js中可以用setInterval实现简单轮询，也可以使用node-cron这样的库来支持更灵活的cron表达式。

二、用Node.js连接Elasticsearch并执行监控查询

首先安装官方客户端，命令是npm install @elastic/elasticsearch。这个库支持连接池、自动重试和节点发现，相比直接使用HTTP请求要可靠得多。初始化客户端时建议显式设置请求超时和重试次数，避免某个节点无响应导致整个监控任务卡住。

const { Client } = require('@elastic/elasticsearch');
const client = new Client({
  node: 'http://localhost:9200',
  maxRetries: 3,
  requestTimeout: 10000,
  // 如果开启了安全认证，可以这样写
  // auth: { username: 'elastic', password: 'your_password' }
});

接下来编写一个查询函数，负责去ES中拿回指标数据。这里以统计错误日志数量为例，通过client.count执行查询。注意count API的参数是index和body，body里放查询体。返回结果的count字段就是文档数量。

async function getErrorCount(index) {
  const result = await client.count({
    index,
    body: {
      query: {
        bool: {
          must: [
            { term: { level: 'ERROR' } },
            { range: { timestamp: { gte: 'now-5m' } } }
          ]
        }
      }
    }
  });
  return result.count;
}

拿到数据后，需要有一个调度逻辑定期调用这个函数。最简单的做法是用setInterval，但在生产环境中更推荐使用node-cron，因为它支持标准的cron表达式，也能避免多个执行实例重叠导致的并发问题。下面是一个node-cron的简单示例，每1分钟执行一次。

const cron = require('node-cron');
cron.schedule('* * * * *', async () => {
  try {
    const count = await getErrorCount('app-logs-*');
    console.log(`当前错误数量：${count}`);
  } catch (err) {
    console.error('查询失败', err);
  }
});

这段代码只是打印结果，真正的告警还需要阈值判断和状态管理。如果每次执行都检查是否超过阈值，那么连续多个周期都会满足条件，不可能每个周期都发一条告警，否则很快就会把通知渠道刷爆。这就引出了状态去重的问题。

三、实现告警状态去重与多渠道通知

状态去重的核心思想是：每个监控项维护一个当前状态，只有从正常变为异常时才发送告警，从异常恢复为正常时发送恢复通知；连续异常或连续正常都不重复发送。可以用一个简单的Map存储状态，键是监控项标识，值是布尔值true表示异常、false表示正常。

const alertState = new Map();
async function checkAndAlert(rule) {
  const count = await getErrorCount(rule.index);
  const isTriggered = count > rule.threshold;
  const previous = alertState.get(rule.name) || false;
  if (isTriggered && !previous) {
    await sendAlert(rule, count);
    alertState.set(rule.name, true);
  } else if (!isTriggered && previous) {
    await sendRecovery(rule, count);
    alertState.set(rule.name, false);
  }
}

上面的代码中，rule是一个配置对象，包含name、index和threshold等字段。sendAlert是发送告警通知的函数，sendRecovery发送恢复通知。这样就能有效避免告警轰炸。

通知渠道的选择很多。邮件可以用nodemailer，企业微信和钉钉都提供了群机器人的Webhook接口。下面以钉钉群机器人为例，发送一个文本消息。钉钉的接口地址和access_token需要提前在群设置中获取。

const https = require('https');
function sendDingTalkAlert(title, text) {
  const payload = JSON.stringify({
    msgtype: 'text',
    text: { content: `【${title}】${text}` }
  });
  const options = {
    hostname: 'oapi.dingtalk.com',
    path: '/robot/send?access_token=YOUR_ACCESS_TOKEN',
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Content-Length': Buffer.byteLength(payload)
    }
  };
  const req = https.request(options, (res) => {
    res.on('data', () => {});
  });
  req.on('error', (e) => console.error('钉钉通知失败', e));
  req.write(payload);
  req.end();
}

如果团队使用Slack或者其他IM工具，实现方式类似，都是发送HTTP POST请求。关键是要把Webhook地址和请求模板抽离成配置，方便替换渠道。

把前面几个部分组合起来，就可以形成一个可运行的最小监控程序。你可以把监控规则放在一个数组里，定时遍历每个规则执行checkAndAlert。这样增加新的监控项只需要添加配置即可，无需修改核心代码。

四、增强可靠性：失败重试、状态持久化与自身监控

在实际运行中，网络抖动、ES节点重启、查询超时等问题都会导致监控查询失败。如果查询失败就静默丢弃，可能会错过关键告警。因此在客户端初始化时要设置maxRetries和requestTimeout，在每次调用的catch块中记录错误，最好能结合简单的指数退避重试机制，把失败的任务重新入队。

async function runWithRetry(fn, maxAttempts = 3) {
  let attempt = 0;
  while (attempt < maxAttempts) {
    try {
      return await fn();
    } catch (err) {
      attempt++;
      if (attempt >= maxAttempts) throw err;
      const delay = Math.pow(2, attempt) * 1000;
      await new Promise(resolve => setTimeout(resolve, delay));
    }
  }
}

状态如果只保存在内存中，进程一旦重启，之前记录的异常状态就会丢失，可能导致恢复通知漏发或者重复告警。对于稳定性要求较高的场景，可以考虑把状态持久化到本地JSON文件或Redis。下面是一个简单的文件持久化实现。

const fs = require('fs');
const stateFile = './alert-state.json';
function loadState() {
  try {
    return new Map(Object.entries(JSON.parse(fs.readFileSync(stateFile, 'utf8'))));
  } catch {
    return new Map();
  }
}
function saveState(state) {
  fs.writeFileSync(stateFile, JSON.stringify(Object.fromEntries(state)), 'utf8');
}

在每次状态变化后调用saveState，进程启动时调用loadState恢复状态。如果使用Redis，可以利用其持久化特性，并且支持多实例共享状态，适合分布式部署。

最后，监控程序自身也需要被监控。可以给程序增加健康检查接口，或者定期输出运行日志，配合外部监控工具（如Prometheus、Zabbix）检测进程是否存活。日志应当包含每次查询的耗时、返回数量、是否触发告警等信息，方便事后排查。Node.js中可以使用pino或winston等日志库，避免直接用console.log难以控制级别和格式。

上面就是完整的实现思路和代码。相比官方Watcher，这种自建方案更加轻量灵活，不受商业授权限制，而且可以随意调整监控逻辑。对于中小规模团队来说，用一两个小时搭一套这样的告警系统，往往比深入学习X-Pack配置要划算得多。