downstream reverse tunnel: added handshake metrics (#42593)

<!--
!!!ATTENTION!!!

If you are fixing *any* crash or *any* potential security issue, *do
not*
open a pull request in this repo. Please report the issue via emailing
envoy-security@googlegroups.com where the issue will be triaged
appropriately.
Thank you in advance for helping to keep Envoy secure.

!!!ATTENTION!!!

For an explanation of how to fill out the fields, please see the
relevant section
in
[PULL_REQUESTS.md](https://github.com/envoyproxy/envoy/blob/main/PULL_REQUESTS.md)

!!!ATTENTION!!!

Please check the [use of generative AI
policy](https://github.com/envoyproxy/envoy/blob/main/CONTRIBUTING.md?plain=1#L41).

You may use generative AI only if you fully understand the code. You
need to disclose
this usage in the PR description to ensure transparency.
-->

Commit Message: added reverse tunnel downstream handshake metrics
Additional Description: 
1. reverse tunnel downstream handshake metrics only on worker level, not
aggregated across worker threads.
2. stats are only captured if enable_detailed_stats flag is enabled on
downstream reverse tunnel extension.

Risk Level: Low

Testing: 
1. When handshake fails
```
# TYPE envoy_downstream_reverse_connection_handshake counter
envoy_downstream_reverse_connection_handshake{worker="worker_0", cluster="upstream-cluster", result="failed", failure_reason="http.403"} 3 
envoy_downstream_reverse_connection_handshake{worker="worker_1", cluster="upstream-cluster", result="failed"，failure_reason="http.403"} 3
envoy_downstream_reverse_connection_handshake{worker="worker_2", cluster="upstream-cluster", result="failed", failure_reason="http. 403"} 3
envoy_downstream_reverse_connection_handshake{worker="worker_3",cluster="upstream-cluster", result="failed" failure_reason="http.403"} 3
```
2. When handshake passes
```
# TYPE envoy_downstream_reverse_connection_handshake counter
envoy_downstream_reverse_connection_handshake{worker="worker_0",cluster="upstream-cluster" ,result="success"} 1
envoy_downstream_reverse_connection_handshake{worker="worker_1",cluster="upstream-cluster" ,result="success"} 1
envoy_downstream_reverse_connection_handshake{worker="worker_2" ,cluster="upstream-cluster", result="success"} 1
envoy_downstream_reverse_connection_handshake{worker="worker_3",cluster="upstream-cluster" ,result="success"} 1
```

Docs Changes: None
Release Notes: None
Platform Specific Features:
[Optional Runtime guard:]
[Optional Fixes #Issue]
[Optional Fixes commit #PR or SHA]
[Optional Deprecated:]
[Optional [API
Considerations](https://github.com/envoyproxy/envoy/blob/main/api/review_checklist.md):]

---------

Signed-off-by: Krishna Sharma <krishnagpl2001@gmail.com>

Author: roll-no-21

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/rc_connection_wrapper.cc

@@ -11,6 +11,7 @@
 #include "source/common/network/connection_socket_impl.h"
 #include "source/extensions/bootstrap/reverse_tunnel/common/reverse_connection_utility.h"
 #include "source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_connection_io_handle.h"
+#include "source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_tunnel_initiator_extension.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -143,7 +144,7 @@ std::string RCConnectionWrapper::connect(const std::string& src_tenant_id,
   const Http::Status encode_status = request_encoder.encodeHeaders(*headers, true);
   if (!encode_status.ok()) {
     ENVOY_LOG(error, "RCConnectionWrapper: encodeHeaders failed: {}", encode_status.message());
-    onHandshakeFailure("HTTP handshake encode failed");
+    onHandshakeFailure(HandshakeFailureReason::encodeError());
   }
 
   return connection_->connectionInfoProvider().localAddress()->asString();
@@ -156,7 +157,7 @@ void RCConnectionWrapper::decodeHeaders(Http::ResponseHeaderMapPtr&& headers, bo
     onHandshakeSuccess();
   } else {
     ENVOY_LOG(error, "Received non-200 HTTP response: {}", status);
-    onHandshakeFailure(absl::StrCat("HTTP handshake failed with status ", status));
+    onHandshakeFailure(HandshakeFailureReason::httpStatusError(absl::StrCat(status)));
   }
 }
 
@@ -169,15 +170,36 @@ void RCConnectionWrapper::dispatchHttp1(Buffer::Instance& buffer) {
   }
 }
 
+ReverseTunnelInitiatorExtension* RCConnectionWrapper::getDownstreamExtension() const {
+  return parent_.getDownstreamExtension();
+}
+
 void RCConnectionWrapper::onHandshakeSuccess() {
   std::string message = "reverse connection accepted";
   ENVOY_LOG(debug, "handshake succeeded: {}", message);
+
+  // Track handshake success stats.
+  auto* extension = getDownstreamExtension();
+  if (extension) {
+    extension->incrementHandshakeStats(cluster_name_, true, "");
+  }
+
   parent_.onConnectionDone(message, this, false);
 }
 
-void RCConnectionWrapper::onHandshakeFailure(const std::string& message) {
-  ENVOY_LOG(debug, "handshake failed: {}", message);
-  parent_.onConnectionDone(message, this, false);
+void RCConnectionWrapper::onHandshakeFailure(const HandshakeFailureReason& reason) {
+  const std::string error_message = reason.getDetailedName();
+  const std::string stats_failure_reason = reason.getNameForStats();
+
+  ENVOY_LOG(trace, "handshake failed: {}", error_message);
+
+  // Track handshake failure stats.
+  auto* extension = getDownstreamExtension();
+  if (extension) {
+    extension->incrementHandshakeStats(cluster_name_, false, stats_failure_reason);
+  }
+
+  parent_.onConnectionDone(error_message, this, false);
 }
 
 void RCConnectionWrapper::shutdown() {

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/rc_connection_wrapper.h

@@ -15,13 +15,77 @@
 #include "source/common/http/response_decoder_impl_base.h"
 #include "source/common/network/filter_impl.h"
 
+#include "absl/strings/str_cat.h"
+#include "absl/strings/string_view.h"
+#include "absl/types/optional.h"
+
 namespace Envoy {
 namespace Extensions {
 namespace Bootstrap {
 namespace ReverseConnection {
 
-// Forward declaration.
+// Forward declarations.
 class ReverseConnectionIOHandle;
+class ReverseTunnelInitiatorExtension;
+
+/**
+ * Class representing handshake failure with type and context.
+ * Provides methods to generate detailed error messages and stat names.
+ */
+class HandshakeFailureReason {
+public:
+  enum class Type {
+    HttpStatusError, // HTTP response with non-200 status code
+    EncodeError,     // HTTP request encoding failed
+  };
+
+  /**
+   * Create a handshake failure reason for HTTP status errors.
+   * @param status_code the HTTP status code received
+   */
+  static HandshakeFailureReason httpStatusError(absl::string_view status_code) {
+    return {Type::HttpStatusError, status_code};
+  }
+
+  /**
+   * Create a handshake failure reason for encoding errors.
+   */
+  static HandshakeFailureReason encodeError() { return {Type::EncodeError, ""}; }
+
+  /**
+   * Get a detailed human-readable error message.
+   * @return detailed error message string
+   */
+  std::string getDetailedName() const {
+    switch (type_) {
+    case Type::HttpStatusError:
+      return absl::StrCat("HTTP handshake failed with status ", context_);
+    case Type::EncodeError:
+      return "HTTP handshake encode failed";
+    }
+    return "Unknown handshake failure";
+  }
+
+  /**
+   * Get the stat name suffix for this failure.
+   * @return stat name suffix (e.g., "http.401", "encode_error")
+   */
+  std::string getNameForStats() const {
+    switch (type_) {
+    case Type::HttpStatusError:
+      return absl::StrCat("http.", context_);
+    case Type::EncodeError:
+      return "encode_error";
+    }
+    return "unknown";
+  }
+
+private:
+  HandshakeFailureReason(Type type, absl::string_view context) : type_(type), context_(context) {}
+
+  Type type_;
+  std::string context_;
+};
 
 /**
  * Simple read filter for handling reverse connection handshake responses.
@@ -120,9 +184,9 @@ class RCConnectionWrapper : public Network::ConnectionCallbacks,
 
   /**
    * Handle handshake failure.
-   * @param message error message
+   * @param reason the failure reason with type and context
    */
-  void onHandshakeFailure(const std::string& message);
+  void onHandshakeFailure(const HandshakeFailureReason& reason);
 
   /**
    * Perform graceful shutdown of the connection.
@@ -151,6 +215,12 @@ class RCConnectionWrapper : public Network::ConnectionCallbacks,
   bool handshake_completed_{false};
   bool shutdown_called_{false};
 
+  /**
+   * Get the downstream extension for accessing stats.
+   * @return pointer to ReverseTunnelInitiatorExtension
+   */
+  ReverseTunnelInitiatorExtension* getDownstreamExtension() const;
+
 public:
   // Dispatch incoming bytes to HTTP/1 codec.
   void dispatchHttp1(Buffer::Instance& buffer);

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_tunnel_initiator_extension.cc

@@ -7,6 +7,7 @@
 
 #include "source/common/common/logger.h"
 #include "source/common/stats/symbol_table.h"
+#include "source/common/stats/utility.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -281,6 +282,75 @@ absl::flat_hash_map<std::string, uint64_t> ReverseTunnelInitiatorExtension::getP
   return stats_map;
 }
 
+void ReverseTunnelInitiatorExtension::incrementHandshakeStats(const std::string& cluster_id,
+                                                              bool success,
+                                                              const std::string& failure_reason) {
+  // Check if detailed stats are enabled via configuration flag.
+  if (!enable_detailed_stats_) {
+    return;
+  }
+
+  auto& stats_store = context_.scope();
+
+  // Get dispatcher name (worker name).
+  std::string dispatcher_name = "main_thread"; // Default for main thread
+  auto* local_registry = getLocalRegistry();
+  if (local_registry) {
+    // Dispatcher name is of the form "worker_x" where x is the worker index.
+    dispatcher_name = local_registry->dispatcher().name();
+  }
+
+  // Base stat name: <stat_prefix>.handshake
+  // Labels: worker=<worker_name>, cluster=<cluster_id>, result=<success|failed>,
+  //         failure_reason=<failure_reason> (only for failures)
+  std::string base_stat_name = fmt::format("{}.handshake", stat_prefix_);
+  Stats::StatNameManagedStorage stat_storage(base_stat_name, stats_store.symbolTable());
+
+  // Create storage for all tag keys and values - must be kept alive for the entire function.
+  Stats::StatNameManagedStorage worker_key_storage("worker", stats_store.symbolTable());
+  Stats::StatNameManagedStorage worker_value_storage(dispatcher_name, stats_store.symbolTable());
+  Stats::StatNameManagedStorage cluster_key_storage("cluster", stats_store.symbolTable());
+  Stats::StatNameManagedStorage cluster_value_storage(cluster_id, stats_store.symbolTable());
+  std::string result_value = success ? "success" : "failed";
+  Stats::StatNameManagedStorage result_key_storage("result", stats_store.symbolTable());
+  Stats::StatNameManagedStorage result_value_storage(result_value, stats_store.symbolTable());
+  Stats::StatNameManagedStorage failure_reason_key_storage("failure_reason",
+                                                           stats_store.symbolTable());
+  Stats::StatNameManagedStorage failure_reason_value_storage(failure_reason,
+                                                             stats_store.symbolTable());
+
+  // Now create tags vector using the stored StatNames.
+  Stats::StatNameTagVector tags;
+
+  // Add worker tag.
+  tags.push_back({worker_key_storage.statName(), worker_value_storage.statName()});
+
+  // Add cluster tag.
+  if (!cluster_id.empty()) {
+    tags.push_back({cluster_key_storage.statName(), cluster_value_storage.statName()});
+  }
+
+  // Add result tag.
+  tags.push_back({result_key_storage.statName(), result_value_storage.statName()});
+
+  // Add failure_reason tag for failures.
+  if (!success && !failure_reason.empty()) {
+    tags.push_back(
+        {failure_reason_key_storage.statName(), failure_reason_value_storage.statName()});
+  }
+
+  // Get or create the counter with tags and increment it.
+  // The third parameter takes the tags vector (StatNameTagVectorOptConstRef).
+  auto& handshake_counter =
+      Stats::Utility::counterFromStatNames(stats_store, {stat_storage.statName()}, tags);
+  handshake_counter.inc();
+
+  ENVOY_LOG(trace,
+            "reverse_tunnel: incremented handshake stat {} with tags worker={}, cluster={}, "
+            "result={}, failure_reason={}",
+            base_stat_name, dispatcher_name, cluster_id, result_value, failure_reason);
+}
+
 } // namespace ReverseConnection
 } // namespace Bootstrap
 } // namespace Extensions

source/extensions/bootstrap/reverse_tunnel/downstream_socket_interface/reverse_tunnel_initiator_extension.h

@@ -92,6 +92,16 @@ class ReverseTunnelInitiatorExtension : public Server::BootstrapExtension,
    */
   Stats::Scope& getStatsScope() const { return context_.scope(); }
 
+  /**
+   * Increment handshake stats for reverse tunnel connections (per-worker only).
+   * Only tracks stats if enable_detailed_stats flag is true.
+   * @param cluster_id the cluster identifier for the connection
+   * @param success true for successful handshake, false for failure
+   * @param failure_reason optional failure reason (e.g., "encode_error", "http.401", "http.500")
+   */
+  void incrementHandshakeStats(const std::string& cluster_id, bool success,
+                               const std::string& failure_reason = "");
+
   /**
    * Test-only method to set the thread local slot for testing purposes.
    * This allows tests to inject a custom thread local registry and is used