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 <grpc/credentials.h>
20 #include <grpc/event_engine/event_engine.h>
21 #include <grpc/grpc.h>
22 #include <grpc/grpc_security.h>
23 #include <grpc/impl/channel_arg_names.h>
24 #include <grpc/impl/propagation_bits.h>
25 #include <grpc/slice.h>
26 #include <grpc/status.h>
27 #include <grpc/support/sync.h>
28 #include <grpc/support/time.h>
29 #include <stdint.h>
30 #include <string.h>
31
32 #include <algorithm>
33 #include <functional>
34 #include <memory>
35 #include <string>
36 #include <utility>
37 #include <vector>
38
39 #include "absl/log/check.h"
40 #include "absl/log/log.h"
41 #include "absl/status/status.h"
42 #include "absl/status/statusor.h"
43 #include "absl/strings/str_cat.h"
44 #include "absl/strings/string_view.h"
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/iomgr/closure.h"
49 #include "src/core/lib/iomgr/error.h"
50 #include "src/core/lib/iomgr/exec_ctx.h"
51 #include "src/core/lib/iomgr/iomgr_fwd.h"
52 #include "src/core/lib/iomgr/resolve_address.h"
53 #include "src/core/lib/iomgr/resolve_address_impl.h"
54 #include "src/core/lib/iomgr/resolved_address.h"
55 #include "src/core/lib/iomgr/sockaddr.h"
56 #include "src/core/lib/iomgr/socket_utils.h"
57 #include "src/core/resolver/dns/c_ares/grpc_ares_wrapper.h"
58 #include "src/core/resolver/endpoint_addresses.h"
59 #include "src/core/util/debug_location.h"
60 #include "src/core/util/time.h"
61 #include "test/core/end2end/cq_verifier.h"
62 #include "test/core/test_util/port.h"
63 #include "test/core/test_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 LOG(ERROR) << "Skipping iomgr-specific DNS test because EventEngine DNS is "
214 "enabled";
215 return 0;
216 }
217 grpc_completion_queue* cq;
218 grpc_op ops[6];
219 grpc_op* op;
220
221 grpc::testing::TestEnvironment env(&argc, argv);
222
223 gpr_mu_init(&g_mu);
224 grpc_init();
225 grpc_core::ResetDNSResolver(
226 std::make_unique<TestDNSResolver>(grpc_core::GetDNSResolver()));
227 iomgr_dns_lookup_ares = grpc_dns_lookup_hostname_ares;
228 iomgr_cancel_ares_request = grpc_cancel_ares_request;
229 grpc_dns_lookup_hostname_ares = my_dns_lookup_ares;
230 grpc_cancel_ares_request = my_cancel_ares_request;
231
232 int was_cancelled1;
233 int was_cancelled2;
234
235 grpc_metadata_array trailing_metadata_recv1;
236 grpc_metadata_array request_metadata1;
237 grpc_call_details request_details1;
238 grpc_status_code status1;
239 grpc_slice details1;
240 grpc_metadata_array_init(&trailing_metadata_recv1);
241 grpc_metadata_array_init(&request_metadata1);
242 grpc_call_details_init(&request_details1);
243
244 grpc_metadata_array trailing_metadata_recv2;
245 grpc_metadata_array request_metadata2;
246 grpc_call_details request_details2;
247 grpc_status_code status2;
248 grpc_slice details2;
249 grpc_metadata_array_init(&trailing_metadata_recv2);
250 grpc_metadata_array_init(&request_metadata2);
251 grpc_call_details_init(&request_details2);
252
253 cq = grpc_completion_queue_create_for_next(nullptr);
254 grpc_core::CqVerifier cqv(cq);
255
256 // reserve two ports
257 int port1 = grpc_pick_unused_port_or_die();
258 int port2 = grpc_pick_unused_port_or_die();
259
260 std::string addr;
261
262 auto client_args =
263 grpc_core::ChannelArgs()
264 .Set(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, 1000)
265 .Set(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, 1000)
266 .Set(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, 5000)
267 // When this test brings down server1 and then brings up server2,
268 // the targeted server port number changes, and the client channel
269 // needs to re-resolve to pick this up. This test requires that
270 // happen within 10 seconds, but gRPC's DNS resolvers rate limit
271 // resolution attempts to at most once every 30 seconds by default.
272 // So we tweak it for this test.
273 .Set(GRPC_ARG_DNS_MIN_TIME_BETWEEN_RESOLUTIONS_MS, 1000)
274 .ToC();
275
276 // create a channel that picks first amongst the servers
277 grpc_channel_credentials* creds = grpc_insecure_credentials_create();
278 grpc_channel* chan = grpc_channel_create("test", creds, client_args.get());
279 grpc_channel_credentials_release(creds);
280 // and an initial call to them
281 grpc_slice host = grpc_slice_from_static_string("127.0.0.1");
282 grpc_call* call1 =
283 grpc_channel_create_call(chan, nullptr, GRPC_PROPAGATE_DEFAULTS, cq,
284 grpc_slice_from_static_string("/foo"), &host,
285 grpc_timeout_seconds_to_deadline(20), nullptr);
286 // send initial metadata to probe connectivity
287 memset(ops, 0, sizeof(ops));
288 op = ops;
289 op->op = GRPC_OP_SEND_INITIAL_METADATA;
290 op->data.send_initial_metadata.count = 0;
291 op->flags = GRPC_INITIAL_METADATA_WAIT_FOR_READY;
292 op->reserved = nullptr;
293 op++;
294 CHECK_EQ(GRPC_CALL_OK,
295 grpc_call_start_batch(call1, ops, (size_t)(op - ops),
296 grpc_core::CqVerifier::tag(0x101), nullptr));
297 // and receive status to probe termination
298 memset(ops, 0, sizeof(ops));
299 op = ops;
300 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
301 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv1;
302 op->data.recv_status_on_client.status = &status1;
303 op->data.recv_status_on_client.status_details = &details1;
304 op->flags = 0;
305 op->reserved = nullptr;
306 op++;
307 CHECK_EQ(GRPC_CALL_OK,
308 grpc_call_start_batch(call1, ops, (size_t)(op - ops),
309 grpc_core::CqVerifier::tag(0x102), nullptr));
310
311 // bring a server up on the first port
312 grpc_server* server1 = grpc_server_create(nullptr, nullptr);
313 addr = absl::StrCat("127.0.0.1:", port1);
314 grpc_server_credentials* server_creds =
315 grpc_insecure_server_credentials_create();
316 grpc_server_add_http2_port(server1, addr.c_str(), server_creds);
317 grpc_server_credentials_release(server_creds);
318 grpc_server_register_completion_queue(server1, cq, nullptr);
319 grpc_server_start(server1);
320
321 // request a call to the server
322 grpc_call* server_call1;
323 CHECK_EQ(GRPC_CALL_OK,
324 grpc_server_request_call(server1, &server_call1, &request_details1,
325 &request_metadata1, cq, cq,
326 grpc_core::CqVerifier::tag(0x301)));
327
328 set_resolve_port(port1);
329
330 // first call should now start
331 cqv.Expect(grpc_core::CqVerifier::tag(0x101), true);
332 cqv.Expect(grpc_core::CqVerifier::tag(0x301), true);
333 cqv.Verify();
334
335 CHECK(GRPC_CHANNEL_READY == grpc_channel_check_connectivity_state(chan, 0));
336 grpc_channel_watch_connectivity_state(chan, GRPC_CHANNEL_READY,
337 gpr_inf_future(GPR_CLOCK_REALTIME), cq,
338 grpc_core::CqVerifier::tag(0x9999));
339
340 // listen for close on the server call to probe for finishing
341 memset(ops, 0, sizeof(ops));
342 op = ops;
343 op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
344 op->data.recv_close_on_server.cancelled = &was_cancelled1;
345 op->flags = 0;
346 op++;
347 CHECK_EQ(GRPC_CALL_OK,
348 grpc_call_start_batch(server_call1, ops, (size_t)(op - ops),
349 grpc_core::CqVerifier::tag(0x302), nullptr));
350
351 // shutdown first server:
352 // we should see a connectivity change and then nothing
353 set_resolve_port(-1);
354 grpc_server_shutdown_and_notify(server1, cq,
355 grpc_core::CqVerifier::tag(0xdead1));
356 cqv.Expect(grpc_core::CqVerifier::tag(0x9999), true);
357 cqv.Verify();
358 cqv.VerifyEmpty();
359
360 // and a new call: should go through to server2 when we start it
361 grpc_call* call2 =
362 grpc_channel_create_call(chan, nullptr, GRPC_PROPAGATE_DEFAULTS, cq,
363 grpc_slice_from_static_string("/foo"), &host,
364 grpc_timeout_seconds_to_deadline(20), nullptr);
365 // send initial metadata to probe connectivity
366 memset(ops, 0, sizeof(ops));
367 op = ops;
368 op->op = GRPC_OP_SEND_INITIAL_METADATA;
369 op->data.send_initial_metadata.count = 0;
370 op->flags = GRPC_INITIAL_METADATA_WAIT_FOR_READY;
371 op->reserved = nullptr;
372 op++;
373 CHECK_EQ(GRPC_CALL_OK,
374 grpc_call_start_batch(call2, ops, (size_t)(op - ops),
375 grpc_core::CqVerifier::tag(0x201), nullptr));
376 // and receive status to probe termination
377 memset(ops, 0, sizeof(ops));
378 op = ops;
379 op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
380 op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv2;
381 op->data.recv_status_on_client.status = &status2;
382 op->data.recv_status_on_client.status_details = &details2;
383 op->flags = 0;
384 op->reserved = nullptr;
385 op++;
386 CHECK_EQ(GRPC_CALL_OK,
387 grpc_call_start_batch(call2, ops, (size_t)(op - ops),
388 grpc_core::CqVerifier::tag(0x202), nullptr));
389
390 // and bring up second server
391 set_resolve_port(port2);
392 grpc_server* server2 = grpc_server_create(nullptr, nullptr);
393 addr = absl::StrCat("127.0.0.1:", port2);
394 grpc_server_credentials* another_server_creds =
395 grpc_insecure_server_credentials_create();
396 grpc_server_add_http2_port(server2, addr.c_str(), another_server_creds);
397 grpc_server_credentials_release(another_server_creds);
398 grpc_server_register_completion_queue(server2, cq, nullptr);
399 grpc_server_start(server2);
400
401 // request a call to the server
402 grpc_call* server_call2;
403 CHECK_EQ(GRPC_CALL_OK,
404 grpc_server_request_call(server2, &server_call2, &request_details2,
405 &request_metadata2, cq, cq,
406 grpc_core::CqVerifier::tag(0x401)));
407
408 // second call should now start
409 cqv.Expect(grpc_core::CqVerifier::tag(0x201), true);
410 cqv.Expect(grpc_core::CqVerifier::tag(0x401), true);
411 cqv.Verify();
412
413 // listen for close on the server call to probe for finishing
414 memset(ops, 0, sizeof(ops));
415 op = ops;
416 op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
417 op->data.recv_close_on_server.cancelled = &was_cancelled2;
418 op->flags = 0;
419 op++;
420 CHECK_EQ(GRPC_CALL_OK,
421 grpc_call_start_batch(server_call2, ops, (size_t)(op - ops),
422 grpc_core::CqVerifier::tag(0x402), nullptr));
423
424 // shutdown second server: we should see nothing
425 grpc_server_shutdown_and_notify(server2, cq,
426 grpc_core::CqVerifier::tag(0xdead2));
427 cqv.VerifyEmpty();
428
429 grpc_call_cancel(call1, nullptr);
430 grpc_call_cancel(call2, nullptr);
431
432 // now everything else should finish
433 cqv.Expect(grpc_core::CqVerifier::tag(0x102), true);
434 cqv.Expect(grpc_core::CqVerifier::tag(0x202), true);
435 cqv.Expect(grpc_core::CqVerifier::tag(0x302), true);
436 cqv.Expect(grpc_core::CqVerifier::tag(0x402), true);
437 cqv.Expect(grpc_core::CqVerifier::tag(0xdead1), true);
438 cqv.Expect(grpc_core::CqVerifier::tag(0xdead2), true);
439 cqv.Verify();
440
441 grpc_call_unref(call1);
442 grpc_call_unref(call2);
443 grpc_call_unref(server_call1);
444 grpc_call_unref(server_call2);
445 grpc_server_destroy(server1);
446 grpc_server_destroy(server2);
447 grpc_channel_destroy(chan);
448
449 grpc_metadata_array_destroy(&trailing_metadata_recv1);
450 grpc_metadata_array_destroy(&request_metadata1);
451 grpc_call_details_destroy(&request_details1);
452 grpc_slice_unref(details1);
453 grpc_metadata_array_destroy(&trailing_metadata_recv2);
454 grpc_metadata_array_destroy(&request_metadata2);
455 grpc_call_details_destroy(&request_details2);
456 grpc_slice_unref(details2);
457
458 grpc_completion_queue_destroy(cq);
459
460 grpc_shutdown();
461 gpr_mu_destroy(&g_mu);
462
463 return 0;
464 }
465