Add ClusterInfo as a property in StreamInfo (envoyproxy#10432) (#188)

Add ClusterName as a property in StreamInfo. This is because for TCP, there is no RouteEntry and thus we are not able to get clustername for TCP in scenarios where upstream cluster is not healthy

Risk Level: Low
Testing: Unit Tests
Docs Changes:
Release Notes:

Signed-off-by: gargnupur <gargnupur@google.com>
Co-authored-by: Matt Klein <mattklein123@gmail.com>

Co-authored-by: Matt Klein <mattklein123@gmail.com>

include/envoy/stream_info/stream_info.h

@@ -25,6 +25,11 @@ namespace Router {
 class RouteEntry;
 } // namespace Router
 
+namespace Upstream {
+class ClusterInfo;
+using ClusterInfoConstSharedPtr = std::shared_ptr<const ClusterInfo>;
+} // namespace Upstream
+
 namespace StreamInfo {
 
 enum ResponseFlag {
@@ -520,6 +525,19 @@ class StreamInfo {
    * @return request headers.
    */
   virtual const Http::HeaderMap* getRequestHeaders() const PURE;
+
+  /**
+   * @param Upstream Connection's ClusterInfo.
+   */
+  virtual void
+  setUpstreamClusterInfo(const Upstream::ClusterInfoConstSharedPtr& upstream_cluster_info) PURE;
+
+  /**
+   * @return Upstream Connection's ClusterInfo.
+   * This returns an optional to differentiate between unset(absl::nullopt),
+   * no route or cluster does not exist(nullptr), and set to a valid cluster(not nullptr).
+   */
+  virtual absl::optional<Upstream::ClusterInfoConstSharedPtr> upstreamClusterInfo() const PURE;
 };
 
 } // namespace StreamInfo

source/common/http/conn_manager_impl.cc

@@ -1365,6 +1365,7 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
     cached_cluster_info_ = (nullptr == local_cluster) ? nullptr : local_cluster->info();
   }
 
+  stream_info_.setUpstreamClusterInfo(cached_cluster_info_.value());
   refreshCachedTracingCustomTags();
 }
 

source/common/stream_info/stream_info_impl.h

@@ -251,6 +251,15 @@ struct StreamInfoImpl : public StreamInfo {
        << DUMP_MEMBER(health_check_request_) << DUMP_MEMBER(route_name_) << "\n";
   }
 
+  void setUpstreamClusterInfo(
+      const Upstream::ClusterInfoConstSharedPtr& upstream_cluster_info) override {
+    upstream_cluster_info_ = upstream_cluster_info;
+  }
+
+  absl::optional<Upstream::ClusterInfoConstSharedPtr> upstreamClusterInfo() const override {
+    return upstream_cluster_info_;
+  }
+
   TimeSource& time_source_;
   const SystemTime start_time_;
   const MonotonicTime start_time_monotonic_;
@@ -285,6 +294,7 @@ struct StreamInfoImpl : public StreamInfo {
   const Http::HeaderMap* request_headers_{};
   UpstreamTiming upstream_timing_;
   std::string upstream_transport_failure_reason_;
+  absl::optional<Upstream::ClusterInfoConstSharedPtr> upstream_cluster_info_;
 };
 
 } // namespace StreamInfo

source/common/tcp_proxy/tcp_proxy.cc

@@ -387,6 +387,7 @@ Network::FilterStatus Filter::initializeUpstreamConnection() {
   }
 
   Upstream::ClusterInfoConstSharedPtr cluster = thread_local_cluster->info();
+  getStreamInfo().setUpstreamClusterInfo(cluster);
 
   // Check this here because the TCP conn pool will queue our request waiting for a connection that
   // will never be released.

test/common/stream_info/stream_info_impl_test.cc

@@ -11,6 +11,7 @@
 
 #include "test/common/stream_info/test_int_accessor.h"
 #include "test/mocks/router/mocks.h"
+#include "test/mocks/upstream/cluster_info.h"
 #include "test/mocks/upstream/mocks.h"
 
 #include "gmock/gmock.h"
@@ -175,6 +176,12 @@ TEST_F(StreamInfoImplTest, MiscSettersAndGetters) {
     absl::string_view sni_name = "stubserver.org";
     stream_info.setRequestedServerName(sni_name);
     EXPECT_EQ(std::string(sni_name), stream_info.requestedServerName());
+
+    EXPECT_EQ(absl::nullopt, stream_info.upstreamClusterInfo());
+    Upstream::ClusterInfoConstSharedPtr cluster_info(new NiceMock<Upstream::MockClusterInfo>());
+    stream_info.setUpstreamClusterInfo(cluster_info);
+    EXPECT_NE(absl::nullopt, stream_info.upstreamClusterInfo());
+    EXPECT_EQ("fake_cluster", stream_info.upstreamClusterInfo().value()->name());
   }
 }
 

test/common/stream_info/test_util.h

@@ -207,6 +207,14 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
 
   Event::TimeSystem& timeSystem() { return test_time_.timeSystem(); }
 
+  void setUpstreamClusterInfo(
+      const Upstream::ClusterInfoConstSharedPtr& upstream_cluster_info) override {
+    upstream_cluster_info_ = upstream_cluster_info;
+  }
+  absl::optional<Upstream::ClusterInfoConstSharedPtr> upstreamClusterInfo() const override {
+    return upstream_cluster_info_;
+  }
+
   SystemTime start_time_;
   MonotonicTime start_time_monotonic_;
 
@@ -243,6 +251,7 @@ class TestStreamInfo : public StreamInfo::StreamInfo {
   std::string upstream_transport_failure_reason_;
   const Http::HeaderMap* request_headers_{};
   Envoy::Event::SimulatedTimeSystem test_time_;
+  absl::optional<Upstream::ClusterInfoConstSharedPtr> upstream_cluster_info_{};
 };
 
 } // namespace Envoy

test/common/tcp_proxy/tcp_proxy_test.cc

@@ -1892,6 +1892,45 @@ TEST_F(TcpProxyRoutingTest, DEPRECATED_FEATURE_TEST(ApplicationProtocols)) {
   filter_->onNewConnection();
 }
 
+class TcpProxyNonDeprecatedConfigRoutingTest : public TcpProxyRoutingTest {
+public:
+  TcpProxyNonDeprecatedConfigRoutingTest() = default;
+
+  void setup() {
+    const std::string yaml = R"EOF(
+    stat_prefix: name
+    cluster: fake_cluster
+    )EOF";
+
+    config_.reset(new Config(constructConfigFromYaml(yaml, factory_context_)));
+  }
+};
+
+TEST_F(TcpProxyNonDeprecatedConfigRoutingTest, ClusterNameSet) {
+  setup();
+
+  initializeFilter();
+
+  // Port 9999 is within the specified destination port range.
+  connection_.local_address_ = std::make_shared<Network::Address::Ipv4Instance>("1.2.3.4", 9999);
+
+  // Expect filter to try to open a connection to specified cluster.
+  EXPECT_CALL(factory_context_.cluster_manager_, tcpConnPoolForCluster("fake_cluster", _, _))
+      .WillOnce(Return(nullptr));
+  absl::optional<Upstream::ClusterInfoConstSharedPtr> cluster_info;
+  EXPECT_CALL(connection_.stream_info_, setUpstreamClusterInfo(_))
+      .WillOnce(
+          Invoke([&cluster_info](const Upstream::ClusterInfoConstSharedPtr& upstream_cluster_info) {
+            cluster_info = upstream_cluster_info;
+          }));
+  EXPECT_CALL(connection_.stream_info_, upstreamClusterInfo())
+      .WillOnce(ReturnPointee(&cluster_info));
+
+  filter_->onNewConnection();
+
+  EXPECT_EQ(connection_.stream_info_.upstreamClusterInfo().value()->name(), "fake_cluster");
+}
+
 class TcpProxyHashingTest : public testing::Test {
 public:
   TcpProxyHashingTest() = default;

test/mocks/stream_info/mocks.h

@@ -89,6 +89,9 @@ class MockStreamInfo : public StreamInfo {
   MOCK_METHOD(const std::string&, upstreamTransportFailureReason, (), (const));
   MOCK_METHOD(void, setRequestHeaders, (const Http::HeaderMap&));
   MOCK_METHOD(const Http::HeaderMap*, getRequestHeaders, (), (const));
+  MOCK_METHOD(void, setUpstreamClusterInfo, (const Upstream::ClusterInfoConstSharedPtr&));
+  MOCK_METHOD(absl::optional<Upstream::ClusterInfoConstSharedPtr>, upstreamClusterInfo, (),
+              (const));
 
   std::shared_ptr<testing::NiceMock<Upstream::MockHostDescription>> host_{
       new testing::NiceMock<Upstream::MockHostDescription>()};