Line data Source code
1 : /*
2 : *
3 : * Copyright 2015, Google Inc.
4 : * All rights reserved.
5 : *
6 : * Redistribution and use in source and binary forms, with or without
7 : * modification, are permitted provided that the following conditions are
8 : * met:
9 : *
10 : * * Redistributions of source code must retain the above copyright
11 : * notice, this list of conditions and the following disclaimer.
12 : * * Redistributions in binary form must reproduce the above
13 : * copyright notice, this list of conditions and the following disclaimer
14 : * in the documentation and/or other materials provided with the
15 : * distribution.
16 : * * Neither the name of Google Inc. nor the names of its
17 : * contributors may be used to endorse or promote products derived from
18 : * this software without specific prior written permission.
19 : *
20 : * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 : * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 : * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 : * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 : * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 : * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 : * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 : * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 : * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 : * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 : * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 : *
32 : */
33 :
34 : #include <grpc/support/port_platform.h>
35 :
36 : #include "src/core/client_config/resolvers/sockaddr_resolver.h"
37 :
38 : #include <stdio.h>
39 : #include <string.h>
40 : #ifdef GPR_POSIX_SOCKET
41 : #include <sys/un.h>
42 : #endif
43 :
44 : #include <grpc/support/alloc.h>
45 : #include <grpc/support/host_port.h>
46 : #include <grpc/support/string_util.h>
47 :
48 : #include "src/core/client_config/lb_policy_registry.h"
49 : #include "src/core/iomgr/resolve_address.h"
50 : #include "src/core/support/string.h"
51 :
52 : typedef struct {
53 : /** base class: must be first */
54 : grpc_resolver base;
55 : /** refcount */
56 : gpr_refcount refs;
57 : /** subchannel factory */
58 : grpc_subchannel_factory *subchannel_factory;
59 : /** load balancing policy name */
60 : char *lb_policy_name;
61 :
62 : /** the addresses that we've 'resolved' */
63 : struct sockaddr_storage *addrs;
64 : /** the corresponding length of the addresses */
65 : size_t *addrs_len;
66 : /** how many elements in \a addrs */
67 : size_t num_addrs;
68 :
69 : /** mutex guarding the rest of the state */
70 : gpr_mu mu;
71 : /** have we published? */
72 : int published;
73 : /** pending next completion, or NULL */
74 : grpc_closure *next_completion;
75 : /** target config address for next completion */
76 : grpc_client_config **target_config;
77 : } sockaddr_resolver;
78 :
79 : static void sockaddr_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
80 :
81 : static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
82 : sockaddr_resolver *r);
83 :
84 : static void sockaddr_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
85 : static void sockaddr_channel_saw_error(grpc_exec_ctx *exec_ctx,
86 : grpc_resolver *r,
87 : struct sockaddr *failing_address,
88 : int failing_address_len);
89 : static void sockaddr_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
90 : grpc_client_config **target_config,
91 : grpc_closure *on_complete);
92 :
93 : static const grpc_resolver_vtable sockaddr_resolver_vtable = {
94 : sockaddr_destroy, sockaddr_shutdown, sockaddr_channel_saw_error,
95 : sockaddr_next};
96 :
97 343 : static void sockaddr_shutdown(grpc_exec_ctx *exec_ctx,
98 : grpc_resolver *resolver) {
99 343 : sockaddr_resolver *r = (sockaddr_resolver *)resolver;
100 343 : gpr_mu_lock(&r->mu);
101 343 : if (r->next_completion != NULL) {
102 279 : *r->target_config = NULL;
103 279 : grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
104 279 : r->next_completion = NULL;
105 : }
106 343 : gpr_mu_unlock(&r->mu);
107 343 : }
108 :
109 0 : static void sockaddr_channel_saw_error(grpc_exec_ctx *exec_ctx,
110 : grpc_resolver *resolver,
111 0 : struct sockaddr *sa, int len) {}
112 :
113 558 : static void sockaddr_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
114 : grpc_client_config **target_config,
115 : grpc_closure *on_complete) {
116 558 : sockaddr_resolver *r = (sockaddr_resolver *)resolver;
117 558 : gpr_mu_lock(&r->mu);
118 558 : GPR_ASSERT(!r->next_completion);
119 558 : r->next_completion = on_complete;
120 558 : r->target_config = target_config;
121 558 : sockaddr_maybe_finish_next_locked(exec_ctx, r);
122 558 : gpr_mu_unlock(&r->mu);
123 558 : }
124 :
125 558 : static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
126 : sockaddr_resolver *r) {
127 : grpc_client_config *cfg;
128 : grpc_lb_policy *lb_policy;
129 : grpc_lb_policy_args lb_policy_args;
130 : grpc_subchannel **subchannels;
131 : grpc_subchannel_args args;
132 :
133 558 : if (r->next_completion != NULL && !r->published) {
134 : size_t i;
135 279 : cfg = grpc_client_config_create();
136 279 : subchannels = gpr_malloc(sizeof(grpc_subchannel *) * r->num_addrs);
137 579 : for (i = 0; i < r->num_addrs; i++) {
138 300 : memset(&args, 0, sizeof(args));
139 300 : args.addr = (struct sockaddr *)&r->addrs[i];
140 300 : args.addr_len = r->addrs_len[i];
141 300 : subchannels[i] = grpc_subchannel_factory_create_subchannel(
142 : exec_ctx, r->subchannel_factory, &args);
143 : }
144 279 : memset(&lb_policy_args, 0, sizeof(lb_policy_args));
145 279 : lb_policy_args.subchannels = subchannels;
146 279 : lb_policy_args.num_subchannels = r->num_addrs;
147 279 : lb_policy = grpc_lb_policy_create(r->lb_policy_name, &lb_policy_args);
148 279 : gpr_free(subchannels);
149 279 : grpc_client_config_set_lb_policy(cfg, lb_policy);
150 279 : GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "sockaddr");
151 279 : r->published = 1;
152 279 : *r->target_config = cfg;
153 279 : grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
154 279 : r->next_completion = NULL;
155 : }
156 558 : }
157 :
158 343 : static void sockaddr_destroy(grpc_exec_ctx *exec_ctx, grpc_resolver *gr) {
159 343 : sockaddr_resolver *r = (sockaddr_resolver *)gr;
160 343 : gpr_mu_destroy(&r->mu);
161 343 : grpc_subchannel_factory_unref(exec_ctx, r->subchannel_factory);
162 343 : gpr_free(r->addrs);
163 343 : gpr_free(r->addrs_len);
164 343 : gpr_free(r->lb_policy_name);
165 343 : gpr_free(r);
166 343 : }
167 :
168 : #ifdef GPR_POSIX_SOCKET
169 316 : static int parse_unix(grpc_uri *uri, struct sockaddr_storage *addr,
170 : size_t *len) {
171 316 : struct sockaddr_un *un = (struct sockaddr_un *)addr;
172 :
173 316 : un->sun_family = AF_UNIX;
174 316 : strcpy(un->sun_path, uri->path);
175 316 : *len = strlen(un->sun_path) + sizeof(un->sun_family) + 1;
176 :
177 316 : return 1;
178 : }
179 :
180 316 : static char *unix_get_default_authority(grpc_resolver_factory *factory,
181 : grpc_uri *uri) {
182 316 : return gpr_strdup("localhost");
183 : }
184 : #endif
185 :
186 27 : static char *ip_get_default_authority(grpc_uri *uri) {
187 27 : const char *path = uri->path;
188 27 : if (path[0] == '/') ++path;
189 27 : return gpr_strdup(path);
190 : }
191 :
192 15 : static char *ipv4_get_default_authority(grpc_resolver_factory *factory,
193 : grpc_uri *uri) {
194 15 : return ip_get_default_authority(uri);
195 : }
196 :
197 12 : static char *ipv6_get_default_authority(grpc_resolver_factory *factory,
198 : grpc_uri *uri) {
199 12 : return ip_get_default_authority(uri);
200 : }
201 :
202 34 : static int parse_ipv4(grpc_uri *uri, struct sockaddr_storage *addr,
203 : size_t *len) {
204 34 : const char *host_port = uri->path;
205 : char *host;
206 : char *port;
207 : int port_num;
208 34 : int result = 0;
209 34 : struct sockaddr_in *in = (struct sockaddr_in *)addr;
210 :
211 34 : if (*host_port == '/') ++host_port;
212 34 : if (!gpr_split_host_port(host_port, &host, &port)) {
213 0 : return 0;
214 : }
215 :
216 34 : memset(in, 0, sizeof(*in));
217 34 : *len = sizeof(*in);
218 34 : in->sin_family = AF_INET;
219 34 : if (inet_pton(AF_INET, host, &in->sin_addr) == 0) {
220 0 : gpr_log(GPR_ERROR, "invalid ipv4 address: '%s'", host);
221 0 : goto done;
222 : }
223 :
224 34 : if (port != NULL) {
225 68 : if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 ||
226 34 : port_num > 65535) {
227 0 : gpr_log(GPR_ERROR, "invalid ipv4 port: '%s'", port);
228 0 : goto done;
229 : }
230 34 : in->sin_port = htons((gpr_uint16)port_num);
231 : } else {
232 0 : gpr_log(GPR_ERROR, "no port given for ipv4 scheme");
233 0 : goto done;
234 : }
235 :
236 34 : result = 1;
237 : done:
238 34 : gpr_free(host);
239 34 : gpr_free(port);
240 34 : return result;
241 : }
242 :
243 14 : static int parse_ipv6(grpc_uri *uri, struct sockaddr_storage *addr,
244 : size_t *len) {
245 14 : const char *host_port = uri->path;
246 : char *host;
247 : char *port;
248 : int port_num;
249 14 : int result = 0;
250 14 : struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)addr;
251 :
252 14 : if (*host_port == '/') ++host_port;
253 14 : if (!gpr_split_host_port(host_port, &host, &port)) {
254 0 : return 0;
255 : }
256 :
257 14 : memset(in6, 0, sizeof(*in6));
258 14 : *len = sizeof(*in6);
259 14 : in6->sin6_family = AF_INET6;
260 14 : if (inet_pton(AF_INET6, host, &in6->sin6_addr) == 0) {
261 0 : gpr_log(GPR_ERROR, "invalid ipv6 address: '%s'", host);
262 0 : goto done;
263 : }
264 :
265 14 : if (port != NULL) {
266 28 : if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 ||
267 14 : port_num > 65535) {
268 0 : gpr_log(GPR_ERROR, "invalid ipv6 port: '%s'", port);
269 0 : goto done;
270 : }
271 14 : in6->sin6_port = htons((gpr_uint16)port_num);
272 : } else {
273 0 : gpr_log(GPR_ERROR, "no port given for ipv6 scheme");
274 0 : goto done;
275 : }
276 :
277 14 : result = 1;
278 : done:
279 14 : gpr_free(host);
280 14 : gpr_free(port);
281 14 : return result;
282 : }
283 :
284 348 : static void do_nothing(void *ignored) {}
285 :
286 343 : static grpc_resolver *sockaddr_create(
287 : grpc_resolver_args *args, const char *default_lb_policy_name,
288 : int parse(grpc_uri *uri, struct sockaddr_storage *dst, size_t *len)) {
289 : size_t i;
290 343 : int errors_found = 0; /* GPR_FALSE */
291 : sockaddr_resolver *r;
292 : gpr_slice path_slice;
293 : gpr_slice_buffer path_parts;
294 :
295 343 : if (0 != strcmp(args->uri->authority, "")) {
296 0 : gpr_log(GPR_ERROR, "authority based uri's not supported by the %s scheme",
297 0 : args->uri->scheme);
298 0 : return NULL;
299 : }
300 :
301 343 : r = gpr_malloc(sizeof(sockaddr_resolver));
302 343 : memset(r, 0, sizeof(*r));
303 :
304 343 : r->lb_policy_name = NULL;
305 343 : if (0 != strcmp(args->uri->query, "")) {
306 : gpr_slice query_slice;
307 : gpr_slice_buffer query_parts;
308 :
309 5 : query_slice =
310 5 : gpr_slice_new(args->uri->query, strlen(args->uri->query), do_nothing);
311 5 : gpr_slice_buffer_init(&query_parts);
312 5 : gpr_slice_split(query_slice, "=", &query_parts);
313 5 : GPR_ASSERT(query_parts.count == 2);
314 5 : if (0 == gpr_slice_str_cmp(query_parts.slices[0], "lb_policy")) {
315 5 : r->lb_policy_name = gpr_dump_slice(query_parts.slices[1], GPR_DUMP_ASCII);
316 : }
317 5 : gpr_slice_buffer_destroy(&query_parts);
318 5 : gpr_slice_unref(query_slice);
319 : }
320 343 : if (r->lb_policy_name == NULL) {
321 338 : r->lb_policy_name = gpr_strdup(default_lb_policy_name);
322 : }
323 :
324 343 : path_slice =
325 343 : gpr_slice_new(args->uri->path, strlen(args->uri->path), do_nothing);
326 343 : gpr_slice_buffer_init(&path_parts);
327 :
328 343 : gpr_slice_split(path_slice, ",", &path_parts);
329 343 : r->num_addrs = path_parts.count;
330 343 : r->addrs = gpr_malloc(sizeof(struct sockaddr_storage) * r->num_addrs);
331 343 : r->addrs_len = gpr_malloc(sizeof(*r->addrs_len) * r->num_addrs);
332 :
333 707 : for (i = 0; i < r->num_addrs; i++) {
334 364 : grpc_uri ith_uri = *args->uri;
335 364 : char *part_str = gpr_dump_slice(path_parts.slices[i], GPR_DUMP_ASCII);
336 364 : ith_uri.path = part_str;
337 364 : if (!parse(&ith_uri, &r->addrs[i], &r->addrs_len[i])) {
338 0 : errors_found = 1; /* GPR_TRUE */
339 : }
340 364 : gpr_free(part_str);
341 364 : if (errors_found) break;
342 : }
343 :
344 343 : gpr_slice_buffer_destroy(&path_parts);
345 343 : gpr_slice_unref(path_slice);
346 343 : if (errors_found) {
347 0 : gpr_free(r);
348 0 : return NULL;
349 : }
350 :
351 343 : gpr_ref_init(&r->refs, 1);
352 343 : gpr_mu_init(&r->mu);
353 343 : grpc_resolver_init(&r->base, &sockaddr_resolver_vtable);
354 343 : r->subchannel_factory = args->subchannel_factory;
355 343 : grpc_subchannel_factory_ref(r->subchannel_factory);
356 :
357 343 : return &r->base;
358 : }
359 :
360 : /*
361 : * FACTORY
362 : */
363 :
364 7503 : static void sockaddr_factory_ref(grpc_resolver_factory *factory) {}
365 :
366 7503 : static void sockaddr_factory_unref(grpc_resolver_factory *factory) {}
367 :
368 : #define DECL_FACTORY(name) \
369 : static grpc_resolver *name##_factory_create_resolver( \
370 : grpc_resolver_factory *factory, grpc_resolver_args *args) { \
371 : return sockaddr_create(args, "pick_first", parse_##name); \
372 : } \
373 : static const grpc_resolver_factory_vtable name##_factory_vtable = { \
374 : sockaddr_factory_ref, sockaddr_factory_unref, \
375 : name##_factory_create_resolver, name##_get_default_authority, #name}; \
376 : static grpc_resolver_factory name##_resolver_factory = { \
377 : &name##_factory_vtable}; \
378 : grpc_resolver_factory *grpc_##name##_resolver_factory_create() { \
379 : return &name##_resolver_factory; \
380 : }
381 :
382 : #ifdef GPR_POSIX_SOCKET
383 2817 : DECL_FACTORY(unix)
384 : #endif
385 5029 : DECL_FACTORY(ipv4) DECL_FACTORY(ipv6)
|
