enable external publishing (#66)

.gitignore

@@ -3,6 +3,10 @@
 .idea/
 cmake-build-debug
 cmake-build/
+metatron/auth_context.pb.cc
+metatron/auth_context.pb.h
+metatron/auth_context.proto
+metatron/metatron_config.cc
 ska/
 spectator/netflix_config.cc
 venv/

CMakeLists.txt

@@ -14,6 +14,7 @@ conan_basic_setup()
 
 add_subdirectory(admin)
 add_subdirectory(bench)
+add_subdirectory(metatron)
 add_subdirectory(server)
 add_subdirectory(spectator)
 add_subdirectory(tools)

README.md

@@ -24,9 +24,14 @@ spectatord: A daemon that listens for metrics and reports them to Atlas.
       default: "";
     --debug (Debug spectatord. All values will be sent to a dev aggregator and
       dropped.); default: false;
+    --enable_external (Enable external publishing.); default: false;
     --enable_socket (Enable UNIX domain socket support. Default is true on Linux
       and false on MacOS.); default: true;
     --enable_statsd (Enable statsd support.); default: false;
+    --metatron_dir (Path to the Metatron certificates, which are used for
+      external publishing. A number of well-known directories are searched by
+      default. This option is only necessary if your certificates are in an
+      unusual location.); default: "";
     --meter_ttl (Meter TTL: expire meters after this period of inactivity.);
       default: 15m;
     --no_common_tags (No common tags will be provided for metrics. Since no

admin/admin_server.cc

@@ -49,18 +49,21 @@ void GET_config(HTTPServerRequest& req, HTTPServerResponse& res,
   Object::Ptr obj = new Object(true);
 
   Object::Ptr common_tags = new Object(true);
-  for (auto& pair : config.common_tags)
+  for (auto& pair : config.common_tags) {
     common_tags->set(pair.first, pair.second);
-  obj->set("common_tags", common_tags);
+  }
 
-  obj->set("read_timeout", ToDoubleMilliseconds(config.read_timeout));
-  obj->set("connect_timeout", ToDoubleMilliseconds(config.connect_timeout));
+  obj->set("age_gauge_limit", config.age_gauge_limit);
   obj->set("batch_size", config.batch_size);
+  obj->set("common_tags", common_tags);
+  obj->set("connect_timeout", ToDoubleMilliseconds(config.connect_timeout));
+  obj->set("expiration_frequency", ToDoubleMilliseconds(config.expiration_frequency));
+  obj->set("external_enabled", config.external_enabled);
   obj->set("frequency", ToDoubleMilliseconds(config.frequency));
-  obj->set("expiration_frequency",
-           ToDoubleMilliseconds(config.expiration_frequency));
+  obj->set("metatron_dir", config.metatron_dir);
   obj->set("meter_ttl", ToDoubleMilliseconds(config.meter_ttl));
-  obj->set("age_gauge_limit", config.age_gauge_limit);
+  obj->set("read_timeout", ToDoubleMilliseconds(config.read_timeout));
+  obj->set("status_metrics_enabled", config.status_metrics_enabled);
   obj->set("uri", config.uri);
 
   res.setStatus(HTTPResponse::HTTP_OK);

admin/admin_server_test.cc

@@ -161,17 +161,19 @@ TEST_F(AdminServerTest, GET_config) {
   Var result {parser.parse(rr)};
   Object::Ptr object = result.extract<Object::Ptr>();
 
-  int count = 0;
   std::vector<std::string> expected_keys {"age_gauge_limit", "batch_size", "common_tags",
-                                         "connect_timeout", "expiration_frequency", "frequency",
-                                         "meter_ttl", "read_timeout", "uri"};
+                                         "connect_timeout", "expiration_frequency",
+                                         "external_enabled", "frequency", "metatron_dir",
+                                         "meter_ttl", "read_timeout", "status_metrics_enabled",
+                                         "uri"};
+  std::vector<std::string> found_keys;
+
   for (auto &it : *object) {
-    if (contains(expected_keys, it.first)) {
-      count += 1;
-    }
+    found_keys.emplace_back(it.first);
   }
+  std::sort(found_keys.begin(), found_keys.end());
 
-  EXPECT_EQ(count, expected_keys.size());
+  EXPECT_EQ(found_keys, expected_keys);
 }
 
 TEST_F(AdminServerTest, GET_config_common_tags) {

bin/spectatord_main.cc

@@ -43,14 +43,19 @@ auto AbslParseFlag(absl::string_view text, PortNumber* p, std::string* error)
   return true;
 }
 
-ABSL_FLAG(PortNumber, port, PortNumber(1234),
-          "Port number for the UDP socket.");
-ABSL_FLAG(bool, enable_statsd, false,
-          "Enable statsd support.");
-ABSL_FLAG(PortNumber, statsd_port, PortNumber(8125),
-          "Port number for the statsd socket.");
 ABSL_FLAG(PortNumber, admin_port, PortNumber(1234),
           "Port number for the admin server.");
+ABSL_FLAG(size_t, age_gauge_limit, 1000,
+          "The maximum number of age gauges that may be reported by this process.");
+ABSL_FLAG(std::string, common_tags, "",
+          "Common tags: nf.app=app,nf.cluster=cluster. Override the default common "
+          "tags. If empty, then spectatord will use the default set. "
+          "This flag should only be used by experts who understand the risks.");
+ABSL_FLAG(bool, debug, false,
+          "Debug spectatord. All values will be sent to a dev aggregator and "
+          "dropped.");
+ABSL_FLAG(bool, enable_external, false,
+          "Enable external publishing.");
 #ifdef __linux__
 ABSL_FLAG(bool, enable_socket, true,
           "Enable UNIX domain socket support. Default is true on Linux and false "
@@ -60,30 +65,31 @@ ABSL_FLAG(bool, enable_socket, false,
           "Enable UNIX domain socket support. Default is true on Linux and false "
           "on MacOS and Windows.");
 #endif
-ABSL_FLAG(std::string, socket_path, "/run/spectatord/spectatord.unix",
-          "Path to the UNIX domain socket.");
-ABSL_FLAG(std::string, uri, "",
-          "Optional override URI for the aggregator.");
+ABSL_FLAG(bool, enable_statsd, false,
+          "Enable statsd support.");
+ABSL_FLAG(std::string, metatron_dir, "",
+          "Path to the Metatron certificates, which are used for external publishing. A number "
+          "of well-known directories are searched by default. This option is only necessary "
+          "if your certificates are in an unusual location.");
 ABSL_FLAG(absl::Duration, meter_ttl, absl::Minutes(15),
           "Meter TTL: expire meters after this period of inactivity.");
-ABSL_FLAG(size_t, age_gauge_limit, 1000,
-          "The maximum number of age gauges that may be reported by this process.");
-ABSL_FLAG(std::string, common_tags, "",
-          "Common tags: nf.app=app,nf.cluster=cluster. Override the default common "
-          "tags. If empty, then spectatord will use the default set. "
-          "This flag should only be used by experts who understand the risks.");
 ABSL_FLAG(bool, no_common_tags, false,
           "No common tags will be provided for metrics. Since no common tags are available, no "
           "internal status metrics will be recorded. Only use this feature for special cases "
           "where it is absolutely necessary to override common tags such as nf.app, and only "
           "use it with a secondary spectatord process.");
+ABSL_FLAG(PortNumber, port, PortNumber(1234),
+          "Port number for the UDP socket.");
+ABSL_FLAG(std::string, socket_path, "/run/spectatord/spectatord.unix",
+          "Path to the UNIX domain socket.");
+ABSL_FLAG(PortNumber, statsd_port, PortNumber(8125),
+          "Port number for the statsd socket.");
+ABSL_FLAG(std::string, uri, "",
+          "Optional override URI for the aggregator.");
 ABSL_FLAG(bool, verbose, false,
           "Use verbose logging.");
 ABSL_FLAG(bool, verbose_http, false,
           "Output debug info for HTTP requests.");
-ABSL_FLAG(bool, debug, false,
-          "Debug spectatord. All values will be sent to a dev aggregator and "
-          "dropped.");
 
 auto main(int argc, char** argv) -> int {
   auto logger = Logger();
@@ -93,18 +99,20 @@ auto main(int argc, char** argv) -> int {
                 signals.end());
   backward::SignalHandling sh{signals};
 
-  absl::SetProgramUsageMessage(
-      "A daemon that listens for metrics and reports them to Atlas.");
+  absl::SetProgramUsageMessage("A daemon that listens for metrics and reports them to Atlas.");
   absl::ParseCommandLine(argc, argv);
 
   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";
+    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)) {
+    cfg->external_enabled = true;
+    cfg->metatron_dir = absl::GetFlag(FLAGS_metatron_dir);
+    cfg->uri = cfg->external_uri;
   }
 
   if (absl::GetFlag(FLAGS_verbose_http)) {

build.sh

@@ -12,7 +12,14 @@ NC="\033[0m"
 if [[ "$1" == "clean" ]]; then
   echo -e "${BLUE}==== clean ====${NC}"
   rm -rf $BUILD_DIR
-  rm -f nflx_spectator_cppconf-*.zip
+  # intermediate download files
+  rm -f netflix_spectator_cppconf-*.zip
+  rm -f spectatord_metatron-*.zip
+  # extracted and generated files
+  rm -f metatron/auth_context.pb.cc
+  rm -f metatron/auth_context.pb.h
+  rm -f metatron/auth_context.proto
+  rm -f metatron/metatron_config.cc
   rm -rf ska
   rm -f spectator/*.inc
   rm -f spectator/netflix_config.cc

conanfile.py

@@ -17,6 +17,7 @@ class SpectatorDConan(ConanFile):
         "libcurl/8.4.0",
         "openssl/3.2.0",
         "poco/1.12.5p2",
+        "protobuf/3.21.12",
         "rapidjson/cci.20230929",
         "spdlog/1.12.0",
         "tsl-hopscotch-map/2.3.1",
@@ -28,6 +29,7 @@ class SpectatorDConan(ConanFile):
 
     def configure(self):
         self.options["libcurl"].with_c_ares = True
+        self.options["libcurl"].with_ssl = "openssl"
         self.options["poco"].enable_data_mysql = False
         self.options["poco"].enable_data_postgresql = False
         self.options["poco"].enable_data_sqlite = False
@@ -55,14 +57,30 @@ def get_netflix_spectator_cppconf():
             return
         dir_name = "netflix_spectator_cppconf"
         commit = "d44c6513f52fba019181e8c59c4c306bd6451b8d"
-        zip_name = f"nflx_spectator_cfg-{commit}.zip"
+        zip_name = f"netflix_spectator_cppconf-{commit}.zip"
         download(f"https://stash.corp.netflix.com/rest/api/latest/projects/CLDMTA/repos/netflix-spectator-cppconf/archive?at={commit}&format=zip", zip_name)
         check_sha256(zip_name, "87cafb9306c2cd96477aea2d26ef311ff0b4342a3fa57fd29432411ce355cf6a")
         unzip(zip_name, destination=dir_name)
         shutil.move(f"{dir_name}/netflix_config.cc", "spectator")
         os.unlink(zip_name)
         shutil.rmtree(dir_name)
 
+    @staticmethod
+    def get_spectatord_metatron():
+        if os.environ.get("NFLX_INTERNAL") != "ON":
+            return
+        dir_name = "spectatord_metatron"
+        commit = "07f0cbcf2d606561d636a1e22931aa8d23bcb7a3"
+        zip_name = f"spectatord_metatron-{commit}.zip"
+        download(f"https://stash.corp.netflix.com/rest/api/latest/projects/CLDMTA/repos/spectatord-metatron/archive?at={commit}&format=zip", zip_name)
+        check_sha256(zip_name, "a367d20d62d1ec57622fa325268e7be67b99e58b36ea22dd2e71eba2af853a6c")
+        unzip(zip_name, destination=dir_name)
+        shutil.move(f"{dir_name}/metatron/auth_context.proto", "metatron")
+        shutil.move(f"{dir_name}/metatron/metatron_config.cc", "metatron")
+        os.unlink(zip_name)
+        shutil.rmtree(dir_name)
+
     def source(self):
         self.get_flat_hash_map()
         self.get_netflix_spectator_cppconf()
+        self.get_spectatord_metatron()

metatron/CMakeLists.txt

@@ -0,0 +1,26 @@
+#-- metatron verifier library
+if(NFLX_INTERNAL)
+    add_library(metatron
+        "auth_context.pb.cc"
+        "auth_context.pb.h"
+        "metatron_config.cc"
+        "metatron_config.h"
+    )
+else()
+    add_library(metatron
+        "metatron_config.h"
+        "metatron_sample.cc"
+    )
+endif()
+
+target_include_directories(metatron PRIVATE
+    ${CMAKE_SOURCE_DIR}
+)
+target_link_libraries(metatron util ${CONAN_LIBS})
+
+#-- protobuf generated files must exist in both the SOURCE_DIR and the BINARY_DIR
+add_custom_command(
+    OUTPUT auth_context.pb.cc auth_context.pb.h
+    COMMAND ${CONAN_PROTOBUF_ROOT}/bin/protoc --proto_path=${CMAKE_CURRENT_SOURCE_DIR} --cpp_out=${CMAKE_CURRENT_SOURCE_DIR} auth_context.proto
+    COMMAND ${CONAN_PROTOBUF_ROOT}/bin/protoc --proto_path=${CMAKE_CURRENT_SOURCE_DIR} --cpp_out=${CMAKE_CURRENT_BINARY_DIR} auth_context.proto
+)

metatron/metatron_config.h

@@ -0,0 +1,22 @@
+#pragma once
+
+#include <string>
+#include <curl/curl.h>
+#include <openssl/ssl.h>
+
+namespace metatron {
+
+struct CertInfo {
+  std::string ssl_cert;
+  std::string ssl_key;
+  std::string ca_info;
+  std::string app_name;
+};
+
+CertInfo find_certificate(bool external_enabled, const std::string& metatron_dir);
+
+int metatron_verify_callback(int preverify_ok, X509_STORE_CTX *x509_ctx);
+
+CURLcode sslctx_metatron_verify(CURL *curl, void *ssl_ctx, void *parm);
+
+}  // namespace metatron

metatron/metatron_sample.cc

@@ -0,0 +1,22 @@
+#include "metatron_config.h"
+#include "util/logger.h"
+
+namespace metatron {
+
+CertInfo find_certificate(bool /*unused*/, const std::string& /*unused*/) {
+  return CertInfo{"ssl_cert", "ssl_key", "ca_info", "app_name"};
+}
+
+int metatron_verify_callback(int /*unused*/, X509_STORE_CTX* /*unused*/) {
+  static auto logger = spectatord::Logger();
+  logger->warn("Always stop the verification process with a verification failed state");
+  return 0;
+}
+
+CURLcode sslctx_metatron_verify(CURL* /*unused*/, void* /*unused*/, void* /*unused*/) {
+  static auto logger = spectatord::Logger();
+  logger->warn("Always halt SSL processing with the sample config");
+  return CURLE_SSL_CONNECT_ERROR;
+}
+
+}  // namespace metatron

spectator/CMakeLists.txt

@@ -65,7 +65,7 @@ add_library(spectator OBJECT
     "version.h"
 )
 target_include_directories(spectator PRIVATE ${CMAKE_SOURCE_DIR})
-target_link_libraries(spectator util ${CONAN_LIBS})
+target_link_libraries(spectator metatron util ${CONAN_LIBS})
 
 #-- file generators, must exist where the outputs are referenced
 add_custom_command(

spectator/config.h

@@ -40,8 +40,11 @@ class Config {
   size_t age_gauge_limit{};
   std::string uri;
   std::string external_uri;
-  bool verbose_http = false;
+
+  std::string metatron_dir;
+  bool external_enabled = false;
   bool status_metrics_enabled = true;
+  bool verbose_http = false;
 
   // sub-classes can override this method implementing custom logic
   // that can disable publishing under certain conditions