client: add --envoy-stats-sinks for Envoy stats sink plugins

README.md

@@ -166,6 +166,7 @@ bazel-bin/nighthawk_client  [--user-defined-plugin-config <string>] ...
 [--latency-response-header-name <string>]
 [--stats-flush-interval-duration <duration>]
 [--stats-flush-interval <uint32_t>]
+[--envoy-stats-sinks <string>] ...
 [--stats-sinks <string>] ... [--no-duration]
 [--simple-warmup]
 [--rate-limiter-plugin-config <string>]
@@ -238,6 +239,18 @@ For example '1s' or '1.000000001s'. Mutually exclusive with
 Time interval (in seconds) between flushes to configured stats sinks.
 Mutually exclusive with --stats-flush-interval-duration. Default: 5.
 
+--envoy-stats-sinks <string>  (accepted multiple times)
+Stats sinks (in json) implemented as Envoy stats sink plugins,
+resolved through Envoy's own factory registry rather than Nighthawk's,
+so any Envoy sink linked into this binary can be used without a
+Nighthawk specific factory. This argument is intended to be specified
+multiple times, and can be combined with --stats-sinks. Sinks that
+hold a gRPC client are not usable yet, see the version history.
+Example (json): {name:"envoy.stat_sinks.dog_statsd"
+,typed_config:{"@type":"type.googleapis.com/envoy.config.metrics.v3.Do
+gStatsdSink",address:{socket_address:{address:"127.0.0.1"
+,port_value:8125}}}}
+
 --stats-sinks <string>  (accepted multiple times)
 Stats sinks (in json) where Nighthawk metrics will be flushed. This
 argument is intended to be specified multiple times. Example (json):

api/client/options.proto

@@ -308,6 +308,12 @@ message CommandLineOptions {
   // Optional set of stat sinks where Nighthawk metrics will be flushed to.
   repeated envoy.config.metrics.v3.StatsSink stats_sinks = 34;
 
+  // Stats sinks implemented as Envoy stats sink plugins, resolved through
+  // envoy.server.configuration.StatsSinkFactory rather than Nighthawk's own
+  // NighthawkStatsSinkFactory, so that any Envoy sink linked into the binary can be used without
+  // a Nighthawk specific factory. Sinks configured here and in stats_sinks are both flushed.
+  repeated envoy.config.metrics.v3.StatsSink envoy_stats_sinks = 124;
+
   oneof oneof_stats_flush_interval {
     // Time interval (number of seconds) between periodical flushes to
     // configured stats sinks. If not specified the default is 5 seconds.

docs/root/version_history.md

@@ -14,6 +14,8 @@ Version history
 
 ### Changelist
 
+- `--envoy-stats-sinks` configures stats sinks implemented as Envoy stats sink plugins, resolved through `Envoy::Server::Configuration::StatsSinkFactory` rather than Nighthawk's own `NighthawkStatsSinkFactory`, so any Envoy sink linked into the binary can be used without writing a Nighthawk specific factory. It can be combined with `--stats-sinks`. Envoy's UDP statsd and DogStatsD sinks are linked in. Sinks that hold a gRPC client, such as the OpenTelemetry and metrics service sinks, are not usable yet: Envoy binds the client to the dispatcher of the thread that creates the sink, while Nighthawk flushes stats from its own flush worker thread, so flushing one trips `ASSERT(isThreadSafe())`.
+- Nighthawk's latency statistics are now recorded into an Envoy store histogram as well as their own HdrHistogram/Circllhist, so they appear in `MetricSnapshot::histograms()`. Stats sinks that only read the snapshot on flush, such as the OpenTelemetry and metrics service sinks, previously received nothing for them: Nighthawk delivered samples exclusively through `deliverHistogramToSinks()`, which those sinks implement as a no-op. Sinks reading `onHistogramComplete()` are unaffected, and the emitted metric names are unchanged, since the mirror is created in the worker's `cluster.<n>.` scope. The mirror declares `Microseconds` and samples are converted into it, so sinks can scale it by unit rather than knowing Nighthawk's convention; Envoy's statsd sinks label every histogram `|ms` and scale only by unit, so an unspecified nanosecond histogram would be exported a factor of 10^6 out. The conversion costs sub-microsecond precision on the statsd path, which receives every sample as recorded rather than binned, until Envoy gains a nanosecond histogram unit. Nighthawk's own output still comes from the statistic itself and keeps its nanosecond resolution.
 - `--request-body-file` sends a file's bytes verbatim as the request body (binary safe; no Content-Type is set). The `RequestOptions.request_body` (bytes) field carries it over the gRPC service API. Mutually exclusive with `--request-body-size`.
 - `OptionsImpl::toCommandLineOptions()` now always emits `request_options.request_body_size`; it was only set when at least one `--request-header` was configured, so the size was lost on the gRPC service path otherwise.
 - The Envoy exception on the tunneling startup path is logged instead of printed to stdout, keeping stdout reserved for the formatted output.

extensions_build_config.bzl

@@ -7,6 +7,8 @@ EXTENSIONS = {
     "envoy.tracers.zipkin": "//source/extensions/tracers/zipkin:config",
     "envoy.transport_sockets.raw_buffer": "//source/extensions/transport_sockets/raw_buffer:config",
     "envoy.access_loggers.file": "//source/extensions/access_loggers/file:config",
+    "envoy.stat_sinks.statsd": "//source/extensions/stat_sinks/statsd:config",
+    "envoy.stat_sinks.dog_statsd": "//source/extensions/stat_sinks/dog_statsd:config",
     "envoy.clusters.dns": "//source/extensions/clusters/dns:dns_cluster_lib",
     "envoy.clusters.eds": "//source/extensions/clusters/eds:eds_lib",
     "envoy.clusters.static": "//source/extensions/clusters/static:static_cluster_lib",

include/nighthawk/client/options.h

@@ -104,6 +104,11 @@ class Options {
   virtual bool simpleWarmup() const PURE;
   virtual bool noDuration() const PURE;
   virtual std::vector<envoy::config::metrics::v3::StatsSink> statsSinks() const PURE;
+  /**
+   * @return std::vector<envoy::config::metrics::v3::StatsSink> sinks to be created through Envoy's
+   * own Server::Configuration::StatsSinkFactory.
+   */
+  virtual std::vector<envoy::config::metrics::v3::StatsSink> envoyStatsSinks() const PURE;
   virtual uint32_t statsFlushInterval() const PURE;
   virtual Envoy::Protobuf::Duration statsFlushIntervalDuration() const PURE;
   virtual std::string responseHeaderWithLatencyInput() const PURE;

source/client/options_impl.cc

@@ -401,6 +401,17 @@ OptionsImpl::OptionsImpl(int argc, const char* const* argv) {
       "googleapis.com/"
       "envoy.config.metrics.v3.StatsdSink\",tcp_cluster_name:\"statsd\"}}",
       false, "string", cmd);
+  TCLAP::MultiArg<std::string> envoy_stats_sinks(
+      "", "envoy-stats-sinks",
+      "Stats sinks (in json) implemented as Envoy stats sink plugins, resolved through Envoy's own "
+      "factory registry rather than Nighthawk's, so any Envoy sink linked into this binary can be "
+      "used without a Nighthawk specific factory. This argument is intended to be specified "
+      "multiple times, and can be combined with --stats-sinks. Sinks that hold a gRPC client are "
+      "not usable yet, see the version history. Example (json): "
+      "{name:\"envoy.stat_sinks.dog_statsd\",typed_config:{\"@type\":\"type.googleapis.com/"
+      "envoy.config.metrics.v3.DogStatsdSink\",address:{socket_address:{address:\"127.0.0.1\","
+      "port_value:8125}}}}",
+      false, "string", cmd);
 
   TCLAP::ValueArg<uint32_t> stats_flush_interval(
       "", "stats-flush-interval",
@@ -460,10 +471,10 @@ OptionsImpl::OptionsImpl(int argc, const char* const* argv) {
   // Verify that if --stats-flush-interval or --stats-flush-interval-duration is
   // set, then --stats-sinks must also be set.
   if ((stats_flush_interval.isSet() || stats_flush_interval_duration.isSet()) &&
-      !stats_sinks.isSet()) {
+      !stats_sinks.isSet() && !envoy_stats_sinks.isSet()) {
     throw MalformedArgvException(
         "if --stats-flush-interval or --stats-flush-interval-duration is set, "
-        "then --stats-sinks must also be set");
+        "then --stats-sinks or --envoy-stats-sinks must also be set");
   }
 
   TCLAP_SET_IF_SPECIFIED(requests_per_second, requests_per_second_);
@@ -622,6 +633,18 @@ OptionsImpl::OptionsImpl(int argc, const char* const* argv) {
       stats_sinks_.push_back(sink);
     }
   }
+  if (envoy_stats_sinks.isSet()) {
+    for (const std::string& stats_sink : envoy_stats_sinks.getValue()) {
+      envoy::config::metrics::v3::StatsSink sink;
+      try {
+        Envoy::MessageUtil::loadFromJson(stats_sink, sink,
+                                         Envoy::ProtobufMessage::getStrictValidationVisitor());
+      } catch (const Envoy::EnvoyException& e) {
+        throw MalformedArgvException(e.what());
+      }
+      envoy_stats_sinks_.push_back(sink);
+    }
+  }
   TCLAP_SET_IF_SPECIFIED(stats_flush_interval, stats_flush_interval_);
   if (stats_flush_interval_duration.isSet()) {
     if (Envoy::Protobuf::util::TimeUtil::FromString(stats_flush_interval_duration.getValue(),
@@ -978,6 +1001,9 @@ OptionsImpl::OptionsImpl(const nighthawk::client::CommandLineOptions& options) {
   for (const envoy::config::metrics::v3::StatsSink& stats_sink : options.stats_sinks()) {
     stats_sinks_.push_back(stats_sink);
   }
+  for (const envoy::config::metrics::v3::StatsSink& stats_sink : options.envoy_stats_sinks()) {
+    envoy_stats_sinks_.push_back(stats_sink);
+  }
   stats_flush_interval_ =
       PROTOBUF_GET_WRAPPED_OR_DEFAULT(options, stats_flush_interval, stats_flush_interval_);
   if (options.has_stats_flush_interval_duration()) {
@@ -1231,6 +1257,9 @@ CommandLineOptionsPtr OptionsImpl::toCommandLineOptionsInternal() const {
   for (const envoy::config::metrics::v3::StatsSink& stats_sink : stats_sinks_) {
     *command_line_options->add_stats_sinks() = stats_sink;
   }
+  for (const envoy::config::metrics::v3::StatsSink& stats_sink : envoy_stats_sinks_) {
+    *command_line_options->add_envoy_stats_sinks() = stats_sink;
+  }
   if (stats_flush_interval_duration_.seconds() > 0 || stats_flush_interval_duration_.nanos() > 0) {
     *command_line_options->mutable_stats_flush_interval_duration() = stats_flush_interval_duration_;
   } else {

source/client/options_impl.h

@@ -116,6 +116,9 @@ class OptionsImpl : public Options, public Envoy::Logger::Loggable<Envoy::Logger
   bool multiTargetUseHttps() const override { return multi_target_use_https_; }
   bool simpleWarmup() const override { return simple_warmup_; }
   bool noDuration() const override { return no_duration_; }
+  std::vector<envoy::config::metrics::v3::StatsSink> envoyStatsSinks() const override {
+    return envoy_stats_sinks_;
+  }
   std::vector<envoy::config::metrics::v3::StatsSink> statsSinks() const override {
     return stats_sinks_;
   }
@@ -211,6 +214,7 @@ class OptionsImpl : public Options, public Envoy::Logger::Loggable<Envoy::Logger
   bool simple_warmup_{false};
   bool no_duration_{false};
   std::vector<envoy::config::metrics::v3::StatsSink> stats_sinks_;
+  std::vector<envoy::config::metrics::v3::StatsSink> envoy_stats_sinks_;
   uint32_t stats_flush_interval_{5};
   Envoy::Protobuf::Duration stats_flush_interval_duration_;
   std::string latency_response_header_name_;

source/client/process_impl.cc

@@ -41,6 +41,7 @@
 #include "source/common/stats/tag_producer_impl.h"
 #include "source/common/thread_local/thread_local_impl.h"
 #include "source/exe/main_common.h"
+#include "source/server/configuration_impl.h"
 #include "source/server/null_overload_manager.h"
 #include "source/server/server.h"
 
@@ -879,6 +880,22 @@ void ProcessImpl::setupStatsSinks(const envoy::config::bootstrap::v3::Bootstrap&
         Envoy::Config::Utility::getAndCheckFactory<NighthawkStatsSinkFactory>(stats_sink);
     stats_sinks.emplace_back(factory.createStatsSink(store_root_.symbolTable()));
   }
+  // Sinks implemented as Envoy stats sink plugins. These are resolved from Envoy's own factory
+  // registry, so any Envoy sink linked into this binary works without a Nighthawk specific
+  // factory. They receive Nighthawk's statistics through the same paths Envoy's own stats take:
+  // MetricSnapshot::histograms() on flush, and onHistogramComplete() per sample.
+  for (const envoy::config::metrics::v3::StatsSink& stats_sink : options_.envoyStatsSinks()) {
+    ENVOY_LOG(info, "loading Envoy stats sink configuration in Nighthawk");
+    auto& factory =
+        Envoy::Config::Utility::getAndCheckFactory<Envoy::Server::Configuration::StatsSinkFactory>(
+            stats_sink);
+    Envoy::ProtobufTypes::MessagePtr message = Envoy::Config::Utility::translateToFactoryConfig(
+        stats_sink, Envoy::ProtobufMessage::getStrictValidationVisitor(), factory);
+    absl::StatusOr<Envoy::Stats::SinkPtr> sink_or_error =
+        factory.createStatsSink(*message, server_->serverFactoryContext());
+    THROW_IF_NOT_OK_REF(sink_or_error.status());
+    stats_sinks.emplace_back(std::move(sink_or_error.value()));
+  }
   for (std::unique_ptr<Envoy::Stats::Sink>& sink : stats_sinks) {
     store_root_.addSink(*sink);
   }
@@ -969,7 +986,7 @@ bool ProcessImpl::runInternal(OutputCollector& collector, const UriPtr& tracing_
 
       // Needs to happen as early as possible (before createWorkers()) in the instantiation to
       // preempt the objects that require stats.
-      if (!options_.statsSinks().empty()) {
+      if (!options_.statsSinks().empty() || !options_.envoyStatsSinks().empty()) {
         absl::StatusOr<Envoy::Stats::TagProducerPtr> producer_or_error =
             Envoy::Stats::TagProducerImpl::createTagProducer(bootstrap_.stats_config(),
                                                              envoy_options_.statsTags());
@@ -1063,7 +1080,11 @@ bool ProcessImpl::runInternal(OutputCollector& collector, const UriPtr& tracing_
 
       ENVOY_LOG(error, bootstrap_.DebugString());
 
-      if (!options_.statsSinks().empty()) {
+      // The flush worker takes ownership of the sinks, which the stats store holds references to
+      // via addSink(). It must therefore be created whenever any sink was configured, of either
+      // kind: otherwise the sinks are destroyed at the end of this scope and the next
+      // deliverHistogramToSinks() call dereferences freed memory.
+      if (!options_.statsSinks().empty() || !options_.envoyStatsSinks().empty()) {
         // There should be only a single live flush worker instance at any time.
         flush_worker_ = std::make_unique<FlushWorkerImpl>(
             stats_flush_interval, *api_, tls_, store_root_, stats_sinks, *cluster_manager_);
@@ -1091,9 +1112,10 @@ bool ProcessImpl::runInternal(OutputCollector& collector, const UriPtr& tracing_
     w->waitForCompletion();
   }
 
-  if (!options_.statsSinks().empty() && flush_worker_ != nullptr) {
+  if (flush_worker_ != nullptr) {
     // Stop the running dispatcher in flush_worker_. Needs to be called after all
-    // client workers are complete so that all the metrics can be flushed.
+    // client workers are complete so that all the metrics can be flushed. Gating this on the
+    // flush worker existing is enough; it is created for sinks of either kind.
     flush_worker_->exitDispatcher();
     flush_worker_->waitForCompletion();
   }

source/common/statistic_impl.cc

@@ -413,19 +413,47 @@ SinkableStatistic::~SinkableStatistic() {
 }
 
 Envoy::Stats::Histogram::Unit SinkableStatistic::unit() const {
-  return Envoy::Stats::Histogram::Unit::Unspecified;
+  return Envoy::Stats::Histogram::Unit::Microseconds;
 }
 
 Envoy::Stats::SymbolTable& SinkableStatistic::symbolTable() { return scope_.symbolTable(); }
 
+Envoy::Stats::Histogram& SinkableStatistic::storeHistogram() {
+  if (store_histogram_ == nullptr) {
+    bindStoreHistogram();
+  }
+  return *store_histogram_;
+}
+
+void SinkableStatistic::bindStoreHistogram() {
+  // name() is the statistic's id; the scope supplies the "cluster.<n>." prefix. Resolved once and
+  // cached, so recording a sample stays a pointer dereference.
+  store_histogram_ = &scope_.histogramFromString(name(), unit());
+}
+
+uint64_t SinkableStatistic::toStoreHistogramUnit(uint64_t nanoseconds) {
+  // Nighthawk records nanoseconds; the store histogram declares Microseconds. See unit().
+  // Rounded rather than truncated: truncation biases every sample low by up to a microsecond,
+  // which is a systematic error rather than a wash. Samples below half a microsecond still land
+  // in the mirror as zero -- see unit() for why that boundary is acceptable here.
+  return (nanoseconds + 500) / 1000;
+}
+
 SinkableHdrStatistic::SinkableHdrStatistic(Envoy::Stats::Scope& scope, std::optional<int> worker_id)
     : SinkableStatistic(scope, worker_id) {}
 
 void SinkableHdrStatistic::recordValue(uint64_t value) {
   HdrStatistic::addValue(value);
-  // Currently in Envoy Scope implementation, deliverHistogramToSinks() will flush the histogram
-  // value directly to stats Sinks.
-  scope_.store().deliverHistogramToSinks(*this, value);
+  // Recording into the store histogram delivers the sample to sinks implementing
+  // onHistogramComplete() and, unlike a direct deliverHistogramToSinks() call, also makes the
+  // statistic visible to sinks that only read MetricSnapshot::histograms() on flush, such as the
+  // OpenTelemetry and metrics service sinks. See SinkableStatistic::storeHistogram().
+  storeHistogram().recordValue(toStoreHistogramUnit(value));
+}
+
+void SinkableHdrStatistic::setId(absl::string_view id) {
+  HdrStatistic::setId(id);
+  bindStoreHistogram();
 }
 
 std::string SinkableHdrStatistic::tagExtractedName() const {
@@ -442,9 +470,13 @@ SinkableCircllhistStatistic::SinkableCircllhistStatistic(Envoy::Stats::Scope& sc
 
 void SinkableCircllhistStatistic::recordValue(uint64_t value) {
   CircllhistStatistic::addValue(value);
-  // Currently in Envoy Scope implementation, deliverHistogramToSinks() will flush the histogram
-  // value directly to stats Sinks.
-  scope_.store().deliverHistogramToSinks(*this, value);
+  // See the comment in SinkableHdrStatistic::recordValue().
+  storeHistogram().recordValue(toStoreHistogramUnit(value));
+}
+
+void SinkableCircllhistStatistic::setId(absl::string_view id) {
+  CircllhistStatistic::setId(id);
+  bindStoreHistogram();
 }
 
 std::string SinkableCircllhistStatistic::tagExtractedName() const {