xds Istio Interop: Handle failures properly (#30036)

* xds Istio Interop: Handle failures properly

* Unused parameters

* Remove unnecessary sleep

test/cpp/interop/istio_echo_server.cc

@@ -88,8 +88,8 @@ namespace grpc {
 namespace testing {
 namespace {
 
-void RunServer(std::vector<int> grpc_ports, std::set<int> xds_ports,
-               std::set<int> tls_ports) {
+void RunServer(const std::set<int>& grpc_ports, const std::set<int>& xds_ports,
+               const std::set<int>& tls_ports) {
   // Get hostname
   std::string hostname;
   char* hostname_p = grpc_gethostname();
@@ -140,10 +140,6 @@ void RunServer(std::vector<int> grpc_ports, std::set<int> xds_ports,
   if (has_xds_listeners) {
     xds_server = xds_builder.BuildAndStart();
   }
-  // 3333 is the magic port that the istio testing for k8s health checks. And
-  // it only needs TCP. So also make the gRPC server to listen on 3333.
-  builder.AddListeningPort(grpc_core::JoinHostPort("0.0.0.0", 3333),
-                           grpc::InsecureServerCredentials());
   std::unique_ptr<Server> server(builder.BuildAndStart());
   server->Wait();
 }
@@ -153,12 +149,12 @@ void RunServer(std::vector<int> grpc_ports, std::set<int> xds_ports,
 }  // namespace grpc
 
 int main(int argc, char** argv) {
-  // Preprocess argv, for two things:
-  // 1. merge duplciate flags. So "--grpc=8080 --grpc=9090" becomes
-  // "--grpc=8080,9090".
-  // 2. replace '-' to '_'. So "--istio-version=123" becomes
-  // "--istio_version=123".
-  // 3. remove --version since that is specially interpretted by absl
+  //  Preprocess argv, for two things:
+  //  1. merge duplciate flags. So "--grpc=8080 --grpc=9090" becomes
+  //  "--grpc=8080,9090".
+  //  2. replace '-' to '_'. So "--istio-version=123" becomes
+  //  "--istio_version=123".
+  //  3. remove --version since that is specially interpretted by absl
   std::map<std::string, std::vector<std::string>> argv_dict;
   for (int i = 0; i < argc; i++) {
     std::string arg(argv[i]);
@@ -195,10 +191,10 @@ int main(int argc, char** argv) {
   grpc::testing::TestEnvironment env(&new_argc, new_argv);
   grpc::testing::InitTest(&new_argc, &new_argv, true);
   // Turn gRPC ports from a string vector to an int vector.
-  std::vector<int> grpc_ports;
+  std::set<int> grpc_ports;
   for (const auto& p : absl::GetFlag(FLAGS_grpc)) {
     int grpc_port = std::stoi(p);
-    grpc_ports.push_back(grpc_port);
+    grpc_ports.insert(grpc_port);
   }
   // Create a map of which ports are supposed to use xds
   std::set<int> xds_ports;
@@ -209,6 +205,10 @@ int main(int argc, char** argv) {
       return 1;
     }
     xds_ports.insert(port);
+    // If the port does not exist in gRPC ports set, add it.
+    if (grpc_ports.find(port) == grpc_ports.end()) {
+      grpc_ports.insert(port);
+    }
   }
   // Create a map of which ports are supposed to use tls
   std::set<int> tls_ports;

test/cpp/interop/istio_echo_server_lib.cc

@@ -197,6 +197,8 @@ Status EchoTestServiceImpl::ForwardEcho(ServerContext* context,
       gpr_log(GPR_ERROR, "RPC %d failed %d: %s", i,
               calls[i].status.error_code(),
               calls[i].status.error_message().c_str());
+      response->clear_output();
+      return calls[i].status;
     }
   }
   return Status::OK;

test/cpp/interop/istio_echo_server_test.cc

@@ -25,6 +25,7 @@
 #include <grpcpp/create_channel.h>
 #include <grpcpp/security/credentials.h>
 #include <grpcpp/server_builder.h>
+#include <grpcpp/support/status.h>
 
 #include "src/core/lib/gprpp/host_port.h"
 #include "test/core/util/port.h"
@@ -47,9 +48,9 @@ class SimpleEchoTestServerImpl : public proto::EchoTestService::Service {
  public:
   explicit SimpleEchoTestServerImpl() {}
 
-  grpc::Status Echo(grpc::ServerContext* context,
+  grpc::Status Echo(grpc::ServerContext* /* context */,
                     const proto::EchoRequest* request,
-                    proto::EchoResponse* response) override {
+                    proto::EchoResponse* /* response */) override {
     GPR_ASSERT(false);
     return Status(StatusCode::INVALID_ARGUMENT, "Unexpected");
   }
@@ -57,13 +58,19 @@ class SimpleEchoTestServerImpl : public proto::EchoTestService::Service {
   grpc::Status ForwardEcho(grpc::ServerContext* /*context*/,
                            const proto::ForwardEchoRequest* request,
                            proto::ForwardEchoResponse* response) override {
+    if (fail_rpc_) {
+      return Status(UNAVAILABLE, "fail rpc");
+    }
     response->add_output(request->message());
     return Status::OK;
   }
 
+  void set_fail_rpc(bool fail_rpc) { fail_rpc_ = fail_rpc; }
+
  private:
   std::string hostname_;
   std::string forwarding_address_;
+  std::atomic<bool> fail_rpc_{false};
   // The following fields are not set yet. But we may need them later.
   //  int port_;
   //  std::string version_;
@@ -164,6 +171,22 @@ TEST_F(EchoTest, ForwardEchoTestUnhandledProtocols) {
   EXPECT_EQ(response.output()[0], "hello");
 }
 
+TEST_F(EchoTest, ForwardEchoFailure) {
+  simple_test_service_impl_.set_fail_rpc(true);
+  ClientContext context;
+  ForwardEchoRequest request;
+  ForwardEchoResponse response;
+  request.set_count(3);
+  request.set_qps(1);
+  request.set_timeout_micros(20 * 1000 * 1000);  // 20 seconds
+  // Use the unhandled protocol to make sure that we forward the request to
+  // SimpleEchoTestServerImpl.
+  request.set_url(absl::StrCat("http://", server_address_));
+  request.set_message("hello");
+  auto status = stub_->ForwardEcho(&context, request, &response);
+  ASSERT_EQ(status.error_code(), UNAVAILABLE);
+}
+
 }  // namespace
 }  // namespace testing
 }  // namespace grpc