Support ADS mode in cilium proxy

cilium/BUILD

@@ -111,6 +111,7 @@ envoy_cc_library(
     ],
     repository = "@envoy",
     deps = [
+        "@com_google_absl//absl/strings",
         "@envoy//envoy/network:connection_interface",
         "@envoy//source/common/config:type_to_endpoint_lib",
         "@envoy//source/extensions/config_subscription/grpc:grpc_subscription_lib",
@@ -336,13 +337,16 @@ envoy_cc_library(
         "//cilium:socket_option_lib",
         "//cilium/api:bpf_metadata_cc_proto",
         "//cilium/api:nphds_cc_proto",
+        "@com_google_absl//absl/strings",
         "@envoy//envoy/buffer:buffer_interface",
         "@envoy//envoy/config:subscription_interface",
         "@envoy//envoy/network:connection_interface",
         "@envoy//envoy/network:filter_interface",
         "@envoy//envoy/registry",
         "@envoy//envoy/server:filter_config_interface",
         "@envoy//envoy/singleton:manager_interface",
+        "@envoy//envoy/stats:stats_interface",
+        "@envoy//envoy/stats:stats_macros",
         "@envoy//source/common/common:assert_lib",
         "@envoy//source/common/common:logger_lib",
         "@envoy//source/common/network:address_lib",

cilium/grpc_subscription.cc

@@ -121,7 +121,8 @@ const Protobuf::MethodDescriptor& sotwGrpcMethod(absl::string_view type_url) {
 }
 
 std::unique_ptr<Config::GrpcSubscriptionImpl>
-subscribe(const std::string& type_url, const envoy::config::core::v3::ConfigSource& npds_config,
+subscribe(const absl::string_view type_url,
+          const envoy::config::core::v3::ConfigSource& npds_config,
           const LocalInfo::LocalInfo& local_info, Upstream::ClusterManager& cm,
           Event::Dispatcher& dispatcher, Random::RandomGenerator& random, Stats::Scope& scope,
           Config::SubscriptionCallbacks& callbacks,

cilium/grpc_subscription.h

@@ -2,7 +2,6 @@
 
 #include <chrono>
 #include <memory>
-#include <string>
 
 #include "envoy/common/random_generator.h"
 #include "envoy/config/core/v3/config_source.pb.h"
@@ -16,6 +15,8 @@
 #include "source/extensions/config_subscription/grpc/grpc_mux_impl.h"
 #include "source/extensions/config_subscription/grpc/grpc_subscription_impl.h"
 
+#include "absl/strings/string_view.h"
+
 namespace Envoy {
 namespace Cilium {
 
@@ -44,7 +45,8 @@ class GrpcMuxImpl : public Config::GrpcMuxImpl {
 };
 
 std::unique_ptr<Config::GrpcSubscriptionImpl>
-subscribe(const std::string& type_url, const envoy::config::core::v3::ConfigSource& npds_config,
+subscribe(const absl::string_view type_url,
+          const envoy::config::core::v3::ConfigSource& npds_config,
           const LocalInfo::LocalInfo& local_info, Upstream::ClusterManager& cm,
           Event::Dispatcher& dispatcher, Random::RandomGenerator& random, Stats::Scope& scope,
           Config::SubscriptionCallbacks& callbacks,

cilium/host_map.cc

@@ -16,6 +16,8 @@
 #include "envoy/config/subscription.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/server/factory_context.h"
+#include "envoy/stats/scope.h"
+#include "envoy/stats/stats_macros.h"
 #include "envoy/thread_local/thread_local.h"
 #include "envoy/thread_local/thread_local_object.h"
 
@@ -24,13 +26,19 @@
 
 #include "absl/numeric/int128.h"
 #include "absl/status/status.h"
+#include "absl/strings/str_cat.h"
 #include "absl/strings/string_view.h"
 #include "cilium/api/nphds.pb.h"
 #include "cilium/grpc_subscription.h"
 
 namespace Envoy {
 namespace Cilium {
 
+namespace {
+
+static constexpr absl::string_view NetworkPolicyHostsTypeUrl =
+    "type.googleapis.com/cilium.NetworkPolicyHosts";
+
 template <typename T>
 unsigned int checkPrefix(T addr, bool have_prefix, unsigned int plen, absl::string_view host) {
   const unsigned int plen_max = sizeof(T) * 8;
@@ -47,6 +55,8 @@ unsigned int checkPrefix(T addr, bool have_prefix, unsigned int plen, absl::stri
   return plen;
 }
 
+} // namespace
+
 struct ThreadLocalHostMapInitializer : public PolicyHostMap::ThreadLocalHostMap {
 protected:
   friend class PolicyHostMap; // PolicyHostMap can insert();
@@ -154,7 +164,11 @@ struct ThreadLocalHostMapInitializer : public PolicyHostMap::ThreadLocalHostMap
 uint64_t PolicyHostMap::instance_id_ = 0;
 
 // This is used directly for testing with a file-based subscription
-PolicyHostMap::PolicyHostMap(ThreadLocal::SlotAllocator& tls) : tls_(tls.allocateSlot()) {
+PolicyHostMap::PolicyHostMap(ThreadLocal::SlotAllocator& tls, Stats::Scope& scope)
+    : tls_(tls.allocateSlot()),
+      name_(absl::StrCat("cilium.hostmap.", fmt::format("{}", instance_id_ + 1), ".")),
+      scope_(scope.createScope(name_)), stats_scope_(scope.createScope("cilium.hostmap.")),
+      stats_({CILIUM_POLICY_HOSTS_STATS(POOL_COUNTER(*stats_scope_))}) {
   instance_id_++;
   name_ = "cilium.hostmap." + fmt::format("{}", instance_id_) + ".";
   ENVOY_LOG(debug, "PolicyHostMap({}) created.", name_);
@@ -167,16 +181,36 @@ PolicyHostMap::PolicyHostMap(ThreadLocal::SlotAllocator& tls) : tls_(tls.allocat
 
 // This is used in production
 PolicyHostMap::PolicyHostMap(Server::Configuration::CommonFactoryContext& context)
-    : PolicyHostMap(context.threadLocal()) {
-  scope_ = context.serverScope().createScope(name_);
+    : tls_(context.threadLocal().allocateSlot()),
+      name_(absl::StrCat("cilium.hostmap.", fmt::format("{}", instance_id_ + 1), ".")),
+      scope_(context.serverScope().createScope(name_)),
+      stats_scope_(context.serverScope().createScope("cilium.hostmap.")),
+      stats_({CILIUM_POLICY_HOSTS_STATS(POOL_COUNTER(*stats_scope_))}) {
+  instance_id_++;
+  ENVOY_LOG(debug, "PolicyHostMap({}) created.", name_);
+
+  auto empty_map = std::make_shared<ThreadLocalHostMapInitializer>();
+  tls_->set([empty_map](Event::Dispatcher&) -> ThreadLocal::ThreadLocalObjectSharedPtr {
+    return empty_map;
+  });
 }
 
 void PolicyHostMap::startSubscription(Server::Configuration::CommonFactoryContext& context,
                                       const envoy::config::core::v3::ConfigSource& npds_config) {
-  subscription_ = subscribe("type.googleapis.com/cilium.NetworkPolicyHosts", npds_config,
-                            context.localInfo(), context.clusterManager(),
-                            context.mainThreadDispatcher(), context.api().randomGenerator(),
-                            *scope_, *this, std::make_shared<Cilium::PolicyHostDecoder>());
+  if (npds_config.config_source_specifier_case() == envoy::config::core::v3::ConfigSource::kAds) {
+    auto ads_mux = context.xdsManager().adsMux();
+    subscription_ = THROW_OR_RETURN_VALUE(
+        context.clusterManager().subscriptionFactory().subscriptionOverAdsGrpcMux(
+            ads_mux, npds_config, NetworkPolicyHostsTypeUrl, *scope_, *this,
+            std::make_shared<Cilium::PolicyHostDecoder>(), {}),
+        Config::SubscriptionPtr);
+  } else {
+    subscription_ = subscribe(NetworkPolicyHostsTypeUrl, npds_config, context.localInfo(),
+                              context.clusterManager(), context.mainThreadDispatcher(),
+                              context.api().randomGenerator(), *scope_, *this,
+                              std::make_shared<Cilium::PolicyHostDecoder>());
+  }
+
   subscription_->start({});
 }
 
@@ -211,6 +245,7 @@ PolicyHostMap::onConfigUpdate(const std::vector<Envoy::Config::DecodedResourceRe
     return newmap;
   });
   logmaps("onConfigUpdate");
+  stats_.update_success_.inc();
   return absl::OkStatus();
 }
 

cilium/host_map.h

@@ -21,6 +21,7 @@
 #include "envoy/server/factory_context.h"
 #include "envoy/singleton/instance.h"
 #include "envoy/stats/scope.h"
+#include "envoy/stats/stats_macros.h"
 #include "envoy/thread_local/thread_local.h"
 #include "envoy/thread_local/thread_local_object.h"
 
@@ -91,13 +92,25 @@ class PolicyHostDecoder : public Envoy::Config::OpaqueResourceDecoder {
   ProtobufMessage::ValidationVisitor& validation_visitor_;
 };
 
+// clang-format off
+#define CILIUM_POLICY_HOSTS_STATS(COUNTER)	\
+  COUNTER(update_success)
+// clang-format on
+
+/**
+ * Struct definition for all policy stats. @see stats_macros.h
+ */
+struct PolicyHostsStats {
+  CILIUM_POLICY_HOSTS_STATS(GENERATE_COUNTER_STRUCT)
+};
+
 class PolicyHostMap : public Singleton::Instance,
                       public Config::SubscriptionCallbacks,
                       public std::enable_shared_from_this<PolicyHostMap>,
                       public Logger::Loggable<Logger::Id::config> {
 public:
   PolicyHostMap(Server::Configuration::CommonFactoryContext& context);
-  PolicyHostMap(ThreadLocal::SlotAllocator& tls);
+  PolicyHostMap(ThreadLocal::SlotAllocator& tls, Stats::Scope& scope);
   ~PolicyHostMap() override {
     ENVOY_LOG(debug, "Cilium PolicyHostMap({}): PolicyHostMap is deleted NOW!", name_);
   }
@@ -228,10 +241,12 @@ class PolicyHostMap : public Singleton::Instance,
 
 private:
   ThreadLocal::SlotPtr tls_;
+  std::string name_;
   Stats::ScopeSharedPtr scope_;
+  Stats::ScopeSharedPtr stats_scope_;
   std::unique_ptr<Envoy::Config::Subscription> subscription_;
   static uint64_t instance_id_;
-  std::string name_;
+  PolicyHostsStats stats_;
 };
 
 } // namespace Cilium

cilium/network_policy.cc

@@ -60,6 +60,13 @@
 #include "cilium/ipcache.h"
 #include "cilium/secret_watcher.h"
 
+namespace {
+
+static constexpr absl::string_view NetworkPolicyTypeUrl =
+    "type.googleapis.com/cilium.NetworkPolicy";
+
+} // namespace
+
 namespace fmt {
 
 template <> struct formatter<Envoy::Cilium::RuleVerdict> {
@@ -1838,7 +1845,7 @@ NetworkPolicyMap::NetworkPolicyMap(Server::Configuration::FactoryContext& contex
   }
 
   if (subscribe) {
-    getImpl().startSubscription();
+    getImpl().startSubscription(npds_config);
   }
 }
 
@@ -1877,8 +1884,7 @@ NetworkPolicyMapImpl::NetworkPolicyMapImpl(Server::Configuration::FactoryContext
               context_, *npds_stats_scope_,
               context_.messageValidationContext().dynamicValidationVisitor())),
       npds_config_(npds_config),
-      stats_{ALL_CILIUM_POLICY_STATS(POOL_COUNTER(*policy_stats_scope_),
-                                     POOL_HISTOGRAM(*policy_stats_scope_))} {
+      stats_{ALL_CILIUM_POLICY_STATS(POOL_COUNTER(*policy_stats_scope_))} {
   // Use listener init manager for subscription initialization
   context.initManager().add(init_target_);
 
@@ -1894,11 +1900,23 @@ NetworkPolicyMapImpl::~NetworkPolicyMapImpl() {
   delete load();
 }
 
-void NetworkPolicyMapImpl::startSubscription() {
-  subscription_ = subscribe("type.googleapis.com/cilium.NetworkPolicy", npds_config_,
-                            context_.localInfo(), context_.clusterManager(),
-                            context_.mainThreadDispatcher(), context_.api().randomGenerator(),
-                            *npds_stats_scope_, *this, std::make_shared<NetworkPolicyDecoder>());
+void NetworkPolicyMapImpl::startSubscription(
+    const envoy::config::core::v3::ConfigSource& npds_config) {
+  if (npds_config.config_source_specifier_case() == envoy::config::core::v3::ConfigSource::kAds) {
+    auto ads_mux = context_.xdsManager().adsMux();
+    subscription_ = THROW_OR_RETURN_VALUE(
+        context_.clusterManager().subscriptionFactory().subscriptionOverAdsGrpcMux(
+            ads_mux, npds_config, NetworkPolicyTypeUrl, *npds_stats_scope_, *this,
+            std::make_shared<NetworkPolicyDecoder>(), {}),
+        Config::SubscriptionPtr);
+  } else {
+    subscription_ = subscribe(NetworkPolicyTypeUrl, npds_config, context_.localInfo(),
+                              context_.clusterManager(), context_.mainThreadDispatcher(),
+                              context_.api().randomGenerator(), *npds_stats_scope_, *this,
+                              std::make_shared<NetworkPolicyDecoder>());
+  }
+
+  subscription_->start({});
 }
 
 void NetworkPolicyMapImpl::tlsWrapperMissingPolicyInc() const {
@@ -2027,7 +2045,7 @@ absl::Status NetworkPolicyMapImpl::onConfigUpdate(
     // Clean-up in the main thread after all threads have scheduled
     delete old_map;
   });
-
+  stats_.update_success_.inc();
   return absl::OkStatus();
 }
 

cilium/network_policy.h

@@ -215,17 +215,18 @@ class NetworkPolicyDecoder : public Envoy::Config::OpaqueResourceDecoder {
  * All Cilium L7 filter stats. @see stats_macros.h
  */
 // clang-format off
-#define ALL_CILIUM_POLICY_STATS(COUNTER, HISTOGRAM)	\
+#define ALL_CILIUM_POLICY_STATS(COUNTER)	\
   COUNTER(updates_total)				\
   COUNTER(updates_rejected)				\
-  COUNTER(tls_wrapper_missing_policy)
+  COUNTER(tls_wrapper_missing_policy) \
+  COUNTER(update_success)
 // clang-format on
 
 /**
  * Struct definition for all policy stats. @see stats_macros.h
  */
 struct PolicyStats {
-  ALL_CILIUM_POLICY_STATS(GENERATE_COUNTER_STRUCT, GENERATE_HISTOGRAM_STRUCT)
+  ALL_CILIUM_POLICY_STATS(GENERATE_COUNTER_STRUCT)
 };
 
 using RawPolicyMap = absl::flat_hash_map<std::string, std::shared_ptr<const PolicyInstanceImpl>>;
@@ -237,15 +238,15 @@ class NetworkPolicyMapImpl : public Envoy::Config::SubscriptionCallbacks,
                        const envoy::config::core::v3::ConfigSource& npds_config);
   ~NetworkPolicyMapImpl() override;
 
-  void startSubscription();
-
-  const envoy::config::core::v3::ConfigSource& getConfigSource() const { return npds_config_; }
+  void startSubscription(const envoy::config::core::v3::ConfigSource& npds_config);
 
   // This is used for testing with a file-based subscription
   void startSubscription(std::unique_ptr<Envoy::Config::Subscription>&& subscription) {
     subscription_ = std::move(subscription);
   }
 
+  const envoy::config::core::v3::ConfigSource& getConfigSource() const { return npds_config_; }
+
   // run the given function after all the threads have scheduled
   void runAfterAllThreads(std::function<void()>) const;
 

tests/BUILD

@@ -328,6 +328,26 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "bpf_metadata_integration_test",
+    srcs = ["bpf_metadata_integration_test.cc"],
+    repository = "@envoy",
+    deps = [
+        "//cilium:bpf_metadata_lib",
+        "//cilium/api:bpf_metadata_cc_proto",
+        "//cilium/api:npds_cc_proto",
+        "//cilium/api:nphds_cc_proto",
+        "@envoy//envoy/grpc:status",
+        "@envoy//source/extensions/clusters/original_dst:original_dst_cluster_lib",
+        "@envoy//source/extensions/clusters/static:static_cluster_lib",
+        "@envoy//source/extensions/filters/network/tcp_proxy:config",
+        "@envoy//test/common/grpc:grpc_client_integration_lib",
+        "@envoy//test/integration:integration_lib",
+        "@envoy//test/test_common:resources_lib",
+        "@envoy//test/test_common:utility_lib",
+    ],
+)
+
 envoy_cc_test(
     name = "health_check_sink_test",
     srcs = [
@@ -339,9 +359,14 @@ envoy_cc_test(
     deps = [
         ":uds_server_lib",
         "//cilium:health_check_sink_lib",
+        "@envoy//envoy/grpc:status",
+        "@envoy//envoy/http:codec_interface",
+        "@envoy//envoy/network:address_interface",
+        "@envoy//source/common/common:assert_lib",
         "@envoy//test/mocks/access_log:access_log_mocks",
         "@envoy//test/mocks/server:health_checker_factory_context_mocks",
         "@envoy//test/test_common:environment_lib",
         "@envoy//test/test_common:utility_lib",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
     ],
 )