fix spectator.measurements metrics (#125)

copperlight · web-flow · commit a29681e3ef16 · 2025-04-22T16:15:31.000-04:00
* The `spectator.measurements` internal status metrics were not being published
on 200 OK responses, only when partial errors were observed. This change fixes
that, so that we always publish the sent metrics counter on 200 OK.
* The `spectator.measurements` metrics were using the `owner` tag to identify
the source of the metrics. Our internal dashboards use the `nf.process` tag for
this purpose, so we want to use that instead.
* The `ipc.client.call` metrics were using the `owner` tag to identify the source
of the metrics. The `owner` tag is better considered to be a way to identify
a source library, when there could be many sources of a metric. It is better to
use `nf.process` to identify these metrics.
* The `spectator.registrySize` metrics were using the `owner` tag to identify the
source of the metrics. This was swapped to `nf.process`.
* We do not want to set the `nf.process` tag with an environment variable for
`spectatord`, because there are many processes that send metrics through it,
each of which may want to set that tag independently. Thus, we set the
`nf.process` tag manually in the status metrics.
* The internal configuration library version was updated, so that we default to
the test endpoint for external publishing.
* The command line parameter parsing was updated, so that the `--uri` parameter
can be used to switch back to the prod endpoint, for external publishing, if
there is a need.
diff --git a/.gitignore b/.gitignore
@@ -5,6 +5,7 @@
 CMakeUserPresets.json
 cmake-build-debug/
 cmake-build/
+conan_provider.cmake
 metatron/auth_context.pb.cc
 metatron/auth_context.pb.h
 metatron/auth_context.proto
diff --git a/bin/spectatord_main.cc b/bin/spectatord_main.cc
@@ -111,17 +111,24 @@ auto main(int argc, char** argv) -> int {
 
   auto cfg = GetSpectatorConfig();
 
-  auto maybe_agg_uri = absl::GetFlag(FLAGS_uri);
-  if (absl::GetFlag(FLAGS_debug)) {
-    cfg->uri = "https://atlas-aggr-dev.us-east-1.ieptest.netflix.net/api/v4/update";
-  } else if (!maybe_agg_uri.empty()) {
-    cfg->uri = std::move(maybe_agg_uri);
-  } else if (absl::GetFlag(FLAGS_enable_external)) {
+  if (absl::GetFlag(FLAGS_enable_external)) {
     cfg->external_enabled = true;
     cfg->metatron_dir = absl::GetFlag(FLAGS_metatron_dir);
+    // provide a reasonable default for external publishing
     cfg->uri = cfg->external_uri;
   }
 
+  if (!absl::GetFlag(FLAGS_uri).empty()) {
+    // allow overriding the uri for both standard and external publishing
+    cfg->uri = absl::GetFlag(FLAGS_uri);
+  }
+
+  if (absl::GetFlag(FLAGS_debug)) {
+    // when running in debug mode, never publish externally
+    cfg->external_enabled = false;
+    cfg->uri = "https://atlas-aggr-dev.us-east-1.ieptest.netflix.net/api/v4/update";
+  }
+
   if (absl::GetFlag(FLAGS_verbose_http)) {
     cfg->verbose_http = true;
   }
diff --git a/conanfile.py b/conanfile.py
@@ -90,12 +90,12 @@ def get_flat_hash_map(self):
 
     def get_netflix_spectator_cppconf(self, nflx_cfg: NflxConfig) -> None:
         repo = "corp/cldmta-netflix-spectator-cppconf"
-        commit = "a76058e12a5e0928cc0a36acb1bfb8e045f9d589"
+        commit = "aa12add4ba33ac08002573d1a6563cad1b620e08"
         zip_name = repo.replace("corp/", "") + f"-{commit}.zip"
 
         self.maybe_remove_file(zip_name)
         self.download(nflx_cfg, repo, commit, zip_name)
-        check_sha256(self, zip_name, "9ffc2190f0ee676ddd9af31fcdd9a05cfbd990674101d7c8958638a0a432bdef")
+        check_sha256(self, zip_name, "7d72078e5e209ebaa1e7e861edf815c212353bcf2e8c9bf20de5304a2f9a7271")
 
         dir_name = repo.replace("corp/", "")
         self.maybe_remove_dir(dir_name)
diff --git a/spectator/http_client_test.cc b/spectator/http_client_test.cc
@@ -75,11 +75,15 @@ TEST(HttpTest, Post) {
 
   auto timer_for_req = find_timer(&registry, "ipc.client.call", "200");
   ASSERT_TRUE(timer_for_req != nullptr);
-  auto expected_tags =
-      Tags{{"owner", "spectatord"},      {"http.status", "200"},
-           {"http.method", "POST"},      {"ipc.status", "success"},
-           {"ipc.result", "success"},    {"ipc.endpoint", "/foo"},
-           {"ipc.attempt.final", "true"}};
+  auto expected_tags = Tags{
+    {"http.method", "POST"},
+    {"http.status", "200"},
+    {"ipc.attempt.final", "true"},
+    {"ipc.endpoint", "/foo"},
+    {"ipc.result", "success"},
+    {"ipc.status", "success"},
+    {"nf.process", "spectatord"},
+  };
 
   auto& actual_tags = timer_for_req->MeterId().GetTags();
   auto attempt = actual_tags.at(intern_str("ipc.attempt"));
@@ -133,11 +137,16 @@ TEST(HttpTest, PostUncompressed) {
 
   auto timer_for_req = find_timer(&registry, "ipc.client.call", "200");
   ASSERT_TRUE(timer_for_req != nullptr);
-  auto expected_tags =
-      Tags{{"owner", "spectatord"},   {"http.status", "200"},
-           {"http.method", "POST"},   {"ipc.status", "success"},
-           {"ipc.result", "success"}, {"ipc.attempt", "initial"},
-           {"ipc.endpoint", "/foo"},  {"ipc.attempt.final", "true"}};
+  auto expected_tags = Tags{
+    {"http.method", "POST"},
+    {"http.status", "200"},
+    {"ipc.attempt", "initial"},
+    {"ipc.attempt.final", "true"},
+    {"ipc.endpoint", "/foo"},
+    {"ipc.result", "success"},
+    {"ipc.status", "success"},
+    {"nf.process", "spectatord"},
+  };
 
   const auto& actual_tags = timer_for_req->MeterId().GetTags();
   EXPECT_EQ(expected_tags, actual_tags);
@@ -181,11 +190,16 @@ TEST(HttpTest, Timeout) {
   auto timer_for_req = find_timer(&registry, "ipc.client.call", "-1");
   ASSERT_TRUE(timer_for_req != nullptr);
 
-  auto expected_tags =
-      Tags{{"owner", "spectatord"},    {"http.status", "-1"},
-           {"ipc.result", "failure"},  {"ipc.status", "timeout"},
-           {"ipc.attempt", "initial"}, {"ipc.attempt.final", "true"},
-           {"ipc.endpoint", "/foo"},   {"http.method", "POST"}};
+  auto expected_tags = Tags{
+    {"http.method", "POST"},
+    {"http.status", "-1"},
+    {"ipc.attempt", "initial"},
+    {"ipc.attempt.final", "true"},
+    {"ipc.endpoint", "/foo"},
+    {"ipc.result", "failure"},
+    {"ipc.status", "timeout"},
+    {"nf.process", "spectatord"},
+  };
   EXPECT_EQ(expected_tags, timer_for_req->MeterId().GetTags());
 }
 
@@ -262,10 +276,16 @@ TEST(HttpTest, Get) {
   EXPECT_EQ(my_counters(registry).size(), 1);
 
   spectator::Tags timer_tags{
-      {"http.status", "200"},    {"ipc.attempt", "initial"},
-      {"ipc.result", "success"}, {"owner", "spectatord"},
-      {"ipc.status", "success"}, {"ipc.attempt.final", "true"},
-      {"http.method", "GET"},    {"ipc.endpoint", "/get"}};
+    {"http.method", "GET"},
+    {"http.status", "200"},
+    {"ipc.attempt", "initial"},
+    {"ipc.attempt.final", "true"},
+    {"ipc.endpoint", "/get"},
+    {"ipc.result", "success"},
+    {"ipc.status", "success"},
+    {"nf.process", "spectatord"},
+  };
+
   auto timer_id = Id::Of("ipc.client.call", std::move(timer_tags));
   auto timer = registry.GetTimer(timer_id);
   EXPECT_EQ(timer->Count(), 1);
@@ -294,21 +314,30 @@ TEST(HttpTest, Get503) {
   EXPECT_EQ(ipc_meters.size(), 2);
 
   spectator::Tags err_timer_tags{
-      {"http.status", "503"},       {"ipc.attempt", "initial"},
-      {"ipc.result", "failure"},    {"owner", "spectatord"},
-      {"ipc.status", "http_error"}, {"ipc.attempt.final", "false"},
-      {"http.method", "GET"},       {"ipc.endpoint", "/get503"}};
+    {"http.method", "GET"},
+    {"http.status", "503"},
+    {"ipc.attempt", "initial"},
+    {"ipc.attempt.final", "false"},
+    {"ipc.endpoint", "/get503"},
+    {"ipc.result", "failure"},
+    {"ipc.status", "http_error"},
+    {"nf.process", "spectatord"},
+  };
   spectator::Tags success_timer_tags{
-      {"http.status", "200"},    {"ipc.attempt", "second"},
-      {"ipc.result", "success"}, {"owner", "spectatord"},
-      {"ipc.status", "success"}, {"ipc.attempt.final", "true"},
-      {"http.method", "GET"},    {"ipc.endpoint", "/get503"}};
+    {"http.method", "GET"},
+    {"http.status", "200"},
+    {"ipc.attempt", "second"},
+    {"ipc.attempt.final", "true"},
+    {"ipc.endpoint", "/get503"},
+    {"ipc.result", "success"},
+    {"ipc.status", "success"},
+    {"nf.process", "spectatord"},
+  };
   auto err_id = Id::Of("ipc.client.call", std::move(err_timer_tags));
   auto err_timer = registry.GetTimer(err_id);
   EXPECT_EQ(err_timer->Count(), 1);
 
-  auto success_timer =
-      registry.GetTimer("ipc.client.call", std::move(success_timer_tags));
+  auto success_timer = registry.GetTimer("ipc.client.call", std::move(success_timer_tags));
   EXPECT_EQ(success_timer->Count(), 1);
 }
 
@@ -327,19 +356,23 @@ void test_method_header(const std::string& method) {
   std::vector<std::string> headers{"X-Spectator: foo", "X-Other-Header: bar"};
 
   auto url = fmt::format("http://localhost:{}/getheader", port);
-  auto response =
-      method == "GET" ? client.Get(url, headers) : client.Put(url, headers);
+  auto response = method == "GET" ? client.Get(url, headers) : client.Put(url, headers);
 
   server.stop();
   EXPECT_EQ(response.status, 200);
   EXPECT_EQ(response.raw_body, "x-other-header: bar\nx-spectator: foo\n");
   EXPECT_EQ(my_timers(registry).size(), 1);
 
   spectator::Tags timer_tags{
-      {"http.status", "200"},    {"ipc.attempt", "initial"},
-      {"ipc.result", "success"}, {"owner", "spectatord"},
-      {"ipc.status", "success"}, {"ipc.attempt.final", "true"},
-      {"http.method", method},   {"ipc.endpoint", "/getheader"}};
+    {"http.method", method},
+    {"http.status", "200"},
+    {"ipc.attempt", "initial"},
+    {"ipc.attempt.final", "true"},
+    {"ipc.endpoint", "/getheader"},
+    {"ipc.result", "success"},
+    {"ipc.status", "success"},
+    {"nf.process", "spectatord"},
+  };
   auto timer_id = Id::Of("ipc.client.call", std::move(timer_tags));
   auto timer = registry.GetTimer(timer_id);
   EXPECT_EQ(timer->Count(), 1);
diff --git a/spectator/log_entry.h b/spectator/log_entry.h
@@ -12,7 +12,7 @@ class LogEntry {
            const std::string& url)
       : registry_{registry},
         start_{absl::Now()},
-        id_{Id::Of("ipc.client.call", {{"owner", "spectatord"},
+        id_{Id::Of("ipc.client.call", {{"nf.process", "spectatord"},
                                        {"ipc.endpoint", PathFromUrl(url)},
                                        {"http.method", method},
                                        {"http.status", "-1"}})} {}
diff --git a/spectator/publisher.h b/spectator/publisher.h
@@ -28,7 +28,7 @@ namespace detail {
 template <typename R>
 auto get_counter(R* registry, Tags tags) -> std::shared_ptr<Counter> {
   static constexpr auto kSpectatorMeasurements = "spectator.measurements";
-  tags.add("owner", "spectatord");
+  tags.add("nf.process", "spectatord");
   return registry->GetCounter(kSpectatorMeasurements, std::move(tags));
 }
 }  // namespace detail
@@ -41,7 +41,8 @@ class Publisher {
         started_{false},
         should_stop_{false},
         last_successful_send_{absl::GetCurrentTimeNanos()},
-        sentMetrics_{detail::get_counter(registry, Tags{{"id", "sent"}})},
+        sentMetrics_{detail::get_counter(
+            registry, Tags{{"id", "sent"}})},
         invalidMetrics_{detail::get_counter(
             registry, Tags{{"id", "dropped"}, {"error", "validation"}})},
         droppedHttp_{detail::get_counter(
@@ -279,8 +280,12 @@ class Publisher {
     const auto& uri = registry_->GetConfig().uri;
     const auto& status_metrics_enabled = registry_->GetConfig().status_metrics_enabled;
     auto http_code = http_response.status;
+
     if (http_code == 200) {
       num_sent = num_measurements;
+      if (status_metrics_enabled) {
+        sentMetrics_->Add(static_cast<double>(num_sent));
+      }
     } else if (http_code > 200 && http_code < 500) {
       rapidjson::Document body;
       body.Parse(http_response.raw_body.c_str(), http_response.raw_body.size());
@@ -298,7 +303,7 @@ class Publisher {
           num_sent = num_measurements - err_count;
           if (status_metrics_enabled) {
             invalidMetrics_->Add(err_count);
-            sentMetrics_->Add(static_cast<double>(num_measurements - err_count));
+            sentMetrics_->Add(static_cast<double>(num_sent));
           }
           auto messages = body["message"].GetArray();
           for (auto& msg : messages) {
diff --git a/spectator/registry.cc b/spectator/registry.cc
@@ -12,7 +12,7 @@ Registry::Registry(std::unique_ptr<Config> config,
       meter_ttl_{absl::ToInt64Nanoseconds(config->meter_ttl)},
       config_{std::move(config)},
       logger_{std::move(logger)},
-      registry_size_{GetDistributionSummary("spectator.registrySize", Tags{{"owner", "spectatord"}})},
+      registry_size_{GetDistributionSummary("spectator.registrySize", Tags{{"nf.process", "spectatord"}})},
       publisher_(this) {
   if (meter_ttl_ == 0) {
     meter_ttl_ = int64_t(15) * 60 * 1000 * 1000 * 1000;
diff --git a/spectator/registry_test.cc b/spectator/registry_test.cc
@@ -184,7 +184,8 @@ TEST(Registry, DistSummary_Size) {
   r.GetCounter("bar")->Increment();
   // we have 4 measurements from the timer + 1 from the counter
   r.Measurements();
-  EXPECT_DOUBLE_EQ(r.GetDistributionSummary("spectator.registrySize", Tags{{"owner", "spectatord"}})->TotalAmount(), 5.0);
+  EXPECT_DOUBLE_EQ(r.GetDistributionSummary("spectator.registrySize",
+    Tags{{"nf.process", "spectatord"}})->TotalAmount(), 5.0);
 }
 
 TEST(Registry, DeleteMeters) {

Original file line number	Diff line number	Diff line change
`@@ -184,7 +184,8 @@ TEST(Registry, DistSummary_Size) {`
`184`	`184`	`r.GetCounter("bar")->Increment();`
`185`	`185`	`// we have 4 measurements from the timer + 1 from the counter`
`186`	`186`	`r.Measurements();`
`187`		`- EXPECT_DOUBLE_EQ(r.GetDistributionSummary("spectator.registrySize", Tags{{"owner", "spectatord"}})->TotalAmount(), 5.0);`
	`187`	`+ EXPECT_DOUBLE_EQ(r.GetDistributionSummary("spectator.registrySize",`
	`188`	`+ Tags{{"nf.process", "spectatord"}})->TotalAmount(), 5.0);`
`188`	`189`	`}`
`189`	`190`
`190`	`191`	`TEST(Registry, DeleteMeters) {`