Merge pull request #7746 from dgquintas/lb_add_md

Updates to gRPC LB

2 files changed, 128 insertions, 70 deletions

diff --git a/test/cpp/grpclb/grpclb_api_test.cc b/test/cpp/grpclb/grpclb_api_test.cc
index 33de1ee93c..e67189c69e 100644
--- a/test/cpp/grpclb/grpclb_api_test.cc
+++ b/test/cpp/grpclb/grpclb_api_test.cc
@@ -31,10 +31,12 @@
  *
  */
 
+#include <grpc++/impl/codegen/config.h>
 #include <gtest/gtest.h>
-#include <string>
 
 #include "src/core/ext/lb_policy/grpclb/load_balancer_api.h"
+#include "src/core/lib/iomgr/sockaddr.h"
+#include "src/core/lib/iomgr/sockaddr_utils.h"
 #include "src/proto/grpc/lb/v1/load_balancer.pb.h"  // C++ version
 
 namespace grpc {
@@ -45,8 +47,28 @@ using grpc::lb::v1::LoadBalanceResponse;
 
 class GrpclbTest : public ::testing::Test {};
 
+grpc::string Ip4ToPackedString(const char* ip_str) {
+  struct in_addr ip4;
+  GPR_ASSERT(inet_pton(AF_INET, ip_str, &ip4) == 1);
+  return grpc::string(reinterpret_cast<const char*>(&ip4), sizeof(ip4));
+}
+
+grpc::string PackedStringToIp(const grpc_grpclb_ip_address& pb_ip) {
+  char ip_str[46] = {0};
+  int af = -1;
+  if (pb_ip.size == 4) {
+    af = AF_INET;
+  } else if (pb_ip.size == 16) {
+    af = AF_INET6;
+  } else {
+    abort();
+  }
+  GPR_ASSERT(inet_ntop(af, pb_ip.bytes, ip_str, 46) != NULL);
+  return ip_str;
+}
+
 TEST_F(GrpclbTest, CreateRequest) {
-  const std::string service_name = "AServiceName";
+  const grpc::string service_name = "AServiceName";
   LoadBalanceRequest request;
   grpc_grpclb_request* c_req = grpc_grpclb_request_create(service_name.c_str());
   gpr_slice slice = grpc_grpclb_request_encode(c_req);
@@ -65,7 +87,7 @@ TEST_F(GrpclbTest, ParseInitialResponse) {
       initial_response->mutable_client_stats_report_interval();
   client_stats_report_interval->set_seconds(123);
   client_stats_report_interval->set_nanos(456);
-  const std::string encoded_response = response.SerializeAsString();
+  const grpc::string encoded_response = response.SerializeAsString();
   gpr_slice encoded_slice =
       gpr_slice_from_copied_string(encoded_response.c_str());
 
@@ -82,29 +104,31 @@ TEST_F(GrpclbTest, ParseResponseServerList) {
   LoadBalanceResponse response;
   auto* serverlist = response.mutable_server_list();
   auto* server = serverlist->add_servers();
-  server->set_ip_address("127.0.0.1");
+  server->set_ip_address(Ip4ToPackedString("127.0.0.1"));
   server->set_port(12345);
   server->set_drop_request(true);
   server = response.mutable_server_list()->add_servers();
-  server->set_ip_address("10.0.0.1");
+  server->set_ip_address(Ip4ToPackedString("10.0.0.1"));
   server->set_port(54321);
   server->set_drop_request(false);
   auto* expiration_interval = serverlist->mutable_expiration_interval();
   expiration_interval->set_seconds(888);
   expiration_interval->set_nanos(999);
 
-  const std::string encoded_response = response.SerializeAsString();
-  gpr_slice encoded_slice =
-      gpr_slice_from_copied_string(encoded_response.c_str());
+  const grpc::string encoded_response = response.SerializeAsString();
+  const gpr_slice encoded_slice = gpr_slice_from_copied_buffer(
+      encoded_response.data(), encoded_response.size());
   grpc_grpclb_serverlist* c_serverlist =
       grpc_grpclb_response_parse_serverlist(encoded_slice);
   ASSERT_EQ(c_serverlist->num_servers, 2ul);
   EXPECT_TRUE(c_serverlist->servers[0]->has_ip_address);
-  EXPECT_TRUE(strcmp(c_serverlist->servers[0]->ip_address, "127.0.0.1") == 0);
+  EXPECT_EQ(PackedStringToIp(c_serverlist->servers[0]->ip_address),
+            "127.0.0.1");
   EXPECT_EQ(c_serverlist->servers[0]->port, 12345);
   EXPECT_TRUE(c_serverlist->servers[0]->drop_request);
   EXPECT_TRUE(c_serverlist->servers[1]->has_ip_address);
-  EXPECT_TRUE(strcmp(c_serverlist->servers[1]->ip_address, "10.0.0.1") == 0);
+
+  EXPECT_EQ(PackedStringToIp(c_serverlist->servers[1]->ip_address), "10.0.0.1");
   EXPECT_EQ(c_serverlist->servers[1]->port, 54321);
   EXPECT_FALSE(c_serverlist->servers[1]->drop_request);
 
diff --git a/test/cpp/grpclb/grpclb_test.cc b/test/cpp/grpclb/grpclb_test.cc
index 6f6fb83c23..95abe38031 100644
--- a/test/cpp/grpclb/grpclb_test.cc
+++ b/test/cpp/grpclb/grpclb_test.cc
@@ -37,7 +37,10 @@
 #include <cstring>
 #include <string>
 
+#include <gtest/gtest.h>
+
 #include <grpc/grpc.h>
+#include <grpc/impl/codegen/byte_buffer_reader.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/host_port.h>
 #include <grpc/support/log.h>
@@ -46,9 +49,11 @@
 #include <grpc/support/thd.h>
 #include <grpc/support/time.h>
 
+#include <grpc++/impl/codegen/config.h>
 extern "C" {
 #include "src/core/ext/client_config/client_channel.h"
 #include "src/core/lib/channel/channel_stack.h"
+#include "src/core/lib/iomgr/sockaddr.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/support/tmpfile.h"
 #include "src/core/lib/surface/channel.h"
@@ -61,16 +66,19 @@ extern "C" {
 #include "src/proto/grpc/lb/v1/load_balancer.pb.h"
 
 #define NUM_BACKENDS 4
+#define PAYLOAD "hello you"
 
 // TODO(dgq): Other scenarios in need of testing:
 // - Send an empty serverlist update and verify that the client request blocks
 //   until a new serverlist with actual contents is available.
 // - Send identical serverlist update
+// - Send a serverlist with faulty ip:port addresses (port > 2^16, etc).
 // - Test reception of invalid serverlist
 // - Test pinging
 // - Test against a non-LB server. That server should return UNIMPLEMENTED and
 //   the call should fail.
 // - Random LB server closing the stream unexpectedly.
+// - Test using DNS-resolvable names (localhost?)
 
 namespace grpc {
 namespace {
@@ -105,8 +113,8 @@ static gpr_slice build_response_payload_slice(
     int64_t expiration_interval_secs, int32_t expiration_interval_nanos) {
   // server_list {
   //   servers {
-  //     ip_address: "127.0.0.1"
-  //     port: ...
+  //     ip_address: <in_addr/6 bytes of an IP>
+  //     port: <16 bit uint>
   //     load_balance_token: "token..."
   //   }
   //   ...
@@ -125,21 +133,21 @@ static gpr_slice build_response_payload_slice(
   }
   for (size_t i = 0; i < nports; i++) {
     auto *server = serverlist->add_servers();
-    server->set_ip_address(host);
+    // TODO(dgq): test ipv6
+    struct in_addr ip4;
+    GPR_ASSERT(inet_pton(AF_INET, host, &ip4) == 1);
+    server->set_ip_address(
+        grpc::string(reinterpret_cast<const char *>(&ip4), sizeof(ip4)));
     server->set_port(ports[i]);
     // The following long long int cast is meant to work around the
     // disfunctional implementation of std::to_string in gcc 4.4, which doesn't
     // have a version for int but does have one for long long int.
-    server->set_load_balance_token("token" +
-                                   std::to_string((long long int)ports[i]));
+    string token_data = "token" + std::to_string((long long int)ports[i]);
+    token_data.resize(64, '-');
+    server->set_load_balance_token(token_data);
   }
-
-  gpr_log(GPR_INFO, "generating response: %s",
-          response.ShortDebugString().c_str());
-
-  const gpr_slice response_slice =
-      gpr_slice_from_copied_string(response.SerializeAsString().c_str());
-  return response_slice;
+  const grpc::string &enc_resp = response.SerializeAsString();
+  return gpr_slice_from_copied_buffer(enc_resp.data(), enc_resp.size());
 }
 
 static void drain_cq(grpc_completion_queue *cq) {
@@ -181,20 +189,9 @@ static void start_lb_server(server_fixture *sf, int *ports, size_t nports,
   cq_verify(cqv);
   gpr_log(GPR_INFO, "LB Server[%s] after tag 200", sf->servers_hostport);
 
-  op = ops;
-  op->op = GRPC_OP_SEND_INITIAL_METADATA;
-  op->data.send_initial_metadata.count = 0;
-  op->flags = 0;
-  op->reserved = NULL;
-  op++;
-  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
-  op->data.recv_close_on_server.cancelled = &was_cancelled;
-  op->flags = 0;
-  op->reserved = NULL;
-  op++;
-  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(201), NULL);
-  GPR_ASSERT(GRPC_CALL_OK == error);
-  gpr_log(GPR_INFO, "LB Server[%s] after tag 201", sf->servers_hostport);
+  // make sure we've received the initial metadata from the grpclb request.
+  GPR_ASSERT(request_metadata_recv.count > 0);
+  GPR_ASSERT(request_metadata_recv.metadata != NULL);
 
   // receive request for backends
   op = ops;
@@ -208,9 +205,36 @@ static void start_lb_server(server_fixture *sf, int *ports, size_t nports,
   CQ_EXPECT_COMPLETION(cqv, tag(202), 1);
   cq_verify(cqv);
   gpr_log(GPR_INFO, "LB Server[%s] after RECV_MSG", sf->servers_hostport);
-  // TODO(dgq): validate request.
+
+  // validate initial request.
+  grpc_byte_buffer_reader bbr;
+  grpc_byte_buffer_reader_init(&bbr, request_payload_recv);
+  gpr_slice request_payload_slice = grpc_byte_buffer_reader_readall(&bbr);
+  grpc::lb::v1::LoadBalanceRequest request;
+  request.ParseFromArray(GPR_SLICE_START_PTR(request_payload_slice),
+                         GPR_SLICE_LENGTH(request_payload_slice));
+  GPR_ASSERT(request.has_initial_request());
+  GPR_ASSERT(request.initial_request().name() == "load.balanced.service.name");
+  gpr_slice_unref(request_payload_slice);
+  grpc_byte_buffer_reader_destroy(&bbr);
   grpc_byte_buffer_destroy(request_payload_recv);
+
   gpr_slice response_payload_slice;
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op->flags = 0;
+  op->reserved = NULL;
+  op++;
+  error = grpc_call_start_batch(s, ops, (size_t)(op - ops), tag(201), NULL);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  gpr_log(GPR_INFO, "LB Server[%s] after tag 201", sf->servers_hostport);
+
   for (int i = 0; i < 2; i++) {
     if (i == 0) {
       // First half of the ports.
@@ -303,6 +327,16 @@ static void start_backend_server(server_fixture *sf) {
       return;
     }
     GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+
+    // The following long long int cast is meant to work around the
+    // disfunctional implementation of std::to_string in gcc 4.4, which doesn't
+    // have a version for int but does have one for long long int.
+    string expected_token = "token" + std::to_string((long long int)sf->port);
+    expected_token.resize(64, '-');
+    GPR_ASSERT(contains_metadata(&request_metadata_recv,
+                                 "load-reporting-initial",
+                                 expected_token.c_str()));
+
     gpr_log(GPR_INFO, "Server[%s] after tag 100", sf->servers_hostport);
 
     op = ops;
@@ -321,8 +355,7 @@ static void start_backend_server(server_fixture *sf) {
     gpr_log(GPR_INFO, "Server[%s] after tag 101", sf->servers_hostport);
 
     bool exit = false;
-    gpr_slice response_payload_slice =
-        gpr_slice_from_copied_string("hello you");
+    gpr_slice response_payload_slice = gpr_slice_from_copied_string(PAYLOAD);
     while (!exit) {
       op = ops;
       op->op = GRPC_OP_RECV_MESSAGE;
@@ -474,10 +507,9 @@ static void perform_request(client_fixture *cf) {
     error = grpc_call_start_batch(c, ops, (size_t)(op - ops), tag(2), NULL);
     GPR_ASSERT(GRPC_CALL_OK == error);
 
-    peer = grpc_call_get_peer(c);
     CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
     cq_verify(cqv);
-    gpr_free(peer);
+    GPR_ASSERT(byte_buffer_eq_string(response_payload_recv, PAYLOAD));
 
     grpc_byte_buffer_destroy(request_payload);
     grpc_byte_buffer_destroy(response_payload_recv);
@@ -583,27 +615,30 @@ static void fork_lb_server(void *arg) {
                   tf->lb_server_update_delay_ms);
 }
 
-static void setup_test_fixture(test_fixture *tf,
-                               int lb_server_update_delay_ms) {
-  tf->lb_server_update_delay_ms = lb_server_update_delay_ms;
+static test_fixture setup_test_fixture(int lb_server_update_delay_ms) {
+  test_fixture tf;
+  memset(&tf, 0, sizeof(tf));
+  tf.lb_server_update_delay_ms = lb_server_update_delay_ms;
 
   gpr_thd_options options = gpr_thd_options_default();
   gpr_thd_options_set_joinable(&options);
 
   for (int i = 0; i < NUM_BACKENDS; ++i) {
-    setup_server("127.0.0.1", &tf->lb_backends[i]);
-    gpr_thd_new(&tf->lb_backends[i].tid, fork_backend_server,
-                &tf->lb_backends[i], &options);
+    setup_server("127.0.0.1", &tf.lb_backends[i]);
+    gpr_thd_new(&tf.lb_backends[i].tid, fork_backend_server, &tf.lb_backends[i],
+                &options);
   }
 
-  setup_server("127.0.0.1", &tf->lb_server);
-  gpr_thd_new(&tf->lb_server.tid, fork_lb_server, &tf->lb_server, &options);
+  setup_server("127.0.0.1", &tf.lb_server);
+  gpr_thd_new(&tf.lb_server.tid, fork_lb_server, &tf.lb_server, &options);
 
   char *server_uri;
   gpr_asprintf(&server_uri, "ipv4:%s?lb_policy=grpclb&lb_enabled=1",
-               tf->lb_server.servers_hostport);
-  setup_client(server_uri, &tf->client);
+               tf.lb_server.servers_hostport);
+  setup_client(server_uri, &tf.client);
   gpr_free(server_uri);
+
+  return tf;
 }
 
 static void teardown_test_fixture(test_fixture *tf) {
@@ -614,19 +649,13 @@ static void teardown_test_fixture(test_fixture *tf) {
   teardown_server(&tf->lb_server);
 }
 
-// The LB server will send two updates: batch 1 and batch 2. Each batch
-// contains
-// two addresses, both of a valid and running backend server. Batch 1 is
-// readily
-// available and provided as soon as the client establishes the streaming
-// call.
-// Batch 2 is sent after a delay of \a lb_server_update_delay_ms
-// milliseconds.
+// The LB server will send two updates: batch 1 and batch 2. Each batch contains
+// two addresses, both of a valid and running backend server. Batch 1 is readily
+// available and provided as soon as the client establishes the streaming call.
+// Batch 2 is sent after a delay of \a lb_server_update_delay_ms milliseconds.
 static test_fixture test_update(int lb_server_update_delay_ms) {
   gpr_log(GPR_INFO, "start %s(%d)", __func__, lb_server_update_delay_ms);
-  test_fixture tf;
-  memset(&tf, 0, sizeof(tf));
-  setup_test_fixture(&tf, lb_server_update_delay_ms);
+  test_fixture tf = setup_test_fixture(lb_server_update_delay_ms);
   perform_request(
       &tf.client);  // "consumes" 1st backend server of 1st serverlist
   perform_request(
@@ -642,13 +671,7 @@ static test_fixture test_update(int lb_server_update_delay_ms) {
   return tf;
 }
 
-}  // namespace
-}  // namespace grpc
-
-int main(int argc, char **argv) {
-  grpc_test_init(argc, argv);
-  grpc_init();
-
+TEST(GrpclbTest, Updates) {
   grpc::test_fixture tf_result;
   // Clients take a bit over one second to complete a call (the last part of the
   // call sleeps for 1 second while verifying the client's completion queue is
@@ -683,7 +706,18 @@ int main(int argc, char **argv) {
   GPR_ASSERT(tf_result.lb_backends[1].num_calls_serviced > 0);
   GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced > 0);
   GPR_ASSERT(tf_result.lb_backends[3].num_calls_serviced == 0);
+}
 
+TEST(GrpclbTest, InvalidAddressInServerlist) {}
+
+}  // namespace
+}  // namespace grpc
+
+int main(int argc, char **argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  grpc_test_init(argc, argv);
+  grpc_init();
+  const auto result = RUN_ALL_TESTS();
   grpc_shutdown();
-  return 0;
+  return result;
 }