src,lib,test: make metrics interval updates thread-safe

lib/nsolid.js

@@ -774,6 +774,11 @@ function updateConfig(config = {}) {
       if (normalized !== undefined) {
         nsolidConfig.assetsEnabled = normalized;
       }
+    } else if (key === 'interval') {
+      const normalized = parsePositiveFiniteNumber(config.interval);
+      if (normalized !== undefined) {
+        nsolidConfig.interval = normalized;
+      }
     } else if (key === 'traceSampleRate') {
       const normalized = parseTraceSampleRate(config.traceSampleRate);
       if (normalized !== undefined) {
@@ -825,10 +830,6 @@ function updateConfig(config = {}) {
   // initConfig object so it doesn't waste memory.
   initConfig = null;
 
-  if (typeof nsolidConfig.interval === 'number') {
-    binding.setMetricsInterval(nsolidConfig.interval);
-  }
-
   if (typeof nsolidConfig.pubkey === 'string' &&
       nsolidConfig.pubkey.length !== 40) {
     debug('[updateConfig] invalid pubkey; must be 40 bytes');
@@ -917,9 +918,11 @@ function initializeConfig(nsolidConfig) {
     'prod';
 
   // Metrics send interval
+  const envInterval = parsePositiveFiniteNumber(process.env.NSOLID_INTERVAL);
+  const pkgInterval = parsePositiveFiniteNumber(pkgConfig.nsolid.interval);
   nsolidConfig.interval =
-    +process.env.NSOLID_INTERVAL ||
-    pkgConfig.nsolid.interval ||
+    envInterval ??
+    pkgInterval ??
     DEFAULT_INTERVAL;
 
   nsolidConfig.tags = getTags(process.env.NSOLID_TAGS || pkgConfig.nsolid.tags);
@@ -1174,6 +1177,32 @@ function parseTraceSampleRate(value) {
 }
 
 
+function parsePositiveFiniteNumber(value) {
+  if (value === undefined || value === null) {
+    return undefined;
+  }
+
+  let normalized;
+  if (typeof value === 'number') {
+    normalized = value;
+  } else if (typeof value === 'string') {
+    const trimmedValue = StringPrototypeTrim(value);
+    if (trimmedValue === '') {
+      return undefined;
+    }
+    normalized = +trimmedValue;
+  } else {
+    return undefined;
+  }
+
+  if (!NumberIsFinite(normalized) || normalized <= 0) {
+    return undefined;
+  }
+
+  return normalized;
+}
+
+
 function genPackageList() {
   let main_path;
   let last_path;

src/nsolid/nsolid_api.cc

@@ -90,7 +90,6 @@ static void calculatePtiles(std::vector<double>* vals, double* med, double* nn);
 // any thread have been blocked longer than the threadshold passed to
 // OnBlockedLoopHook().
 constexpr uint64_t blocked_loop_interval = 100;
-uint64_t gen_ptiles_interval = 5000;
 constexpr uint64_t datapoints_q_interval = 100;
 constexpr size_t datapoints_q_max_size = 100;
 
@@ -1258,6 +1257,13 @@ void EnvList::UpdateConfig(const nlohmann::json& config) {
       update_continuous_profiler(contCpuProfile, contCpuProfileInterval);
     }
 
+    it = config.find("interval");
+    if (it != config.end() && !it->is_null()) {
+      // If interval changes, next timer cb the timer is reset.
+      uint64_t interval = it->get<uint64_t>();
+      gen_ptiles_interval_.store(interval, std::memory_order_relaxed);
+    }
+
     it = config.find("otlp");
     if (it != config.end() && !it->is_null()) {
       otlp::OTLPAgent::Inst()->start();
@@ -1755,7 +1761,9 @@ void EnvList::env_list_routine_(ns_thread*, EnvList* envlist) {
       blocked_loop_timer_cb_, blocked_loop_interval, blocked_loop_interval);
   CHECK_EQ(er, 0);
   er = envlist->gen_ptiles_timer_.start(
-      gen_ptiles_cb_, gen_ptiles_interval, gen_ptiles_interval);
+      gen_ptiles_cb_,
+      envlist->gen_ptiles_interval_.load(std::memory_order_relaxed),
+      envlist->gen_ptiles_interval_.load(std::memory_order_relaxed));
   CHECK_EQ(er, 0);
   envlist->blocked_loop_timer_.unref();
   envlist->gen_ptiles_timer_.unref();
@@ -1843,7 +1851,7 @@ void EnvList::blocked_loop_timer_cb_(ns_timer*) {
 // This is a dirty simple percentile tracker. Since we support streaming metrics
 // a better solution is to stream them all and do more advanced percentile
 // calculations where it won't affect the process.
-void EnvList::gen_ptiles_cb_(ns_timer*) {
+void EnvList::gen_ptiles_cb_(ns_timer* timer) {
   EnvList* envlist = EnvList::Inst();
   decltype(EnvList::env_map_) env_map;
 
@@ -1866,6 +1874,17 @@ void EnvList::gen_ptiles_cb_(ns_timer*) {
                            envinst_sp->http_server_median_,
                            envinst_sp->http_server99_ptile_);
   }
+
+  // Restart timer if interval changed.
+  uint64_t interval =
+    uv_timer_get_repeat(reinterpret_cast<uv_timer_t*>(timer->base_handle()));
+  uint64_t next_interval =
+      envlist->gen_ptiles_interval_.load(std::memory_order_relaxed);
+  if (next_interval != interval) {
+    int er = envlist->gen_ptiles_timer_.start(
+      gen_ptiles_cb_, next_interval, next_interval);
+    CHECK_EQ(er, 0);
+  }
 }
 
 
@@ -2951,13 +2970,6 @@ static void ResumeMetrics(const FunctionCallbackInfo<Value>& args) {
 }
 
 
-static void SetMetricsInterval(const FunctionCallbackInfo<Value>& args) {
-  CHECK(args[0]->IsNumber());
-  double interval = args[0].As<Number>()->Value();
-  gen_ptiles_interval = static_cast<uint64_t>(interval);
-}
-
-
 static void OnCustomCommand(const FunctionCallbackInfo<Value>& args) {
   Environment* env = Environment::GetCurrent(args);
   CHECK(args[0]->IsFunction());
@@ -3382,7 +3394,6 @@ void BindingData::Initialize(Local<Object> target,
   SetMethod(context, target, "getKernelVersion", GetKernelVersion);
   SetMethod(context, target, "pauseMetrics", PauseMetrics);
   SetMethod(context, target, "resumeMetrics", ResumeMetrics);
-  SetMethod(context, target, "setMetricsInterval", SetMetricsInterval);
   SetMethod(context, target, "onCustomCommand", OnCustomCommand);
   SetMethod(context, target, "customCommandResponse", CustomCommandResponse);
   SetMethod(context,
@@ -3519,7 +3530,6 @@ void BindingData::RegisterExternalReferences(
   registry->Register(GetKernelVersion);
   registry->Register(PauseMetrics);
   registry->Register(ResumeMetrics);
-  registry->Register(SetMetricsInterval);
   registry->Register(OnCustomCommand);
   registry->Register(CustomCommandResponse);
   registry->Register(AttachRequestToCustomCommand);

src/nsolid/nsolid_api.h

@@ -721,6 +721,7 @@ class EnvList {
   std::atomic<uint64_t> min_blocked_threshold_ = { UINT64_MAX };
   // TODO(trevnorris): Temporary until Console supports streaming metrics
   nsuv::ns_timer gen_ptiles_timer_;
+  std::atomic<uint64_t> gen_ptiles_interval_ = { 5000 };
   // exit data
   std::atomic<bool> exiting_ = { false };
   std::atomic<int> exit_code_;

test/agents/test-grpc-metrics-interval-reconfigure.mjs

@@ -0,0 +1,128 @@
+// Flags: --expose-internals
+import { mustSucceed } from '../common/index.mjs';
+import assert from 'node:assert';
+import { setTimeout as delay } from 'node:timers/promises';
+import {
+  GRPCServer,
+  TestClient,
+} from '../common/nsolid-grpc-agent/index.js';
+
+const INITIAL_INTERVAL = 600;
+const UPDATED_INTERVAL = 100;
+const FAST_INTERVAL_UPPER_BOUND = 300;
+const METRICS_TIMEOUT_MS = 1500;
+
+function waitForMetricsEvent(grpcServer, timeoutMs = METRICS_TIMEOUT_MS) {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      grpcServer.off('metrics', onMetrics);
+      reject(new Error(`Timed out waiting for metrics event after ${timeoutMs}ms`));
+    }, timeoutMs);
+
+    function onMetrics() {
+      clearTimeout(timer);
+      grpcServer.off('metrics', onMetrics);
+      resolve(Date.now());
+    }
+
+    grpcServer.on('metrics', onMetrics);
+  });
+}
+
+async function collectMetricsEventTimestamps(grpcServer,
+                                             count,
+                                             timeoutMs = METRICS_TIMEOUT_MS) {
+  const timestamps = [];
+  const deadline = Date.now() + timeoutMs;
+
+  while (timestamps.length < count) {
+    const remaining = deadline - Date.now();
+    if (remaining <= 0) {
+      break;
+    }
+
+    timestamps.push(await waitForMetricsEvent(grpcServer, remaining));
+  }
+
+  return timestamps;
+}
+
+async function runTest({ getEnv }) {
+  return new Promise((resolve) => {
+    const grpcServer = new GRPCServer();
+    grpcServer.start(mustSucceed(async (port) => {
+      const env = getEnv(port);
+      const opts = {
+        stdio: ['inherit', 'inherit', 'inherit', 'ipc'],
+        env,
+      };
+      const child = new TestClient([], opts);
+      const agentId = await child.id();
+      const initialConfig = await child.config({ interval: INITIAL_INTERVAL });
+      assert.strictEqual(initialConfig.interval, INITIAL_INTERVAL);
+
+      // Wait for one periodic metrics push so the next interval boundary is
+      // well-defined before reconfiguring.
+      await waitForMetricsEvent(grpcServer, INITIAL_INTERVAL * 3);
+
+      const { data, requestId } = await grpcServer.reconfigure(agentId, {
+        interval: UPDATED_INTERVAL,
+      });
+      assert.strictEqual(data.msg.common.requestId, requestId);
+      assert.strictEqual(Number(data.msg.body.interval), UPDATED_INTERVAL);
+
+      const updatedConfig = await child.config();
+      assert.strictEqual(updatedConfig.interval, UPDATED_INTERVAL);
+
+      const timestamps = await collectMetricsEventTimestamps(grpcServer,
+                                                             3,
+                                                             METRICS_TIMEOUT_MS);
+      assert.strictEqual(
+        timestamps.length,
+        3,
+        `Expected 3 metrics events after reconfigure, got ${timestamps.length}`,
+      );
+
+      const deltas = [
+        timestamps[1] - timestamps[0],
+        timestamps[2] - timestamps[1],
+      ];
+      for (const delta of deltas) {
+        assert.ok(
+          delta < FAST_INTERVAL_UPPER_BOUND,
+          `Expected fast metrics cadence after reconfigure, got ${delta}ms`,
+        );
+      }
+
+      await child.shutdown(0);
+      grpcServer.close();
+      resolve();
+    }));
+  });
+}
+
+const testConfigs = [
+  {
+    getEnv: (port) => {
+      return {
+        NODE_DEBUG_NATIVE: 'nsolid_grpc_agent',
+        NSOLID_GRPC_INSECURE: 1,
+        NSOLID_GRPC: `localhost:${port}`,
+      };
+    },
+  },
+  {
+    getEnv: (port) => {
+      return {
+        NODE_DEBUG_NATIVE: 'nsolid_grpc_agent',
+        NSOLID_GRPC_INSECURE: 1,
+        NSOLID_SAAS: `aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbbbbbbbbbbbbbbbbbbbbbbbbbbtesting.localhost:${port}`,
+      };
+    },
+  },
+];
+
+for (const testConfig of testConfigs) {
+  await runTest(testConfig);
+  await delay(100);
+}

test/fixtures/nsolid-interval-package.json

@@ -0,0 +1,7 @@
+{
+  "name": "nsolid-interval-fixture",
+  "version": "1.0.0",
+  "nsolid": {
+    "interval": 3000
+  }
+}

test/fixtures/test-nsolid-config-interval-env-script.js

@@ -0,0 +1,10 @@
+'use strict';
+
+require('../common');
+const nsolid = require('nsolid');
+
+nsolid.start();
+
+console.log(JSON.stringify({
+  interval: nsolid.config.interval,
+}));