tap: fix upstream streamed transport socket taps (#13638)

Fixes #13608

Signed-off-by: Matt Klein <[email protected]>

api/envoy/config/common/tap/v2alpha/BUILD

@@ -6,7 +6,6 @@ licenses(["notice"])  # Apache 2
 
 api_proto_package(
     deps = [
-        "//envoy/api/v2/core:pkg",
         "//envoy/service/tap/v2alpha:pkg",
         "@com_github_cncf_udpa//udpa/annotations:pkg",
     ],

api/envoy/config/common/tap/v2alpha/common.proto

@@ -2,7 +2,6 @@ syntax = "proto3";
 
 package envoy.config.common.tap.v2alpha;
 
-import "envoy/api/v2/core/config_source.proto";
 import "envoy/service/tap/v2alpha/common.proto";
 
 import "udpa/annotations/migrate.proto";
@@ -19,15 +18,6 @@ option (udpa.annotations.file_status).package_version_status = FROZEN;
 
 // Common configuration for all tap extensions.
 message CommonExtensionConfig {
-  // [#not-implemented-hide:]
-  message TapDSConfig {
-    // Configuration for the source of TapDS updates for this Cluster.
-    api.v2.core.ConfigSource config_source = 1 [(validate.rules).message = {required: true}];
-
-    // Tap config to request from XDS server.
-    string name = 2 [(validate.rules).string = {min_bytes: 1}];
-  }
-
   oneof config_type {
     option (validate.required) = true;
 
@@ -37,9 +27,6 @@ message CommonExtensionConfig {
     // If specified, the tap filter will be configured via a static configuration that cannot be
     // changed.
     service.tap.v2alpha.TapConfig static_config = 2;
-
-    // [#not-implemented-hide:] Configuration to use for TapDS updates for the filter.
-    TapDSConfig tapds_config = 3;
   }
 }
 

api/envoy/extensions/common/tap/v3/BUILD

@@ -7,7 +7,6 @@ licenses(["notice"])  # Apache 2
 api_proto_package(
     deps = [
         "//envoy/config/common/tap/v2alpha:pkg",
-        "//envoy/config/core/v3:pkg",
         "//envoy/config/tap/v3:pkg",
         "@com_github_cncf_udpa//udpa/annotations:pkg",
         "@com_github_cncf_udpa//udpa/core/v1:pkg",

api/envoy/extensions/common/tap/v3/common.proto

@@ -2,7 +2,6 @@ syntax = "proto3";
 
 package envoy.extensions.common.tap.v3;
 
-import "envoy/config/core/v3/config_source.proto";
 import "envoy/config/tap/v3/common.proto";
 
 import "udpa/core/v1/resource_locator.proto";
@@ -24,23 +23,6 @@ message CommonExtensionConfig {
   option (udpa.annotations.versioning).previous_message_type =
       "envoy.config.common.tap.v2alpha.CommonExtensionConfig";
 
-  // [#not-implemented-hide:]
-  message TapDSConfig {
-    option (udpa.annotations.versioning).previous_message_type =
-        "envoy.config.common.tap.v2alpha.CommonExtensionConfig.TapDSConfig";
-
-    // Configuration for the source of TapDS updates for this Cluster.
-    config.core.v3.ConfigSource config_source = 1 [(validate.rules).message = {required: true}];
-
-    // Tap config to request from XDS server.
-    string name = 2 [(udpa.annotations.field_migrate).oneof_promotion = "name_specifier"];
-
-    // Resource locator for TAP. This is mutually exclusive to *name*.
-    // [#not-implemented-hide:]
-    udpa.core.v1.ResourceLocator tap_resource_locator = 3
-        [(udpa.annotations.field_migrate).oneof_promotion = "name_specifier"];
-  }
-
   oneof config_type {
     option (validate.required) = true;
 
@@ -50,9 +32,6 @@ message CommonExtensionConfig {
     // If specified, the tap filter will be configured via a static configuration that cannot be
     // changed.
     config.tap.v3.TapConfig static_config = 2;
-
-    // [#not-implemented-hide:] Configuration to use for TapDS updates for the filter.
-    TapDSConfig tapds_config = 3;
   }
 }
 

source/extensions/common/tap/extension_config_base.cc

@@ -40,9 +40,6 @@ ExtensionConfigBase::ExtensionConfigBase(
     ENVOY_LOG(debug, "initializing tap extension with static config");
     break;
   }
-  case envoy::extensions::common::tap::v3::CommonExtensionConfig::ConfigTypeCase::kTapdsConfig: {
-    NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
-  }
   default: {
     NOT_REACHED_GCOVR_EXCL_LINE;
   }

source/extensions/transport_sockets/tap/tap_config_impl.cc

@@ -19,15 +19,21 @@ PerSocketTapperImpl::PerSocketTapperImpl(SocketTapConfigSharedPtr config,
       connection_(connection), statuses_(config_->createMatchStatusVector()) {
   config_->rootMatcher().onNewStream(statuses_);
   if (config_->streaming() && config_->rootMatcher().matchStatus(statuses_).matches_) {
+    // TODO(mattklein123): For IP client connections, local address will not be populated until
+    // connection. We should re-emit connection information after connection so the streaming
+    // trace gets the local address.
     TapCommon::TraceWrapperPtr trace = makeTraceSegment();
     fillConnectionInfo(*trace->mutable_socket_streamed_trace_segment()->mutable_connection());
     sink_handle_->submitTrace(std::move(trace));
   }
 }
 
 void PerSocketTapperImpl::fillConnectionInfo(envoy::data::tap::v3::Connection& connection) {
-  Network::Utility::addressToProtobufAddress(*connection_.localAddress(),
-                                             *connection.mutable_local_address());
+  if (connection_.localAddress() != nullptr) {
+    // Local address might not be populated before a client connection is connected.
+    Network::Utility::addressToProtobufAddress(*connection_.localAddress(),
+                                               *connection.mutable_local_address());
+  }
   Network::Utility::addressToProtobufAddress(*connection_.remoteAddress(),
                                              *connection.mutable_remote_address());
 }

test/extensions/common/tap/common.cc

@@ -1,5 +1,7 @@
 #include "test/extensions/common/tap/common.h"
 
+#include <fstream>
+
 #include "envoy/data/tap/v3/wrapper.pb.h"
 
 namespace envoy {
@@ -21,6 +23,37 @@ namespace Extensions {
 namespace Common {
 namespace Tap {
 
+std::vector<envoy::data::tap::v3::TraceWrapper> readTracesFromPath(const std::string& path_prefix) {
+  // Find the written .pb file and verify it.
+  auto files = TestUtility::listFiles(path_prefix, false);
+  auto pb_file_name = std::find_if(files.begin(), files.end(), [](const std::string& s) {
+    return absl::EndsWith(s, MessageUtil::FileExtensions::get().ProtoBinaryLengthDelimited);
+  });
+  EXPECT_NE(pb_file_name, files.end());
+  return readTracesFromFile(*pb_file_name);
+}
+
+std::vector<envoy::data::tap::v3::TraceWrapper> readTracesFromFile(const std::string& file) {
+  std::vector<envoy::data::tap::v3::TraceWrapper> traces;
+  std::ifstream pb_file(file, std::ios_base::binary);
+  Protobuf::io::IstreamInputStream stream(&pb_file);
+  Protobuf::io::CodedInputStream coded_stream(&stream);
+  while (true) {
+    uint32_t message_size;
+    if (!coded_stream.ReadVarint32(&message_size)) {
+      break;
+    }
+
+    traces.emplace_back();
+
+    auto limit = coded_stream.PushLimit(message_size);
+    EXPECT_TRUE(traces.back().ParseFromCodedStream(&coded_stream));
+    coded_stream.PopLimit(limit);
+  }
+
+  return traces;
+}
+
 MockPerTapSinkHandleManager::MockPerTapSinkHandleManager() = default;
 MockPerTapSinkHandleManager::~MockPerTapSinkHandleManager() = default;
 

test/extensions/common/tap/common.h

@@ -36,6 +36,13 @@ MATCHER_P(TraceEqual, rhs, "") {
 namespace Common {
 namespace Tap {
 
+// Reads a PROTO_BINARY_LENGTH_DELIMITED set of messages from a file, found within the specified
+// path prefix.
+std::vector<envoy::data::tap::v3::TraceWrapper> readTracesFromPath(const std::string& path_prefix);
+
+// Reads a PROTO_BINARY_LENGTH_DELIMITED set of messages from a file.
+std::vector<envoy::data::tap::v3::TraceWrapper> readTracesFromFile(const std::string& file);
+
 class MockPerTapSinkHandleManager : public PerTapSinkHandleManager {
 public:
   MockPerTapSinkHandleManager();