Merge pull request #14444 from dgquintas/kill_grpclb_test

Removed outdated grpclb_test
pull/14162/head
David G. Quintas 7 years ago committed by GitHub
commit b0d71823a0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 46
      CMakeLists.txt
  2. 52
      Makefile
  3. 14
      build.yaml
  4. 799
      test/cpp/grpclb/grpclb_test.cc
  5. 23
      tools/run_tests/generated/sources_and_headers.json
  6. 24
      tools/run_tests/generated/tests.json

@ -538,7 +538,6 @@ add_dependencies(buildtests_cxx grpc_cli)
add_dependencies(buildtests_cxx grpc_tool_test)
add_dependencies(buildtests_cxx grpclb_api_test)
add_dependencies(buildtests_cxx grpclb_end2end_test)
add_dependencies(buildtests_cxx grpclb_test)
add_dependencies(buildtests_cxx h2_ssl_cert_test)
add_dependencies(buildtests_cxx health_service_end2end_test)
if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
@ -10631,51 +10630,6 @@ target_link_libraries(grpclb_end2end_test
endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
add_executable(grpclb_test
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/lb/v1/load_balancer.pb.cc
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/lb/v1/load_balancer.grpc.pb.cc
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/lb/v1/load_balancer.pb.h
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/lb/v1/load_balancer.grpc.pb.h
test/cpp/grpclb/grpclb_test.cc
third_party/googletest/googletest/src/gtest-all.cc
third_party/googletest/googlemock/src/gmock-all.cc
)
protobuf_generate_grpc_cpp(
src/proto/grpc/lb/v1/load_balancer.proto
)
target_include_directories(grpclb_test
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}
PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include
PRIVATE ${_gRPC_SSL_INCLUDE_DIR}
PRIVATE ${_gRPC_PROTOBUF_INCLUDE_DIR}
PRIVATE ${_gRPC_ZLIB_INCLUDE_DIR}
PRIVATE ${_gRPC_BENCHMARK_INCLUDE_DIR}
PRIVATE ${_gRPC_CARES_INCLUDE_DIR}
PRIVATE ${_gRPC_GFLAGS_INCLUDE_DIR}
PRIVATE third_party/googletest/googletest/include
PRIVATE third_party/googletest/googletest
PRIVATE third_party/googletest/googlemock/include
PRIVATE third_party/googletest/googlemock
PRIVATE ${_gRPC_PROTO_GENS_DIR}
)
target_link_libraries(grpclb_test
${_gRPC_PROTOBUF_LIBRARIES}
${_gRPC_ALLTARGETS_LIBRARIES}
grpc++_test_util
grpc_test_util
grpc++
grpc
gpr_test_util
gpr
${_gRPC_GFLAGS_LIBRARIES}
)
endif (gRPC_BUILD_TESTS)
if (gRPC_BUILD_TESTS)
add_executable(h2_ssl_cert_test
test/core/end2end/h2_ssl_cert_test.cc
third_party/googletest/googletest/src/gtest-all.cc

@ -1145,7 +1145,6 @@ grpc_ruby_plugin: $(BINDIR)/$(CONFIG)/grpc_ruby_plugin
grpc_tool_test: $(BINDIR)/$(CONFIG)/grpc_tool_test
grpclb_api_test: $(BINDIR)/$(CONFIG)/grpclb_api_test
grpclb_end2end_test: $(BINDIR)/$(CONFIG)/grpclb_end2end_test
grpclb_test: $(BINDIR)/$(CONFIG)/grpclb_test
h2_ssl_cert_test: $(BINDIR)/$(CONFIG)/h2_ssl_cert_test
health_service_end2end_test: $(BINDIR)/$(CONFIG)/health_service_end2end_test
http2_client: $(BINDIR)/$(CONFIG)/http2_client
@ -1604,7 +1603,6 @@ buildtests_cxx: privatelibs_cxx \
$(BINDIR)/$(CONFIG)/grpc_tool_test \
$(BINDIR)/$(CONFIG)/grpclb_api_test \
$(BINDIR)/$(CONFIG)/grpclb_end2end_test \
$(BINDIR)/$(CONFIG)/grpclb_test \
$(BINDIR)/$(CONFIG)/h2_ssl_cert_test \
$(BINDIR)/$(CONFIG)/health_service_end2end_test \
$(BINDIR)/$(CONFIG)/http2_client \
@ -1749,7 +1747,6 @@ buildtests_cxx: privatelibs_cxx \
$(BINDIR)/$(CONFIG)/grpc_tool_test \
$(BINDIR)/$(CONFIG)/grpclb_api_test \
$(BINDIR)/$(CONFIG)/grpclb_end2end_test \
$(BINDIR)/$(CONFIG)/grpclb_test \
$(BINDIR)/$(CONFIG)/h2_ssl_cert_test \
$(BINDIR)/$(CONFIG)/health_service_end2end_test \
$(BINDIR)/$(CONFIG)/http2_client \
@ -2163,8 +2160,6 @@ test_cxx: buildtests_cxx
$(Q) $(BINDIR)/$(CONFIG)/grpclb_api_test || ( echo test grpclb_api_test failed ; exit 1 )
$(E) "[RUN] Testing grpclb_end2end_test"
$(Q) $(BINDIR)/$(CONFIG)/grpclb_end2end_test || ( echo test grpclb_end2end_test failed ; exit 1 )
$(E) "[RUN] Testing grpclb_test"
$(Q) $(BINDIR)/$(CONFIG)/grpclb_test || ( echo test grpclb_test failed ; exit 1 )
$(E) "[RUN] Testing h2_ssl_cert_test"
$(Q) $(BINDIR)/$(CONFIG)/h2_ssl_cert_test || ( echo test h2_ssl_cert_test failed ; exit 1 )
$(E) "[RUN] Testing health_service_end2end_test"
@ -16359,53 +16354,6 @@ endif
$(OBJDIR)/$(CONFIG)/test/cpp/end2end/grpclb_end2end_test.o: $(GENDIR)/src/proto/grpc/lb/v1/load_balancer.pb.cc $(GENDIR)/src/proto/grpc/lb/v1/load_balancer.grpc.pb.cc
GRPCLB_TEST_SRC = \
$(GENDIR)/src/proto/grpc/lb/v1/load_balancer.pb.cc $(GENDIR)/src/proto/grpc/lb/v1/load_balancer.grpc.pb.cc \
test/cpp/grpclb/grpclb_test.cc \
GRPCLB_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(GRPCLB_TEST_SRC))))
ifeq ($(NO_SECURE),true)
# You can't build secure targets if you don't have OpenSSL.
$(BINDIR)/$(CONFIG)/grpclb_test: openssl_dep_error
else
ifeq ($(NO_PROTOBUF),true)
# You can't build the protoc plugins or protobuf-enabled targets if you don't have protobuf 3.0.0+.
$(BINDIR)/$(CONFIG)/grpclb_test: protobuf_dep_error
else
$(BINDIR)/$(CONFIG)/grpclb_test: $(PROTOBUF_DEP) $(GRPCLB_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(E) "[LD] Linking $@"
$(Q) mkdir -p `dirname $@`
$(Q) $(LDXX) $(LDFLAGS) $(GRPCLB_TEST_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBSXX) $(LDLIBS_PROTOBUF) $(LDLIBS) $(LDLIBS_SECURE) $(GTEST_LIB) -o $(BINDIR)/$(CONFIG)/grpclb_test
endif
endif
$(OBJDIR)/$(CONFIG)/src/proto/grpc/lb/v1/load_balancer.o: $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
$(OBJDIR)/$(CONFIG)/test/cpp/grpclb/grpclb_test.o: $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
deps_grpclb_test: $(GRPCLB_TEST_OBJS:.o=.dep)
ifneq ($(NO_SECURE),true)
ifneq ($(NO_DEPS),true)
-include $(GRPCLB_TEST_OBJS:.o=.dep)
endif
endif
$(OBJDIR)/$(CONFIG)/test/cpp/grpclb/grpclb_test.o: $(GENDIR)/src/proto/grpc/lb/v1/load_balancer.pb.cc $(GENDIR)/src/proto/grpc/lb/v1/load_balancer.grpc.pb.cc
H2_SSL_CERT_TEST_SRC = \
test/core/end2end/h2_ssl_cert_test.cc \

@ -4272,20 +4272,6 @@ targets:
- grpc
- gpr_test_util
- gpr
- name: grpclb_test
gtest: false
build: test
language: c++
src:
- src/proto/grpc/lb/v1/load_balancer.proto
- test/cpp/grpclb/grpclb_test.cc
deps:
- grpc++_test_util
- grpc_test_util
- grpc++
- grpc
- gpr_test_util
- gpr
- name: h2_ssl_cert_test
gtest: true
build: test

@ -1,799 +0,0 @@
/*
*
* Copyright 2016 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
#include <cinttypes>
#include <cstdarg>
#include <cstdint>
#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/log.h>
#include <grpc/support/string_util.h>
#include <grpc/support/sync.h>
#include <grpc/support/time.h>
#include <grpc++/impl/codegen/config.h>
#include "src/core/ext/filters/client_channel/client_channel.h"
#include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/gpr/env.h"
#include "src/core/lib/gpr/host_port.h"
#include "src/core/lib/gpr/string.h"
#include "src/core/lib/gpr/thd.h"
#include "src/core/lib/gpr/tmpfile.h"
#include "src/core/lib/gprpp/ref_counted_ptr.h"
#include "src/core/lib/iomgr/sockaddr.h"
#include "src/core/lib/security/credentials/fake/fake_credentials.h"
#include "src/core/lib/surface/channel.h"
#include "src/core/lib/surface/server.h"
#include "test/core/end2end/cq_verifier.h"
#include "test/core/util/port.h"
#include "test/core/util/test_config.h"
#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.
// - Random LB server closing the stream unexpectedly.
// - Test using DNS-resolvable names (localhost?)
// - Test handling of creation of faulty RR instance by having the LB return a
// serverlist with non-existent backends after having initially returned a
// valid one.
//
// Findings from end to end testing to be covered here:
// - Handling of LB servers restart, including reconnection after backing-off
// retries.
// - Destruction of load balanced channel (and therefore of grpclb instance)
// while:
// 1) the internal LB call is still active. This should work by virtue
// of the weak reference the LB call holds. The call should be terminated as
// part of the grpclb shutdown process.
// 2) the retry timer is active. Again, the weak reference it holds should
// prevent a premature call to \a glb_destroy.
// - Restart of backend servers with no changes to serverlist. This exercises
// the RR handover mechanism.
namespace grpc {
namespace {
typedef struct client_fixture {
grpc_channel* client;
char* server_uri;
grpc_completion_queue* cq;
} client_fixture;
typedef struct server_fixture {
grpc_server* server;
grpc_call* server_call;
grpc_completion_queue* cq;
char* servers_hostport;
const char* balancer_name;
int port;
const char* lb_token_prefix;
gpr_thd_id tid;
int num_calls_serviced;
} server_fixture;
typedef struct test_fixture {
server_fixture lb_server;
server_fixture lb_backends[NUM_BACKENDS];
client_fixture client;
int lb_server_update_delay_ms;
} test_fixture;
static void* tag(intptr_t t) { return (void*)t; }
static grpc_slice build_response_payload_slice(const char* host, int* ports,
size_t nports,
const char* token_prefix) {
// server_list {
// servers {
// ip_address: <in_addr/6 bytes of an IP>
// port: <16 bit uint>
// load_balance_token: "token..."
// }
// ...
// }
grpc::lb::v1::LoadBalanceResponse response;
auto* serverlist = response.mutable_server_list();
for (size_t i = 0; i < nports; i++) {
auto* server = serverlist->add_servers();
// TODO(dgq): test ipv6
struct in_addr ip4;
GPR_ASSERT(inet_pton(AF_INET, host, &ip4) == 1);
server->set_ip_address(
string(reinterpret_cast<const char*>(&ip4), sizeof(ip4)));
server->set_port(ports[i]);
// Missing tokens are acceptable. Test that path.
if (strlen(token_prefix) > 0) {
string token_data = token_prefix + std::to_string(ports[i]);
server->set_load_balance_token(token_data);
}
}
const string& enc_resp = response.SerializeAsString();
return grpc_slice_from_copied_buffer(enc_resp.data(), enc_resp.size());
}
static void drain_cq(grpc_completion_queue* cq) {
grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, grpc_timeout_seconds_to_deadline(5),
nullptr);
} while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void sleep_ms(int delay_ms) {
gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_millis(delay_ms, GPR_TIMESPAN)));
}
static void start_lb_server(server_fixture* sf, int* ports, size_t nports,
int update_delay_ms) {
grpc_call* s;
cq_verifier* cqv = cq_verifier_create(sf->cq);
grpc_op ops[6];
grpc_op* op;
grpc_metadata_array request_metadata_recv;
grpc_call_details call_details;
grpc_call_error error;
int was_cancelled = 2;
grpc_byte_buffer* request_payload_recv;
grpc_byte_buffer* response_payload;
memset(ops, 0, sizeof(ops));
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
error = grpc_server_request_call(sf->server, &s, &call_details,
&request_metadata_recv, sf->cq, sf->cq,
tag(200));
GPR_ASSERT(GRPC_CALL_OK == error);
gpr_log(GPR_INFO, "LB Server[%s](%s) up", sf->servers_hostport,
sf->balancer_name);
CQ_EXPECT_COMPLETION(cqv, tag(200), 1);
cq_verify(cqv);
gpr_log(GPR_INFO, "LB Server[%s](%s) after tag 200", sf->servers_hostport,
sf->balancer_name);
// 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 != nullptr);
// receive request for backends
op = ops;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message.recv_message = &request_payload_recv;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(202),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(202), 1);
cq_verify(cqv);
gpr_log(GPR_INFO, "LB Server[%s](%s) after RECV_MSG", sf->servers_hostport,
sf->balancer_name);
// validate initial request.
grpc_byte_buffer_reader bbr;
grpc_byte_buffer_reader_init(&bbr, request_payload_recv);
grpc_slice request_payload_slice = grpc_byte_buffer_reader_readall(&bbr);
grpc::lb::v1::LoadBalanceRequest request;
request.ParseFromArray(GRPC_SLICE_START_PTR(request_payload_slice),
GRPC_SLICE_LENGTH(request_payload_slice));
GPR_ASSERT(request.has_initial_request());
GPR_ASSERT(request.initial_request().name() == sf->servers_hostport);
grpc_slice_unref(request_payload_slice);
grpc_byte_buffer_reader_destroy(&bbr);
grpc_byte_buffer_destroy(request_payload_recv);
grpc_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 = nullptr;
op++;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(201),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
gpr_log(GPR_INFO, "LB Server[%s](%s) after tag 201", sf->servers_hostport,
sf->balancer_name);
for (int i = 0; i < 2; i++) {
if (i == 0) {
// First half of the ports.
response_payload_slice = build_response_payload_slice(
"127.0.0.1", ports, nports / 2, sf->lb_token_prefix);
} else {
// Second half of the ports.
sleep_ms(update_delay_ms);
response_payload_slice = build_response_payload_slice(
"127.0.0.1", ports + (nports / 2), (nports + 1) / 2 /* ceil */,
"" /* this half doesn't get to receive an LB token */);
}
response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1);
op = ops;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message.send_message = response_payload;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
tag(203), nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(203), 1);
cq_verify(cqv);
gpr_log(GPR_INFO, "LB Server[%s](%s) after SEND_MESSAGE, iter %d",
sf->servers_hostport, sf->balancer_name, i);
grpc_byte_buffer_destroy(response_payload);
grpc_slice_unref(response_payload_slice);
}
gpr_log(GPR_INFO, "LB Server[%s](%s) shutting down", sf->servers_hostport,
sf->balancer_name);
op = ops;
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
op->data.send_status_from_server.trailing_metadata_count = 0;
op->data.send_status_from_server.status = GRPC_STATUS_OK;
grpc_slice status_details = grpc_slice_from_static_string("xyz");
op->data.send_status_from_server.status_details = &status_details;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(204),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(201), 1);
CQ_EXPECT_COMPLETION(cqv, tag(204), 1);
cq_verify(cqv);
gpr_log(GPR_INFO, "LB Server[%s](%s) after tag 204. All done. LB server out",
sf->servers_hostport, sf->balancer_name);
grpc_call_unref(s);
cq_verifier_destroy(cqv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
}
static void start_backend_server(server_fixture* sf) {
grpc_call* s;
cq_verifier* cqv;
grpc_op ops[6];
grpc_op* op;
grpc_metadata_array request_metadata_recv;
grpc_call_details call_details;
grpc_call_error error;
int was_cancelled;
grpc_byte_buffer* request_payload_recv;
grpc_byte_buffer* response_payload;
grpc_event ev;
while (true) {
memset(ops, 0, sizeof(ops));
cqv = cq_verifier_create(sf->cq);
was_cancelled = 2;
grpc_metadata_array_init(&request_metadata_recv);
grpc_call_details_init(&call_details);
error = grpc_server_request_call(sf->server, &s, &call_details,
&request_metadata_recv, sf->cq, sf->cq,
tag(100));
GPR_ASSERT(GRPC_CALL_OK == error);
gpr_log(GPR_INFO, "Server[%s] up", sf->servers_hostport);
ev = grpc_completion_queue_next(
sf->cq, grpc_timeout_seconds_to_deadline(60), nullptr);
if (!ev.success) {
gpr_log(GPR_INFO, "Server[%s] being torn down", sf->servers_hostport);
cq_verifier_destroy(cqv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
return;
}
GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
const string expected_token =
strlen(sf->lb_token_prefix) == 0
? ""
: sf->lb_token_prefix + std::to_string(sf->port);
GPR_ASSERT(contains_metadata(&request_metadata_recv, "lb-token",
expected_token.c_str()));
gpr_log(GPR_INFO, "Server[%s] after tag 100", sf->servers_hostport);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
op->data.recv_close_on_server.cancelled = &was_cancelled;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
tag(101), nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
gpr_log(GPR_INFO, "Server[%s] after tag 101", sf->servers_hostport);
bool exit = false;
grpc_slice response_payload_slice = grpc_slice_from_copied_string(PAYLOAD);
while (!exit) {
op = ops;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message.recv_message = &request_payload_recv;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
tag(102), nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
ev = grpc_completion_queue_next(
sf->cq, grpc_timeout_seconds_to_deadline(3), nullptr);
if (ev.type == GRPC_OP_COMPLETE && ev.success) {
GPR_ASSERT(ev.tag = tag(102));
if (request_payload_recv == nullptr) {
exit = true;
gpr_log(GPR_INFO,
"Server[%s] recv \"close\" from client, exiting. Call #%d",
sf->servers_hostport, sf->num_calls_serviced);
}
} else {
gpr_log(GPR_INFO, "Server[%s] forced to shutdown. Call #%d",
sf->servers_hostport, sf->num_calls_serviced);
exit = true;
}
gpr_log(GPR_INFO, "Server[%s] after tag 102. Call #%d",
sf->servers_hostport, sf->num_calls_serviced);
if (!exit) {
response_payload =
grpc_raw_byte_buffer_create(&response_payload_slice, 1);
op = ops;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message.send_message = response_payload;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
tag(103), nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
ev = grpc_completion_queue_next(
sf->cq, grpc_timeout_seconds_to_deadline(3), nullptr);
if (ev.type == GRPC_OP_COMPLETE && ev.success) {
GPR_ASSERT(ev.tag = tag(103));
} else {
gpr_log(GPR_INFO, "Server[%s] forced to shutdown. Call #%d",
sf->servers_hostport, sf->num_calls_serviced);
exit = true;
}
gpr_log(GPR_INFO, "Server[%s] after tag 103. Call #%d",
sf->servers_hostport, sf->num_calls_serviced);
grpc_byte_buffer_destroy(response_payload);
}
grpc_byte_buffer_destroy(request_payload_recv);
}
++sf->num_calls_serviced;
gpr_log(GPR_INFO, "Server[%s] OUT OF THE LOOP", sf->servers_hostport);
grpc_slice_unref(response_payload_slice);
op = ops;
op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
op->data.send_status_from_server.trailing_metadata_count = 0;
op->data.send_status_from_server.status = GRPC_STATUS_OK;
grpc_slice status_details =
grpc_slice_from_static_string("Backend server out a-ok");
op->data.send_status_from_server.status_details = &status_details;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
tag(104), nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
CQ_EXPECT_COMPLETION(cqv, tag(104), 1);
cq_verify(cqv);
gpr_log(GPR_INFO, "Server[%s] DONE. After servicing %d calls",
sf->servers_hostport, sf->num_calls_serviced);
grpc_call_unref(s);
cq_verifier_destroy(cqv);
grpc_metadata_array_destroy(&request_metadata_recv);
grpc_call_details_destroy(&call_details);
}
}
static void perform_request(client_fixture* cf) {
grpc_call* c;
cq_verifier* cqv = cq_verifier_create(cf->cq);
grpc_op ops[6];
grpc_op* op;
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_status_code status;
grpc_call_error error;
grpc_slice details;
grpc_byte_buffer* request_payload;
grpc_byte_buffer* response_payload_recv;
int i;
memset(ops, 0, sizeof(ops));
grpc_slice request_payload_slice =
grpc_slice_from_copied_string("hello world");
grpc_slice host = grpc_slice_from_static_string("foo.test.google.fr:1234");
c = grpc_channel_create_call(cf->client, nullptr, GRPC_PROPAGATE_DEFAULTS,
cf->cq, grpc_slice_from_static_string("/foo"),
&host, grpc_timeout_seconds_to_deadline(5),
nullptr);
gpr_log(GPR_INFO, "Call 0x%" PRIxPTR " created", (intptr_t)c);
GPR_ASSERT(c);
char* peer;
grpc_metadata_array_init(&initial_metadata_recv);
grpc_metadata_array_init(&trailing_metadata_recv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
op->data.send_initial_metadata.count = 0;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_RECV_INITIAL_METADATA;
op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
op->data.recv_status_on_client.status = &status;
op->data.recv_status_on_client.status_details = &details;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
for (i = 0; i < 4; i++) {
request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
op = ops;
op->op = GRPC_OP_SEND_MESSAGE;
op->data.send_message.send_message = request_payload;
op->flags = 0;
op->reserved = nullptr;
op++;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message.recv_message = &response_payload_recv;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(2),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
cq_verify(cqv);
gpr_log(GPR_INFO, "Client after sending msg %d / 4", i + 1);
GPR_ASSERT(byte_buffer_eq_string(response_payload_recv, PAYLOAD));
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload_recv);
}
grpc_slice_unref(request_payload_slice);
op = ops;
op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
op->flags = 0;
op->reserved = nullptr;
op++;
error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(3),
nullptr);
GPR_ASSERT(GRPC_CALL_OK == error);
CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
CQ_EXPECT_COMPLETION(cqv, tag(3), 1);
cq_verify(cqv);
peer = grpc_call_get_peer(c);
gpr_log(GPR_INFO, "Client DONE WITH SERVER %s ", peer);
grpc_call_unref(c);
cq_verify_empty_timeout(cqv, 1 /* seconds */);
cq_verifier_destroy(cqv);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
grpc_slice_unref(details);
gpr_log(GPR_INFO, "Client call (peer %s) DESTROYED.", peer);
gpr_free(peer);
}
#define BALANCERS_NAME "lb.name"
static void setup_client(const server_fixture* lb_server,
const server_fixture* backends, client_fixture* cf) {
grpc_core::ExecCtx exec_ctx;
char* expected_target_names = nullptr;
const char* backends_name = lb_server->servers_hostport;
gpr_asprintf(&expected_target_names, "%s;%s", backends_name, BALANCERS_NAME);
auto response_generator =
grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
grpc_lb_addresses* addresses = grpc_lb_addresses_create(1, nullptr);
char* lb_uri_str;
gpr_asprintf(&lb_uri_str, "ipv4:%s", lb_server->servers_hostport);
grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str, true);
GPR_ASSERT(lb_uri != nullptr);
grpc_lb_addresses_set_address_from_uri(addresses, 0, lb_uri, true,
lb_server->balancer_name, nullptr);
grpc_uri_destroy(lb_uri);
gpr_free(lb_uri_str);
gpr_asprintf(&cf->server_uri, "fake:///%s", lb_server->servers_hostport);
const grpc_arg fake_addresses =
grpc_lb_addresses_create_channel_arg(addresses);
grpc_channel_args* fake_result =
grpc_channel_args_copy_and_add(nullptr, &fake_addresses, 1);
grpc_lb_addresses_destroy(addresses);
grpc_arg new_args[] = {
grpc_fake_transport_expected_targets_arg(expected_target_names),
grpc_core::FakeResolverResponseGenerator::MakeChannelArg(
response_generator.get())};
grpc_channel_args args = {GPR_ARRAY_SIZE(new_args), new_args};
cf->cq = grpc_completion_queue_create_for_next(nullptr);
grpc_channel_credentials* fake_creds =
grpc_fake_transport_security_credentials_create();
cf->client =
grpc_secure_channel_create(fake_creds, cf->server_uri, &args, nullptr);
response_generator->SetResponse(fake_result);
grpc_channel_args_destroy(fake_result);
grpc_channel_credentials_unref(fake_creds);
gpr_free(expected_target_names);
}
static void teardown_client(client_fixture* cf) {
grpc_completion_queue_shutdown(cf->cq);
drain_cq(cf->cq);
grpc_completion_queue_destroy(cf->cq);
cf->cq = nullptr;
grpc_channel_destroy(cf->client);
cf->client = nullptr;
gpr_free(cf->server_uri);
}
static void setup_server(const char* host, server_fixture* sf) {
int assigned_port;
sf->cq = grpc_completion_queue_create_for_next(nullptr);
const char* colon_idx = strchr(host, ':');
if (colon_idx) {
const char* port_str = colon_idx + 1;
sf->port = atoi(port_str);
sf->servers_hostport = gpr_strdup(host);
} else {
sf->port = grpc_pick_unused_port_or_die();
gpr_join_host_port(&sf->servers_hostport, host, sf->port);
}
grpc_server_credentials* server_creds =
grpc_fake_transport_security_server_credentials_create();
sf->server = grpc_server_create(nullptr, nullptr);
grpc_server_register_completion_queue(sf->server, sf->cq, nullptr);
GPR_ASSERT((assigned_port = grpc_server_add_secure_http2_port(
sf->server, sf->servers_hostport, server_creds)) > 0);
grpc_server_credentials_release(server_creds);
GPR_ASSERT(sf->port == assigned_port);
grpc_server_start(sf->server);
}
static void teardown_server(server_fixture* sf) {
if (!sf->server) return;
gpr_log(GPR_INFO, "Server[%s] shutting down", sf->servers_hostport);
grpc_completion_queue* shutdown_cq =
grpc_completion_queue_create_for_pluck(nullptr);
grpc_server_shutdown_and_notify(sf->server, shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, tag(1000),
grpc_timeout_seconds_to_deadline(5),
nullptr)
.type == GRPC_OP_COMPLETE);
grpc_completion_queue_destroy(shutdown_cq);
grpc_server_destroy(sf->server);
gpr_thd_join(sf->tid);
sf->server = nullptr;
grpc_completion_queue_shutdown(sf->cq);
drain_cq(sf->cq);
grpc_completion_queue_destroy(sf->cq);
gpr_log(GPR_INFO, "Server[%s] bye bye", sf->servers_hostport);
gpr_free(sf->servers_hostport);
}
static void fork_backend_server(void* arg) {
server_fixture* sf = static_cast<server_fixture*>(arg);
start_backend_server(sf);
}
static void fork_lb_server(void* arg) {
test_fixture* tf = static_cast<test_fixture*>(arg);
int ports[NUM_BACKENDS];
for (int i = 0; i < NUM_BACKENDS; i++) {
ports[i] = tf->lb_backends[i].port;
}
start_lb_server(&tf->lb_server, ports, NUM_BACKENDS,
tf->lb_server_update_delay_ms);
}
#define LB_TOKEN_PREFIX "token"
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) {
// Only the first half of the servers expect an LB token.
if (i < NUM_BACKENDS / 2) {
tf.lb_backends[i].lb_token_prefix = LB_TOKEN_PREFIX;
} else {
tf.lb_backends[i].lb_token_prefix = "";
}
setup_server("127.0.0.1", &tf.lb_backends[i]);
gpr_thd_new(&tf.lb_backends[i].tid, "grpclb_backend", fork_backend_server,
&tf.lb_backends[i], &options);
}
tf.lb_server.lb_token_prefix = LB_TOKEN_PREFIX;
tf.lb_server.balancer_name = BALANCERS_NAME;
setup_server("127.0.0.1", &tf.lb_server);
gpr_thd_new(&tf.lb_server.tid, "grpclb_server", fork_lb_server, &tf.lb_server,
&options);
setup_client(&tf.lb_server, tf.lb_backends, &tf.client);
return tf;
}
static void teardown_test_fixture(test_fixture* tf) {
teardown_client(&tf->client);
for (int i = 0; i < NUM_BACKENDS; ++i) {
teardown_server(&tf->lb_backends[i]);
}
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.
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 = setup_test_fixture(lb_server_update_delay_ms);
perform_request(
&tf.client); // "consumes" 1st backend server of 1st serverlist
perform_request(
&tf.client); // "consumes" 2nd backend server of 1st serverlist
perform_request(
&tf.client); // "consumes" 1st backend server of 2nd serverlist
perform_request(
&tf.client); // "consumes" 2nd backend server of 2nd serverlist
teardown_test_fixture(&tf);
gpr_log(GPR_INFO, "end %s(%d)", __func__, lb_server_update_delay_ms);
return tf;
}
TEST(GrpclbTest, Updates) {
grpc::test_fixture tf_result;
// Clients take at least one second to complete a call (the last part of the
// call sleeps for 1 second while verifying the client's completion queue is
// empty), more if the system is under load. Therefore:
//
// If the LB server waits 800ms before sending an update, it will arrive
// before the first client request finishes, skipping the second server from
// batch 1. All subsequent picks will come from the second half of the
// backends, those coming in the LB update.
tf_result = grpc::test_update(800);
GPR_ASSERT(tf_result.lb_backends[0].num_calls_serviced +
tf_result.lb_backends[1].num_calls_serviced ==
1);
GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced +
tf_result.lb_backends[3].num_calls_serviced >
0);
int num_serviced_calls = 0;
for (int i = 0; i < 4; i++) {
num_serviced_calls += tf_result.lb_backends[i].num_calls_serviced;
}
GPR_ASSERT(num_serviced_calls == 4);
// If the LB server waits 2500ms, the update arrives after two calls and three
// picks. The third pick will be the 1st server of the 1st update (RR policy
// going around). The fourth and final pick will come from the second LB
// update. In any case, the total number of serviced calls must again be equal
// to four across all the backends.
tf_result = grpc::test_update(2500);
GPR_ASSERT(tf_result.lb_backends[0].num_calls_serviced +
tf_result.lb_backends[1].num_calls_serviced >=
2);
GPR_ASSERT(tf_result.lb_backends[2].num_calls_serviced +
tf_result.lb_backends[3].num_calls_serviced >
0);
num_serviced_calls = 0;
for (int i = 0; i < 4; i++) {
num_serviced_calls += tf_result.lb_backends[i].num_calls_serviced;
}
GPR_ASSERT(num_serviced_calls == 4);
}
TEST(GrpclbTest, InvalidAddressInServerlist) {}
} // namespace
} // namespace grpc
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
grpc_test_init(argc, argv);
// Make the backup poller poll very frequently in order to pick up
// updates from all the subchannels's FDs.
gpr_setenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS", "1");
grpc_init();
const auto result = RUN_ALL_TESTS();
grpc_shutdown();
return result;
}

@ -3418,29 +3418,6 @@
"third_party": false,
"type": "target"
},
{
"deps": [
"gpr",
"gpr_test_util",
"grpc",
"grpc++",
"grpc++_test_util",
"grpc_test_util"
],
"headers": [
"src/proto/grpc/lb/v1/load_balancer.grpc.pb.h",
"src/proto/grpc/lb/v1/load_balancer.pb.h",
"src/proto/grpc/lb/v1/load_balancer_mock.grpc.pb.h"
],
"is_filegroup": false,
"language": "c++",
"name": "grpclb_test",
"src": [
"test/cpp/grpclb/grpclb_test.cc"
],
"third_party": false,
"type": "target"
},
{
"deps": [
"gpr",

@ -3865,30 +3865,6 @@
],
"uses_polling": true
},
{
"args": [],
"benchmark": false,
"ci_platforms": [
"linux",
"mac",
"posix",
"windows"
],
"cpu_cost": 1.0,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
"gtest": false,
"language": "c++",
"name": "grpclb_test",
"platforms": [
"linux",
"mac",
"posix",
"windows"
],
"uses_polling": true
},
{
"args": [],
"benchmark": false,

Loading…
Cancel
Save