如何实现一个 APM watchdog，你学会了吗？

theanarkh 2023-03-28 原文

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 内部的一些知识，实现起来有很多地方需要考虑的，希望后面有时间继续推进。

PR：https://github.com/nodejs/node/pull/45714

有关如何实现一个 APM watchdog，你学会了吗？的更多相关文章

ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2
我正在学习如何使用Nokogiri，根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div
ruby - 如何从 ruby 中的字符串运行任意对象方法？ - 2
总的来说，我对ruby还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用
python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声？ - 2
关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题？更新问题，以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点？我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等)，但我确实想创建一个输出文件。
ruby-on-rails - 如何验证 update_all 是否实际在 Rails 中更新 - 2
给定这段代码defcreate@upgrades=User.update_all(["role=?","upgraded"],:id=>params[:upgrade])redirect_toadmin_upgrades_path,:notice=>"Successfullyupgradeduser."end我如何在该操作中实际验证它们是否已保存或未重定向到适当的页面和消息？最佳答案在Rails3中，update_all不返回任何有意义的信息，除了已更新的记录数(这可能取决于您的DBMS是否返回该信息)。http://ar.ru
ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2
我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗？当我运行compasswatch时，它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行？文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们？我自己编译的.sass文件编译成compiled/t
ruby - 如何将脚本文件的末尾读取为数据文件(Perl 或任何其他语言) - 2
我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚
ruby - 如何指定 Rack 处理程序 - 2
Rackup通过Rack的默认处理程序成功运行任何Rack应用程序。例如:classRackAppdefcall(environment)['200',{'Content-Type'=>'text/html'},["Helloworld"]]endendrunRackApp.new但是当最后一行更改为使用Rack的内置CGI处理程序时，rackup给出“NoMethodErrorat/undefinedmethod`call'fornil:NilClass”:Rack::Handler::CGI.runRackApp.newRack的其他内置处理程序也提出了同样的反对意见。例如Rack
ruby - 使用 Vim Rails，您可以创建一个新的迁移文件并一次性打开它吗？ - 2
使用带有Rails插件的vim，您可以创建一个迁移文件，然后一次性打开该文件吗？textmate也可以这样吗？最佳答案你可以使用rails.vim然后做类似的事情::Rgeneratemigratonadd_foo_to_bar插件将打开迁移生成的文件，这正是您想要的。我不能代表textmate。关于ruby-使用VimRails，您可以创建一个新的迁移文件并一次性打开它吗？，我们在StackOverflow上找到一个类似的问题： https://sta
ruby-on-rails - Rails - 一个 View 中的多个模型 - 2
我需要从一个View访问多个模型。以前，我的links_controller仅用于提供以不同方式排序的链接资源。现在我想包括一个部分(我假设)显示按分数排序的顶级用户(@users=User.all.sort_by(&:score))我知道我可以将此代码插入每个链接操作并从View访问它，但这似乎不是“ruby方式”，我将需要在不久的将来访问更多模型。这可能会变得很脏，是否有针对这种情况的任何技术？注意事项:我认为我的应用程序正朝着单一格式和动态页面内容的方向发展，本质上是一个典型的网络应用程序。我知道before_filter但考虑到我希望应用程序进入的方向，这似乎很麻烦。最终从任何
ruby-on-rails - 渲染另一个 Controller 的 View - 2
我想要做的是有2个不同的Controller，client和test_client。客户端Controller已经构建，我想创建一个test_clientController，我可以使用它来玩弄客户端的UI并根据需要进行调整。我主要是想绕过我在客户端中内置的验证及其对加载数据的管理Controller的依赖。所以我希望test_clientController加载示例数据集，然后呈现客户端Controller的索引View，以便我可以调整客户端UI。就是这样。我在test_clients索引方法中试过这个:classTestClientdefindexrender:template=>

如何实现一个 APM watchdog，你学会了吗？

有关如何实现一个 APM watchdog，你学会了吗？的更多相关文章

随机推荐