Made stats sinks a statically registered component

DEPRECATED.md

@@ -24,3 +24,6 @@ The following features have been DEPRECATED and will be removed in the specified
 * The direction of network and HTTP filters in the configuration will be ignored from 1.4.0 and
   later removed from the configuration in the v2 APIs. Filter direction is now implied at the C++ type
   level.
+
+## Version 1.5.0
+* Config options `statsd_udp_ip_address` and `statsd_tcp_cluster_name` have been deprecated and replaced with a `stats_sinks` array of sink configurations.

docs/configuration/overview/overview.rst

@@ -19,6 +19,7 @@ specify miscellaneous configuration.
     "statsd_local_udp_port": "...",
     "statsd_udp_ip_address": "...",
     "statsd_tcp_cluster_name": "...",
+    "stats_sinks": [],
     "stats_flush_interval_ms": "...",
     "watchdog_miss_timeout_ms": "...",
     "watchdog_megamiss_timeout_ms": "...",
@@ -57,17 +58,22 @@ statsd_local_udp_port (Warning: DEPRECATED and will be removed in 1.4.0)
   *(optional, integer)* The UDP port of a locally running statsd compliant listener. If specified,
   :ref:`statistics <arch_overview_statistics>` will be flushed to this port.
 
-statsd_udp_ip_address
+statsd_udp_ip_address (Warning: DEPRECATED and will be removed in 1.5.0)
   *(optional, string)* The UDP address of a running statsd compliant listener. If specified,
   :ref:`statistics <arch_overview_statistics>` will be flushed to this address. IPv4 addresses should
   have format host:port (ex: 127.0.0.1:855). IPv6 addresses should have URL format [host]:port
   (ex: [::1]:855).
 
-statsd_tcp_cluster_name
+statsd_tcp_cluster_name (Warning: DEPRECATED and will be removed in 1.5.0)
   *(optional, string)* The name of a cluster manager cluster that is running a TCP statsd compliant
   listener. If specified, Envoy will connect to this cluster to flush :ref:`statistics
   <arch_overview_statistics>`.
 
+stats_sinks
+  *(optional, array)* An array of stats sink configuration objects.  Each object is required to
+  have a "name" string field (to look up the statically registered implementation of that sink) and
+  a "config" object field that contains the custom configuration for the sink.
+
 .. _config_overview_stats_flush_interval_ms:
 
 stats_flush_interval_ms

include/envoy/server/configuration.h

@@ -63,21 +63,9 @@ class Main {
   virtual RateLimit::ClientFactory& rateLimitClientFactory() PURE;
 
   /**
-   * @return Optional<std::string> the optional local/remote TCP statsd cluster to write to.
-   *         This cluster must be defined via the cluster manager configuration.
+   * @return std::list<Stats::SinkPtr>& the list of stats sinks initialized from the configuration.
    */
-  virtual Optional<std::string> statsdTcpClusterName() PURE;
-
-  // TODO(hennna): DEPRECATED - will be removed in 1.4.0.
-  /**
-   * @return Optional<uint32_t> the optional local UDP statsd port to write to.
-   */
-  virtual Optional<uint32_t> statsdUdpPort() PURE;
-
-  /**
-   * @return Optional<std::string> the optional UDP statsd address to write to.
-   */
-  virtual Optional<std::string> statsdUdpIpAddress() PURE;
+  virtual std::list<Stats::SinkPtr>& statsSinks() PURE;
 
   /**
    * @return std::chrono::milliseconds the time interval between flushing to configured stat sinks.

source/common/json/config_schemas.cc

@@ -1187,6 +1187,10 @@ const std::string Json::Schema::TOP_LEVEL_CONFIG_SCHEMA(R"EOF(
       "statsd_udp_ip_address" : {"type" : "string"},
       "statsd_tcp_cluster_name" : {"type" : "string"},
       "stats_flush_interval_ms" : {"type" : "integer"},
+      "stats_sinks" : {
+        "type" : "array",
+        "items" : {"type": "object"}
+      },
       "watchdog_miss_timeout_ms" : {"type" : "integer"},
       "watchdog_megamiss_timeout_ms" : {"type" : "integer"},
       "watchdog_kill_timeout_ms" : {"type" : "integer"},

source/exe/BUILD

@@ -47,6 +47,8 @@ envoy_cc_library(
         "//source/server/config/network:ratelimit_lib",
         "//source/server/config/network:redis_proxy_lib",
         "//source/server/config/network:tcp_proxy_lib",
+        "//source/server/config/stats:tcp_statsd_lib",
+        "//source/server/config/stats:udp_statsd_lib",
         "//source/server/http:health_check_lib",
     ],
 )

source/server/BUILD

@@ -235,7 +235,6 @@ envoy_cc_library(
         "//source/common/router:rds_lib",
         "//source/common/runtime:runtime_lib",
         "//source/common/singleton:manager_impl_lib",
-        "//source/common/stats:statsd_lib",
         "//source/common/upstream:cluster_manager_lib",
         "//source/server/http:admin_lib",
     ],

source/server/config/stats/BUILD

@@ -0,0 +1,30 @@
+licenses(["notice"])  # Apache 2
+
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+    name = "tcp_statsd_lib",
+    srcs = ["tcp_statsd.cc"],
+    hdrs = ["tcp_statsd.h"],
+    deps = [
+        "//source/common/stats:statsd_lib",
+        "//source/server:configuration_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "udp_statsd_lib",
+    srcs = ["udp_statsd.cc"],
+    hdrs = ["udp_statsd.h"],
+    deps = [
+        "//source/common/network:address_lib",
+        "//source/common/stats:statsd_lib",
+        "//source/server:configuration_lib",
+    ],
+)

source/server/config/stats/tcp_statsd.cc

@@ -0,0 +1,36 @@
+#include "server/config/stats/tcp_statsd.h"
+
+#include <string>
+
+#include "envoy/registry/registry.h"
+
+#include "common/stats/statsd.h"
+
+namespace Envoy {
+namespace Server {
+namespace Configuration {
+
+Stats::SinkPtr TcpStatsdSinkFactory::createStatsSink(const Json::Object& json_config,
+                                                     Server::Instance& server,
+                                                     Upstream::ClusterManager& cluster_manager) {
+  if (json_config.hasObject("cluster_name")) {
+    const std::string cluster_name = json_config.getString("cluster_name");
+    ENVOY_LOG(info, "statsd TCP cluster: {}", cluster_name);
+    return Stats::SinkPtr(new Stats::Statsd::TcpStatsdSink(
+        server.localInfo(), cluster_name, server.threadLocal(), cluster_manager, server.stats()));
+  }
+
+  throw EnvoyException(
+      fmt::format("Didn't find cluster_name in the {} Stats::Sink config", name()));
+}
+
+std::string TcpStatsdSinkFactory::name() { return "statsd_tcp"; }
+
+/**
+ * Static registration for the tcp statsd sink. @see RegisterFactory.
+ */
+static Registry::RegisterFactory<TcpStatsdSinkFactory, StatsSinkFactory> register_;
+
+} // namespace Configuration
+} // namespace Server
+} // namespace Envoy

source/server/config/stats/tcp_statsd.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <string>
+
+#include "envoy/server/instance.h"
+
+#include "server/configuration_impl.h"
+
+namespace Envoy {
+namespace Server {
+namespace Configuration {
+
+/**
+ * Config registration for the tcp statsd sink. @see StatsSinkFactory.
+ */
+class TcpStatsdSinkFactory : Logger::Loggable<Logger::Id::config>, public StatsSinkFactory {
+public:
+  // StatsSinkFactory
+  Stats::SinkPtr createStatsSink(const Json::Object& json_config, Instance& server,
+                                 Upstream::ClusterManager& cluster_manager) override;
+
+  std::string name() override;
+};
+
+} // namespace Configuration
+} // namespace Server
+} // namespace Envoy

source/server/config/stats/udp_statsd.cc

@@ -0,0 +1,53 @@
+#include "server/config/stats/udp_statsd.h"
+
+#include <string>
+
+#include "envoy/registry/registry.h"
+
+#include "common/network/address_impl.h"
+#include "common/stats/statsd.h"
+
+namespace Envoy {
+namespace Server {
+namespace Configuration {
+
+Stats::SinkPtr UdpStatsdSinkFactory::createStatsSink(const Json::Object& json_config,
+                                                     Server::Instance& server,
+                                                     Upstream::ClusterManager& cluster_manager) {
+  UNREFERENCED_PARAMETER(cluster_manager);
+  if (json_config.hasObject("local_port") && json_config.hasObject("ip_address")) {
+    throw EnvoyException(fmt::format(
+        "local_port and ip_address are mutually exclusive in the {} Stats::Sink config", name()));
+  }
+
+  if (json_config.hasObject("ip_address")) {
+    const std::string udp_ip_address = json_config.getString("ip_address");
+    ENVOY_LOG(info, "statsd UDP ip address: {}", udp_ip_address);
+    return Stats::SinkPtr(new Stats::Statsd::UdpStatsdSink(
+        server.threadLocal(), Network::Utility::parseInternetAddressAndPort(udp_ip_address)));
+  } else if (json_config.hasObject("local_port")) {
+    const int64_t local_upd_port = json_config.getInteger("local_port");
+    // TODO(hennna): DEPRECATED - statsdUdpPort will be removed in 1.4.0.
+    ENVOY_LOG(warn, "local_port has been DEPRECATED and will be removed in 1.4.0. "
+                    "Consider setting ip_address instead.");
+    ENVOY_LOG(info, "statsd UDP port: {}", local_upd_port);
+    Network::Address::InstanceConstSharedPtr address(
+        new Network::Address::Ipv4Instance(local_upd_port));
+    return Stats::SinkPtr(
+        new Stats::Statsd::UdpStatsdSink(server.threadLocal(), std::move(address)));
+  }
+
+  throw EnvoyException(
+      fmt::format("Didn't find local_port or ip_address in the {} Stats::Sink config", name()));
+}
+
+std::string UdpStatsdSinkFactory::name() { return "statsd_udp"; }
+
+/**
+ * Static registration for the udp statsd sink. @see RegisterFactory.
+ */
+static Registry::RegisterFactory<UdpStatsdSinkFactory, StatsSinkFactory> register_;
+
+} // namespace Configuration
+} // namespace Server
+} // namespace Envoy

source/server/config/stats/udp_statsd.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <string>
+
+#include "envoy/server/instance.h"
+
+#include "server/configuration_impl.h"
+
+namespace Envoy {
+namespace Server {
+namespace Configuration {
+
+/**
+ * Config registration for the udp statsd sink. @see StatsSinkFactory.
+ */
+class UdpStatsdSinkFactory : Logger::Loggable<Logger::Id::config>, public StatsSinkFactory {
+public:
+  // StatsSinkFactory
+  Stats::SinkPtr createStatsSink(const Json::Object& json_config, Instance& server,
+                                 Upstream::ClusterManager& cluster_manager) override;
+
+  std::string name() override;
+};
+
+} // namespace Configuration
+} // namespace Server
+} // namespace Envoy

source/server/configuration_impl.cc

@@ -60,24 +60,6 @@ void MainImpl::initialize(const Json::Object& json, const envoy::api::v2::Bootst
                               server.listenerManager()));
   }
 
-  if (json.hasObject("statsd_local_udp_port") && json.hasObject("statsd_udp_ip_address")) {
-    throw EnvoyException("statsd_local_udp_port and statsd_udp_ip_address "
-                         "are mutually exclusive.");
-  }
-
-  // TODO(hennna): DEPRECATED - statsd_local_udp_port will be removed in 1.4.0.
-  if (json.hasObject("statsd_local_udp_port")) {
-    statsd_udp_port_.value(json.getInteger("statsd_local_udp_port"));
-  }
-
-  if (json.hasObject("statsd_udp_ip_address")) {
-    statsd_udp_ip_address_.value(json.getString("statsd_udp_ip_address"));
-  }
-
-  if (json.hasObject("statsd_tcp_cluster_name")) {
-    statsd_tcp_cluster_name_.value(json.getString("statsd_tcp_cluster_name"));
-  }
-
   stats_flush_interval_ =
       std::chrono::milliseconds(json.getInteger("stats_flush_interval_ms", 5000));
 
@@ -102,6 +84,8 @@ void MainImpl::initialize(const Json::Object& json, const envoy::api::v2::Bootst
   } else {
     ratelimit_client_factory_.reset(new RateLimit::NullFactoryImpl());
   }
+
+  initializeStatsSinks(json, server);
 }
 
 void MainImpl::initializeTracers(const Json::Object& configuration, Instance& server) {
@@ -141,6 +125,76 @@ void MainImpl::initializeTracers(const Json::Object& configuration, Instance& se
   }
 }
 
+void MainImpl::initializeStatsSinks(const Json::Object& configuration, Instance& server) {
+  ENVOY_LOG(info, "loading stats sink configuration");
+
+  // TODO(mrice32): DEPRECATED - statsd_local_udp_port, statsd_udp_ip_address,
+  // statsd_tcp_cluster_name fields in the base json configuration will be moved to subobjects
+  // inside the stats_sinks array for 1.5.0.
+  if (configuration.hasObject("statsd_local_udp_port") &&
+      configuration.hasObject("statsd_udp_ip_address")) {
+    throw EnvoyException("statsd_local_udp_port and statsd_udp_ip_address "
+                         "are mutually exclusive.");
+  }
+
+  std::vector<Json::ObjectSharedPtr> sinks = configuration.getObjectArray("stats_sinks", true);
+
+  // Used to convert the deprecated statsd configuration into a json snippet for consumption by the
+  // new sink config parsing logic.
+  const std::string statsd_json = R"EOF(
+    {{
+      "name": "{}",
+      "config": {{
+        "{}": {}
+      }}
+    }}
+    )EOF";
+
+  // TODO(hennna): DEPRECATED - statsd_local_udp_port will be removed in 1.4.0.
+  if (configuration.hasObject("statsd_local_udp_port")) {
+    Json::ObjectSharedPtr json_obj = Json::Factory::loadFromString(
+        fmt::format(statsd_json, "statsd_udp", "local_port",
+                    configuration.getInteger("statsd_local_udp_port")));
+    sinks.emplace_back(std::move(json_obj));
+  }
+
+  if (configuration.hasObject("statsd_udp_ip_address")) {
+    Json::ObjectSharedPtr json_obj = Json::Factory::loadFromString(
+        fmt::format(statsd_json, "statsd_udp", "ip_address",
+                    "\"" + configuration.getString("statsd_udp_ip_address") + "\""));
+    sinks.emplace_back(std::move(json_obj));
+  }
+
+  if (configuration.hasObject("statsd_tcp_cluster_name")) {
+    Json::ObjectSharedPtr json_obj = Json::Factory::loadFromString(
+        fmt::format(statsd_json, "statsd_tcp", "cluster_name",
+                    "\"" + configuration.getString("statsd_tcp_cluster_name") + "\""));
+    sinks.emplace_back(std::move(json_obj));
+  }
+
+  for (const auto& sink_object : sinks) {
+    if (!sink_object->hasObject("name")) {
+      throw EnvoyException(
+          "sink object does not have 'name' attribute to look up the implementation");
+    }
+
+    if (!sink_object->hasObject("config")) {
+      throw EnvoyException(
+          "sink object does not contain the 'config' object to configure the implementation");
+    }
+
+    std::string name = sink_object->getString("name");
+    Json::ObjectSharedPtr sink_config = sink_object->getObject("config");
+
+    StatsSinkFactory* factory = Registry::FactoryRegistry<StatsSinkFactory>::getFactory(name);
+    if (factory != nullptr) {
+      stats_sinks_.emplace_back(factory->createStatsSink(*sink_config, server, *cluster_manager_));
+    } else {
+      throw EnvoyException(fmt::format("No Stats::Sink found for name: {}", name));
+    }
+  }
+}
+
 InitialImpl::InitialImpl(const Json::Object& json) {
   json.validateSchema(Json::Schema::TOP_LEVEL_CONFIG_SCHEMA);
   Json::ObjectSharedPtr admin = json.getObject("admin");