1 //
2 //
3 // Copyright 2016 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18
19 #include <stdint.h>
20 #include <string.h>
21
22 #include <algorithm>
23 #include <functional>
24 #include <memory>
25 #include <string>
26 #include <utility>
27 #include <vector>
28
29 #include "absl/status/status.h"
30 #include "absl/status/statusor.h"
31 #include "absl/strings/str_cat.h"
32 #include "absl/strings/string_view.h"
33
34 #include <grpc/event_engine/event_engine.h>
35 #include <grpc/grpc.h>
36 #include <grpc/grpc_security.h>
37 #include <grpc/impl/channel_arg_names.h>
38 #include <grpc/impl/propagation_bits.h>
39 #include <grpc/slice.h>
40 #include <grpc/status.h>
41 #include <grpc/support/log.h>
42 #include <grpc/support/sync.h>
43 #include <grpc/support/time.h>
44
45 #include "src/core/lib/channel/channel_args.h"
46 #include "src/core/lib/event_engine/default_event_engine.h"
47 #include "src/core/lib/experiments/experiments.h"
48 #include "src/core/lib/gprpp/debug_location.h"
49 #include "src/core/lib/gprpp/time.h"
50 #include "src/core/lib/iomgr/closure.h"
51 #include "src/core/lib/iomgr/error.h"
52 #include "src/core/lib/iomgr/exec_ctx.h"
53 #include "src/core/lib/iomgr/iomgr_fwd.h"
54 #include "src/core/lib/iomgr/resolve_address.h"
55 #include "src/core/lib/iomgr/resolve_address_impl.h"
56 #include "src/core/lib/iomgr/resolved_address.h"
57 #include "src/core/lib/iomgr/sockaddr.h"
58 #include "src/core/lib/iomgr/socket_utils.h"
59 #include "src/core/resolver/dns/c_ares/grpc_ares_wrapper.h"
60 #include "src/core/resolver/endpoint_addresses.h"
61 #include "test/core/end2end/cq_verifier.h"
62 #include "test/core/util/port.h"
63 #include "test/core/util/test_config.h"
64
65 static gpr_mu g_mu;
66 static int g_resolve_port = -1;
67
68 static grpc_ares_request* (*iomgr_dns_lookup_ares)(
69 const char* dns_server, const char* addr, const char* default_port,
70 grpc_pollset_set* interested_parties, grpc_closure* on_done,
71 std::unique_ptr<grpc_core::EndpointAddressesList>* addresses,
72 int query_timeout_ms);
73
74 static void (*iomgr_cancel_ares_request)(grpc_ares_request* request);
75
set_resolve_port(int port)76 static void set_resolve_port(int port) {
77 gpr_mu_lock(&g_mu);
78 g_resolve_port = port;
79 gpr_mu_unlock(&g_mu);
80 }
81
82 namespace {
83
84 class TestDNSResolver : public grpc_core::DNSResolver {
85 public:
TestDNSResolver(std::shared_ptr<grpc_core::DNSResolver> default_resolver)86 explicit TestDNSResolver(
87 std::shared_ptr<grpc_core::DNSResolver> default_resolver)
88 : default_resolver_(std::move(default_resolver)),
89 engine_(grpc_event_engine::experimental::GetDefaultEventEngine()) {}
LookupHostname(std::function<void (absl::StatusOr<std::vector<grpc_resolved_address>>)> on_resolved,absl::string_view name,absl::string_view default_port,grpc_core::Duration timeout,grpc_pollset_set * interested_parties,absl::string_view name_server)90 TaskHandle LookupHostname(
91 std::function<void(absl::StatusOr<std::vector<grpc_resolved_address>>)>
92 on_resolved,
93 absl::string_view name, absl::string_view default_port,
94 grpc_core::Duration timeout, grpc_pollset_set* interested_parties,
95 absl::string_view name_server) override {
96 if (name != "test") {
97 return default_resolver_->LookupHostname(std::move(on_resolved), name,
98 default_port, timeout,
99 interested_parties, name_server);
100 }
101 MakeDNSRequest(std::move(on_resolved));
102 return kNullHandle;
103 }
104
LookupHostnameBlocking(absl::string_view name,absl::string_view default_port)105 absl::StatusOr<std::vector<grpc_resolved_address>> LookupHostnameBlocking(
106 absl::string_view name, absl::string_view default_port) override {
107 return default_resolver_->LookupHostnameBlocking(name, default_port);
108 }
109
LookupSRV(std::function<void (absl::StatusOr<std::vector<grpc_resolved_address>>)> on_resolved,absl::string_view,grpc_core::Duration,grpc_pollset_set *,absl::string_view)110 TaskHandle LookupSRV(
111 std::function<void(absl::StatusOr<std::vector<grpc_resolved_address>>)>
112 on_resolved,
113 absl::string_view /* name */, grpc_core::Duration /* timeout */,
114 grpc_pollset_set* /* interested_parties */,
115 absl::string_view /* name_server */) override {
116 engine_->Run([on_resolved] {
117 grpc_core::ApplicationCallbackExecCtx app_exec_ctx;
118 grpc_core::ExecCtx exec_ctx;
119 on_resolved(absl::UnimplementedError(
120 "The Testing DNS resolver does not support looking up SRV records"));
121 });
122 return {-1, -1};
123 };
124
LookupTXT(std::function<void (absl::StatusOr<std::string>)> on_resolved,absl::string_view,grpc_core::Duration,grpc_pollset_set *,absl::string_view)125 TaskHandle LookupTXT(
126 std::function<void(absl::StatusOr<std::string>)> on_resolved,
127 absl::string_view /* name */, grpc_core::Duration /* timeout */,
128 grpc_pollset_set* /* interested_parties */,
129 absl::string_view /* name_server */) override {
130 // Not supported
131 engine_->Run([on_resolved] {
132 grpc_core::ApplicationCallbackExecCtx app_exec_ctx;
133 grpc_core::ExecCtx exec_ctx;
134 on_resolved(absl::UnimplementedError(
135 "The Testing DNS resolver does not support looking up TXT records"));
136 });
137 return {-1, -1};
138 };
139
Cancel(TaskHandle)140 bool Cancel(TaskHandle /*handle*/) override { return false; }
141
142 private:
MakeDNSRequest(std::function<void (absl::StatusOr<std::vector<grpc_resolved_address>>)> on_done)143 void MakeDNSRequest(
144 std::function<void(absl::StatusOr<std::vector<grpc_resolved_address>>)>
145 on_done) {
146 gpr_mu_lock(&g_mu);
147 if (g_resolve_port < 0) {
148 gpr_mu_unlock(&g_mu);
149 new grpc_core::DNSCallbackExecCtxScheduler(
150 std::move(on_done), absl::UnknownError("Forced Failure"));
151 } else {
152 std::vector<grpc_resolved_address> addrs;
153 grpc_resolved_address addr;
154 grpc_sockaddr_in* sa = reinterpret_cast<grpc_sockaddr_in*>(&addr);
155 sa->sin_family = GRPC_AF_INET;
156 sa->sin_addr.s_addr = 0x100007f;
157 sa->sin_port = grpc_htons(static_cast<uint16_t>(g_resolve_port));
158 addr.len = static_cast<socklen_t>(sizeof(*sa));
159 addrs.push_back(addr);
160 gpr_mu_unlock(&g_mu);
161 new grpc_core::DNSCallbackExecCtxScheduler(std::move(on_done),
162 std::move(addrs));
163 }
164 }
165 std::shared_ptr<grpc_core::DNSResolver> default_resolver_;
166 std::shared_ptr<grpc_event_engine::experimental::EventEngine> engine_;
167 };
168
169 } // namespace
170
my_dns_lookup_ares(const char * dns_server,const char * addr,const char * default_port,grpc_pollset_set * interested_parties,grpc_closure * on_done,std::unique_ptr<grpc_core::EndpointAddressesList> * addresses,int query_timeout_ms)171 static grpc_ares_request* my_dns_lookup_ares(
172 const char* dns_server, const char* addr, const char* default_port,
173 grpc_pollset_set* interested_parties, grpc_closure* on_done,
174 std::unique_ptr<grpc_core::EndpointAddressesList>* addresses,
175 int query_timeout_ms) {
176 if (0 != strcmp(addr, "test")) {
177 // A records should suffice
178 return iomgr_dns_lookup_ares(dns_server, addr, default_port,
179 interested_parties, on_done, addresses,
180 query_timeout_ms);
181 }
182
183 grpc_error_handle error;
184 gpr_mu_lock(&g_mu);
185 if (g_resolve_port < 0) {
186 gpr_mu_unlock(&g_mu);
187 error = GRPC_ERROR_CREATE("Forced Failure");
188 } else {
189 *addresses = std::make_unique<grpc_core::EndpointAddressesList>();
190 grpc_resolved_address address;
191 memset(&address, 0, sizeof(address));
192 auto* sa = reinterpret_cast<grpc_sockaddr_in*>(&address.addr);
193 sa->sin_family = GRPC_AF_INET;
194 sa->sin_addr.s_addr = 0x100007f;
195 sa->sin_port = grpc_htons(static_cast<uint16_t>(g_resolve_port));
196 address.len = sizeof(grpc_sockaddr_in);
197 (*addresses)->emplace_back(address, grpc_core::ChannelArgs());
198 gpr_mu_unlock(&g_mu);
199 }
200 grpc_core::ExecCtx::Run(DEBUG_LOCATION, on_done, error);
201 return nullptr;
202 }
203
my_cancel_ares_request(grpc_ares_request * request)204 static void my_cancel_ares_request(grpc_ares_request* request) {
205 if (request != nullptr) {
206 iomgr_cancel_ares_request(request);
207 }
208 }
209
main(int argc,char ** argv)210 int main(int argc, char** argv) {
211 // TODO(yijiem): rewrite this test with a custom EventEngine DNS Resolver
212 if (grpc_core::IsEventEngineDnsEnabled()) {
213 gpr_log(
214 GPR_ERROR,
215 "Skipping iomgr-specific DNS test because EventEngine DNS is enabled");
216 return 0;
217 }
218 grpc_completion_queue* cq;
219 grpc_op ops[6];
220 grpc_op* op;
221
222 grpc::testing::TestEnvironment env(&argc, argv);
223
224 gpr_mu_init(&g_mu);
225 grpc_init();
226 grpc_core::ResetDNSResolver(
227 std::make_unique<TestDNSResolver>(grpc_core::GetDNSResolver()));
228 iomgr_dns_lookup_ares = grpc_dns_lookup_hostname_ares;
229 iomgr_cancel_ares_request = grpc_cancel_ares_request;
230 grpc_dns_lookup_hostname_ares = my_dns_lookup_ares;
231 grpc_cancel_ares_request = my_cancel_ares_request;
232
233 int was_cancelled1;
234 int was_cancelled2;
235
236 grpc_metadata_array trailing_metadata_recv1;
237 grpc_metadata_array request_metadata1;
238 grpc_call_details request_details1;
239 grpc_status_code status1;
240 grpc_slice details1;
241 grpc_metadata_array_init(&trailing_metadata_recv1);
242 grpc_metadata_array_init(&request_metadata1);
243 grpc_call_details_init(&request_details1);
244
245 grpc_metadata_array trailing_metadata_recv2;
246 grpc_metadata_array request_metadata2;
247 grpc_call_details request_details2;
248 grpc_status_code status2;
249 grpc_slice details2;
250 grpc_metadata_array_init(&trailing_metadata_recv2);
251 grpc_metadata_array_init(&request_metadata2);
252 grpc_call_details_init(&request_details2);
253
254 cq = grpc_completion_queue_create_for_next(nullptr);
255 grpc_core::CqVerifier cqv(cq);
256
257 // reserve two ports
258 int port1 = grpc_pick_unused_port_or_die();
259 int port2 = grpc_pick_unused_port_or_die();
260
261 std::string addr;
262
263 auto client_args =
264 grpc_core::ChannelArgs()
265 .Set(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, 1000)
266 .Set(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, 1000)
267 .Set(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, 5000)
268 // When this test brings down server1 and then brings up server2,
269 // the targetted server port number changes, and the client channel
270 // needs to re-resolve to pick this up. This test requires that
271 // happen within 10 seconds, but gRPC's DNS resolvers rate limit
272 // resolution attempts to at most once every 30 seconds by default.
273 // So we tweak it for this test.
274 .Set(GRPC_ARG_DNS_MIN_TIME_BETWEEN_RESOLUTIONS_MS, 1000)
275 .ToC();
276
277 // create a channel that picks first amongst the servers
278 grpc_channel_credentials* creds = grpc_insecure_credentials_create();
279 grpc_channel* chan = grpc_channel_create("test", creds, client_args.get());
280 grpc_channel_credentials_release(creds);
281 // and an initial call to them
282 grpc_slice host = grpc_slice_from_static_string("127.0.0.1");
283 grpc_call* call1 =
284 grpc_channel_create_call(chan, nullptr, GRPC_PROPAGATE_DEFAULTS, cq,
285 grpc_slice_from_static_string("/foo"), &host,
286 grpc_timeout_seconds_to_deadline(20), nullptr);
287 // send initial metadata to probe connectivity
288 memset(ops, 0, sizeof(ops));
289 op = ops;
290 op->op = GRPC_OP_SEND_INITIAL_METADATA;
291 op->data.send_initial_metadata.count = 0;
292 op->flags = GRPC_INITIAL_METADATA_WAIT_FOR_READY;
293 op->reserved = nullptr;
294 op++;
295 GPR_ASSERT(GRPC_CALL_OK ==
296 grpc_call_start_batch(call1, ops, (size_t)(op - ops),
297 grpc_core::CqVerifier::tag(0x101), nullptr));
298 // and receive status to probe termination
299 memset(ops, 0, sizeof(ops));
300 op = ops;
301 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
302 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv1;
303 op->data.recv_status_on_client.status = &status1;
304 op->data.recv_status_on_client.status_details = &details1;
305 op->flags = 0;
306 op->reserved = nullptr;
307 op++;
308 GPR_ASSERT(GRPC_CALL_OK ==
309 grpc_call_start_batch(call1, ops, (size_t)(op - ops),
310 grpc_core::CqVerifier::tag(0x102), nullptr));
311
312 // bring a server up on the first port
313 grpc_server* server1 = grpc_server_create(nullptr, nullptr);
314 addr = absl::StrCat("127.0.0.1:", port1);
315 grpc_server_credentials* server_creds =
316 grpc_insecure_server_credentials_create();
317 grpc_server_add_http2_port(server1, addr.c_str(), server_creds);
318 grpc_server_credentials_release(server_creds);
319 grpc_server_register_completion_queue(server1, cq, nullptr);
320 grpc_server_start(server1);
321
322 // request a call to the server
323 grpc_call* server_call1;
324 GPR_ASSERT(GRPC_CALL_OK ==
325 grpc_server_request_call(server1, &server_call1, &request_details1,
326 &request_metadata1, cq, cq,
327 grpc_core::CqVerifier::tag(0x301)));
328
329 set_resolve_port(port1);
330
331 // first call should now start
332 cqv.Expect(grpc_core::CqVerifier::tag(0x101), true);
333 cqv.Expect(grpc_core::CqVerifier::tag(0x301), true);
334 cqv.Verify();
335
336 GPR_ASSERT(GRPC_CHANNEL_READY ==
337 grpc_channel_check_connectivity_state(chan, 0));
338 grpc_channel_watch_connectivity_state(chan, GRPC_CHANNEL_READY,
339 gpr_inf_future(GPR_CLOCK_REALTIME), cq,
340 grpc_core::CqVerifier::tag(0x9999));
341
342 // listen for close on the server call to probe for finishing
343 memset(ops, 0, sizeof(ops));
344 op = ops;
345 op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
346 op->data.recv_close_on_server.cancelled = &was_cancelled1;
347 op->flags = 0;
348 op++;
349 GPR_ASSERT(GRPC_CALL_OK ==
350 grpc_call_start_batch(server_call1, ops, (size_t)(op - ops),
351 grpc_core::CqVerifier::tag(0x302), nullptr));
352
353 // shutdown first server:
354 // we should see a connectivity change and then nothing
355 set_resolve_port(-1);
356 grpc_server_shutdown_and_notify(server1, cq,
357 grpc_core::CqVerifier::tag(0xdead1));
358 cqv.Expect(grpc_core::CqVerifier::tag(0x9999), true);
359 cqv.Verify();
360 cqv.VerifyEmpty();
361
362 // and a new call: should go through to server2 when we start it
363 grpc_call* call2 =
364 grpc_channel_create_call(chan, nullptr, GRPC_PROPAGATE_DEFAULTS, cq,
365 grpc_slice_from_static_string("/foo"), &host,
366 grpc_timeout_seconds_to_deadline(20), nullptr);
367 // send initial metadata to probe connectivity
368 memset(ops, 0, sizeof(ops));
369 op = ops;
370 op->op = GRPC_OP_SEND_INITIAL_METADATA;
371 op->data.send_initial_metadata.count = 0;
372 op->flags = GRPC_INITIAL_METADATA_WAIT_FOR_READY;
373 op->reserved = nullptr;
374 op++;
375 GPR_ASSERT(GRPC_CALL_OK ==
376 grpc_call_start_batch(call2, ops, (size_t)(op - ops),
377 grpc_core::CqVerifier::tag(0x201), nullptr));
378 // and receive status to probe termination
379 memset(ops, 0, sizeof(ops));
380 op = ops;
381 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
382 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv2;
383 op->data.recv_status_on_client.status = &status2;
384 op->data.recv_status_on_client.status_details = &details2;
385 op->flags = 0;
386 op->reserved = nullptr;
387 op++;
388 GPR_ASSERT(GRPC_CALL_OK ==
389 grpc_call_start_batch(call2, ops, (size_t)(op - ops),
390 grpc_core::CqVerifier::tag(0x202), nullptr));
391
392 // and bring up second server
393 set_resolve_port(port2);
394 grpc_server* server2 = grpc_server_create(nullptr, nullptr);
395 addr = absl::StrCat("127.0.0.1:", port2);
396 grpc_server_credentials* another_server_creds =
397 grpc_insecure_server_credentials_create();
398 grpc_server_add_http2_port(server2, addr.c_str(), another_server_creds);
399 grpc_server_credentials_release(another_server_creds);
400 grpc_server_register_completion_queue(server2, cq, nullptr);
401 grpc_server_start(server2);
402
403 // request a call to the server
404 grpc_call* server_call2;
405 GPR_ASSERT(GRPC_CALL_OK ==
406 grpc_server_request_call(server2, &server_call2, &request_details2,
407 &request_metadata2, cq, cq,
408 grpc_core::CqVerifier::tag(0x401)));
409
410 // second call should now start
411 cqv.Expect(grpc_core::CqVerifier::tag(0x201), true);
412 cqv.Expect(grpc_core::CqVerifier::tag(0x401), true);
413 cqv.Verify();
414
415 // listen for close on the server call to probe for finishing
416 memset(ops, 0, sizeof(ops));
417 op = ops;
418 op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
419 op->data.recv_close_on_server.cancelled = &was_cancelled2;
420 op->flags = 0;
421 op++;
422 GPR_ASSERT(GRPC_CALL_OK ==
423 grpc_call_start_batch(server_call2, ops, (size_t)(op - ops),
424 grpc_core::CqVerifier::tag(0x402), nullptr));
425
426 // shutdown second server: we should see nothing
427 grpc_server_shutdown_and_notify(server2, cq,
428 grpc_core::CqVerifier::tag(0xdead2));
429 cqv.VerifyEmpty();
430
431 grpc_call_cancel(call1, nullptr);
432 grpc_call_cancel(call2, nullptr);
433
434 // now everything else should finish
435 cqv.Expect(grpc_core::CqVerifier::tag(0x102), true);
436 cqv.Expect(grpc_core::CqVerifier::tag(0x202), true);
437 cqv.Expect(grpc_core::CqVerifier::tag(0x302), true);
438 cqv.Expect(grpc_core::CqVerifier::tag(0x402), true);
439 cqv.Expect(grpc_core::CqVerifier::tag(0xdead1), true);
440 cqv.Expect(grpc_core::CqVerifier::tag(0xdead2), true);
441 cqv.Verify();
442
443 grpc_call_unref(call1);
444 grpc_call_unref(call2);
445 grpc_call_unref(server_call1);
446 grpc_call_unref(server_call2);
447 grpc_server_destroy(server1);
448 grpc_server_destroy(server2);
449 grpc_channel_destroy(chan);
450
451 grpc_metadata_array_destroy(&trailing_metadata_recv1);
452 grpc_metadata_array_destroy(&request_metadata1);
453 grpc_call_details_destroy(&request_details1);
454 grpc_slice_unref(details1);
455 grpc_metadata_array_destroy(&trailing_metadata_recv2);
456 grpc_metadata_array_destroy(&request_metadata2);
457 grpc_call_details_destroy(&request_details2);
458 grpc_slice_unref(details2);
459
460 grpc_completion_queue_destroy(cq);
461
462 grpc_shutdown();
463 gpr_mu_destroy(&g_mu);
464
465 return 0;
466 }
467