| /* |
| * |
| * 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/thd.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/tmpfile.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, (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, (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, (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, (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, (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, (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, (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, (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, (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, (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, (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); |
| |
| grpc_fake_resolver_response_generator* response_generator = |
| grpc_fake_resolver_response_generator_create(); |
| |
| 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); |
| |
| const grpc_arg new_args[] = { |
| grpc_fake_transport_expected_targets_arg(expected_target_names), |
| grpc_fake_resolver_response_generator_arg(response_generator)}; |
| |
| grpc_channel_args* args = grpc_channel_args_copy_and_add( |
| nullptr, new_args, GPR_ARRAY_SIZE(new_args)); |
| gpr_free(expected_target_names); |
| |
| 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); |
| grpc_fake_resolver_response_generator_set_response(response_generator, |
| fake_result); |
| grpc_channel_args_destroy(fake_result); |
| grpc_channel_credentials_unref(fake_creds); |
| grpc_channel_args_destroy(args); |
| grpc_fake_resolver_response_generator_unref(response_generator); |
| } |
| |
| 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; |
| } |