如何实现一个 APM watchdog
Hello，大家好，之前说不打算更新公众号了，后面有时间的话还是会偶尔更新下，记录和分享下一些技术相关的内容，今天分享下如何实现一个 APM watchdog。
在 APM 中，保证及时并准确地获取应用的信息是非常重要的，这样才能保证应用出现问题时，我们可以高效地找到并解决问题。本文以之前提交给 Node.js 的 PR 为例，介绍如何实现一个 APM watchdog 来对应用进行监控。这个 PR 的实现思想来自我们在内部实现的 APM watchdog，但是因为逻辑复杂，目前暂时还没有时间去推进。
首先来看一下如何使用，然后看看一下如何实现。
new MemoryProfileWatchdog({
// 内存阈值，达到该阈值则采集堆快照
maxRss: 1024 * 1024,
maxUsedHeapSize: 1024 * 1024,
// 轮询间隔
interval: 1000,
// 快照写到哪个文件
filename: filepath,
});可以看到，启动一个 watchdog 非常简单，我们只需要配置一些监控的阈值和轮训时间。监控的数据是基于定时轮询的，因为没有相关的订阅发布机制，当 watchdog 监控到数据达到阈值时就会采集堆快照，因为这里是一个内存 watchdog，我们也可以实现 CPU watchdog，原理是一样的。接着看看实现，首先看 JS 层的实现。
class MemoryProfileWatchdog {
#handle;
constructor(options) {
this.#handle = new profiler.MemoryProfileWatchdog({
...options,
filename,
});
this.#handle.start();
}
stop() {
if (this.#handle) {
this.#handle.stop();
this.#handle = null;
}
}
}JS 层的实现非常简单，只是对 C++ 层的简单封装，所以直接来看 C++ 层的实现，我们忽略一些细节，只关注核心逻辑。
class ProfileWatchdog : public BaseObject {
public:
enum class ProfileWatchdogState { kInitialized, kRunning, kClosing, kClosed };
ProfileWatchdog(Environment* env, v8::Local<v8::Object> object);
~ProfileWatchdog() override;
static v8::Local<v8::FunctionTemplate> GetConstructorTemplate(Environment* env);
// 启动 / 停止 watchdog
static void Start(const v8::FunctionCallbackInfo<v8::Value>& args);
static void Stop(const v8::FunctionCallbackInfo<v8::Value>& args);
void Start(Environment* env);
void Stop();
// 提交一个任务
template <typename Fn>
void AddTask(Fn&& cb, CallbackFlags::Flags flags = CallbackFlags::Flags::kRefed);
// 处理一个任务
void HandleTasks();
// 启动一个定时器
void SetTimeout();
// 定时器回调，具体的逻辑由子类实现
virtual bool TimeoutHandler() = 0;
protected:
// 轮询间隔
uint64_t interval_;
private:
static void Run(void* arg);
static void Timer(uv_timer_t* timer);
// 子线程
uv_thread_t thread_;
uv_loop_t loop_;
// 主线程和子线程的通信结构体
uv_async_t async_;
// 定时器
uv_timer_t timer_;
// 任务队列
CallbackQueue<void> tasks_;
Mutex task_mutex_;
};ProfileWatchdog 实现了 watchdog 机制，具体需要监控什么数据由子类实现，比如内存 watchdog。
class MemoryProfileWatchdog : public ProfileWatchdog {
public:
MemoryProfileWatchdog(Environment* env,
v8::Local<v8::Object> object,
v8::Local<v8::Object> options);
static void Init(Environment* env, v8::Local<v8::Object> target);
static void New(const v8::FunctionCallbackInfo<v8::Value>& args);
bool TimeoutHandler() override;
private:
// 需要监控的数据指标
size_t max_rss_ = 0;
size_t max_used_heap_size_ = 0;
std::string filename_;
};有了基本的了解后，接下来看具体实现。
void ProfileWatchdog::Start(Environment* env) {
int rc;
// 初始化一个事件循环结构体
rc = uv_loop_init(&loop_);
// 初始化线程间通信结构体
rc = uv_async_init(&loop_, &async_, [](uv_async_t* task_async) {
ProfileWatchdog* w = ContainerOf(&ProfileWatchdog::async_, task_async);
w->HandleTasks();
});
// 初始化并启动一个定时器
rc = uv_timer_init(&loop_, &timer_);
rc = uv_timer_start(&timer_, &ProfileWatchdog::Timer, interval_, 0);
// 创建 watchdog 线程
rc = uv_thread_create(&thread_, &ProfileWatchdog::Run, this);
}当启动一个 watchdog 时就会执行 Start，Start 函数中主要初始化了线程间通信的结构体，然后启动一个定时器，最后创建一个 watchdog 线程。因为 Node.js 是单线程的，为了保证 watchdog 在 JS 繁忙时仍可正常工作，我们需要借助子线程。创建子线程后，子线程就会开始执行 ProfileWatchdog::Run。
void ProfileWatchdog::Run(void* arg) {
ProfileWatchdog* wd = static_cast<ProfileWatchdog*>(arg);
uv_run(&wd->loop_, UV_RUN_DEFAULT);
CheckedUvLoopClose(&wd->loop_);
}Run 的逻辑很简单，就是启动一个事件循环，因为我们前面启动了一个定时器，所以这个事件循环里就会定时执行定时器回调 ProfileWatchdog::Timer。
void ProfileWatchdog::Timer(uv_timer_t* timer) {
ProfileWatchdog* w = ContainerOf(&ProfileWatchdog::timer_, timer);
// 往主线程插入一个任务
env->RequestInterrupt([watchdog = std::move(w)](Environment* env) {
// 执行定时器的逻辑，由具体的 watchdog 实现，返回 true 表示重启定时器，否则监控到此为止
if (watchdog->TimeoutHandler()) {
// 往子线程里插入一个任务，该任务是重启定时器
watchdog->AddTask(
[watchdog = std::move(watchdog)]() { watchdog->SetTimeout(); });
}
});
}Timer 中通过 env->RequestInterrupt 往主线程插入一个任务，因为有些代码是不能在子线程里执行的，另外 RequestInterrupt 可以保证在 JS 繁忙或阻塞在事件驱动模块时仍然可以执行我们的任务，那么这个任务具体做什么呢？看看内存 watchdog 的 TimeoutHandler 实现。
bool MemoryProfileWatchdog::TimeoutHandler() {
bool reached = false;
if (max_rss_) {
size_t rss = 0;
uv_resident_set_memory(&rss);
if (rss >= max_rss_) {
reached = true;
}
}
if (!reached && max_used_heap_size_) {
Isolate* isolate = env()->isolate();
HeapStatistics heap_statistics;
isolate->GetHeapStatistics(&heap_statistics);
if (heap_statistics.used_heap_size() >= max_used_heap_size_) {
reached = true;
}
}
// 内存达到阈值，采集快照
if (reached) {
HeapProfiler::HeapSnapshotOptions options;
options.numerics_mode = HeapProfiler::NumericsMode::kExposeNumericValues;
options.snapshot_mode = HeapProfiler::HeapSnapshotMode::kExposeInternals;
heap::WriteSnapshot(env(), filename_.c_str(), options);
// 采集完快照，停止 watchdog
return false;
}
return true;
}TimeoutHandler 就是获取主线程的内存信息，并判断是否超过了我们配置的阈值，是的话则采集堆快照并停止 watchdog，防止采集过多的重复信息，我们也可以改成隔久一点再开始重新监控，而内存如果没有超过阈值，则重启定时器，等待下一轮判断。从前面的代码可以看到，如果没有达到阈值，我们会调用 AddTask 往子线程插入一个任务。
watchdog->AddTask([watchdog = std::move(watchdog)]() {
watchdog->SetTimeout();
});看一下 AddTask 的实现。
template <typename Fn>
void ProfileWatchdog::AddTask(Fn&& cb, CallbackFlags::Flags flags) {
auto callback = tasks_.CreateCallback(std::move(cb), flags);
{
Mutex::ScopedLock lock(task_mutex_);
// 追加一个任务
tasks_.Push(std::move(callback));
}
// 通知子线程有任务处理
uv_async_send(&async_);
}AddTask 往子线程的任务队列中插入一个任务，并通知子线程处理，接着看看子线程如何处理任务。
void ProfileWatchdog::HandleTasks() {
while (tasks_.size() > 0) {
CallbackQueue<void> queue;
{
Mutex::ScopedLock lock(task_mutex_);
queue.ConcatMove(std::move(tasks_));
}
while (auto head = queue.Shift()) head->Call();
}
}HandleTasks 会逐个任务处理，也就是执行一个个函数，我们刚才插入的函数如下。
void ProfileWatchdog::SetTimeout() {
uv_timer_start(&timer_, &ProfileWatchdog::Timer, interval_, 0);
}也就是重启定时器，这样就开始等待下次超时，直到触发了阈值。
这就是 APM watchdog 的实现原理，核心思想是利用子线程和 env->RequestInterrupt 机制，保证我们对目的线程进行相对实时的监控（取决于设置的轮询时间），并在发现问题采集相关信息来协助我们排查问题，利用这个思路，我们可以实现不同类型的 watchdog 来解决不同的问题，比如 CPU watchdog 可以在 JS 死循环时采集 CPU Profile 信息帮助我们找到有问题的代码，本文就分享到这里，最后贴上目前的实现 PR（见文章末尾）。因为涉及到多线程和 Node.js 内部的一些知识，实现起来有很多地方需要考虑的，希望后面有时间继续推进。