1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.resolver.dns;
17
18 import io.netty.bootstrap.Bootstrap;
19 import io.netty.channel.AddressedEnvelope;
20 import io.netty.channel.Channel;
21 import io.netty.channel.ChannelFuture;
22 import io.netty.channel.ChannelFutureListener;
23 import io.netty.channel.ChannelHandlerContext;
24 import io.netty.channel.ChannelInboundHandlerAdapter;
25 import io.netty.channel.ChannelPromise;
26 import io.netty.handler.codec.dns.AbstractDnsOptPseudoRrRecord;
27 import io.netty.handler.codec.dns.DnsQuery;
28 import io.netty.handler.codec.dns.DnsQuestion;
29 import io.netty.handler.codec.dns.DnsRecord;
30 import io.netty.handler.codec.dns.DnsRecordType;
31 import io.netty.handler.codec.dns.DnsResponse;
32 import io.netty.handler.codec.dns.DnsSection;
33 import io.netty.handler.codec.dns.TcpDnsQueryEncoder;
34 import io.netty.handler.codec.dns.TcpDnsResponseDecoder;
35 import io.netty.util.ReferenceCountUtil;
36 import io.netty.util.concurrent.Future;
37 import io.netty.util.concurrent.FutureListener;
38 import io.netty.util.concurrent.Promise;
39 import io.netty.util.internal.SystemPropertyUtil;
40 import io.netty.util.internal.ThrowableUtil;
41 import io.netty.util.internal.logging.InternalLogger;
42 import io.netty.util.internal.logging.InternalLoggerFactory;
43
44 import java.net.InetSocketAddress;
45 import java.net.SocketAddress;
46 import java.util.concurrent.CancellationException;
47 import java.util.concurrent.TimeUnit;
48
49 import static io.netty.util.internal.ObjectUtil.checkNotNull;
50
51 abstract class DnsQueryContext {
52
53 private static final InternalLogger logger = InternalLoggerFactory.getInstance(DnsQueryContext.class);
54 private static final long ID_REUSE_ON_TIMEOUT_DELAY_MILLIS;
55
56 static {
57 ID_REUSE_ON_TIMEOUT_DELAY_MILLIS =
58 SystemPropertyUtil.getLong("io.netty.resolver.dns.idReuseOnTimeoutDelayMillis", 10000);
59 logger.debug("-Dio.netty.resolver.dns.idReuseOnTimeoutDelayMillis: {}", ID_REUSE_ON_TIMEOUT_DELAY_MILLIS);
60 }
61
62 private static final TcpDnsQueryEncoder TCP_ENCODER = new TcpDnsQueryEncoder();
63
64 private final Channel channel;
65 private final InetSocketAddress nameServerAddr;
66 private final DnsQueryContextManager queryContextManager;
67 private final DnsQueryLifecycleObserver queryLifecycleObserver;
68 private final Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise;
69
70 private final DnsQuestion question;
71 private final DnsRecord[] additionals;
72 private final DnsRecord optResource;
73
74 private final boolean recursionDesired;
75
76 private final Bootstrap socketBootstrap;
77
78 private final boolean retryWithTcpOnTimeout;
79 private final long queryTimeoutMillis;
80
81 private volatile Future<?> timeoutFuture;
82
83 private int id = Integer.MIN_VALUE;
84
85 DnsQueryContext(Channel channel,
86 InetSocketAddress nameServerAddr,
87 DnsQueryContextManager queryContextManager,
88 DnsQueryLifecycleObserver queryLifecycleObserver,
89 int maxPayLoadSize,
90 boolean recursionDesired,
91 long queryTimeoutMillis,
92 DnsQuestion question,
93 DnsRecord[] additionals,
94 Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise,
95 Bootstrap socketBootstrap,
96 boolean retryWithTcpOnTimeout) {
97 this.channel = checkNotNull(channel, "channel");
98 this.queryContextManager = checkNotNull(queryContextManager, "queryContextManager");
99 this.queryLifecycleObserver = checkNotNull(queryLifecycleObserver, "queryLifecycleObserver");
100 this.nameServerAddr = checkNotNull(nameServerAddr, "nameServerAddr");
101 this.question = checkNotNull(question, "question");
102 this.additionals = checkNotNull(additionals, "additionals");
103 this.promise = checkNotNull(promise, "promise");
104 this.recursionDesired = recursionDesired;
105 this.queryTimeoutMillis = queryTimeoutMillis;
106 this.socketBootstrap = socketBootstrap;
107 this.retryWithTcpOnTimeout = retryWithTcpOnTimeout;
108
109 if (maxPayLoadSize > 0 &&
110
111
112
113 !hasOptRecord(additionals)) {
114 optResource = new AbstractDnsOptPseudoRrRecord(maxPayLoadSize, 0, 0) {
115
116 };
117 } else {
118 optResource = null;
119 }
120 }
121
122 private static boolean hasOptRecord(DnsRecord[] additionals) {
123 if (additionals != null && additionals.length > 0) {
124 for (DnsRecord additional: additionals) {
125 if (additional.type() == DnsRecordType.OPT) {
126 return true;
127 }
128 }
129 }
130 return false;
131 }
132
133
134
135
136
137
138 final boolean isDone() {
139 return promise.isDone();
140 }
141
142
143
144
145
146
147 final DnsQuestion question() {
148 return question;
149 }
150
151
152
153
154
155
156
157
158 protected abstract DnsQuery newQuery(int id, InetSocketAddress nameServerAddr);
159
160
161
162
163
164
165 protected abstract String protocol();
166
167
168
169
170
171
172 final void writeQuery(boolean flush) {
173 assert id == Integer.MIN_VALUE : this.getClass().getSimpleName() +
174 ".writeQuery(...) can only be executed once.";
175
176 if ((id = queryContextManager.add(nameServerAddr, this)) == -1) {
177
178 IllegalStateException e = new IllegalStateException("query ID space exhausted: " + question());
179 finishFailure("failed to send a query via " + protocol(), e, false);
180 queryLifecycleObserver.queryWritten(nameServerAddr, channel.newFailedFuture(e));
181 return;
182 }
183
184
185 promise.addListener((FutureListener<AddressedEnvelope<DnsResponse, InetSocketAddress>>) future -> {
186
187 Future<?> timeoutFuture = DnsQueryContext.this.timeoutFuture;
188 if (timeoutFuture != null) {
189 DnsQueryContext.this.timeoutFuture = null;
190 timeoutFuture.cancel(false);
191 }
192
193 Throwable cause = future.cause();
194 if (cause instanceof DnsNameResolverTimeoutException || cause instanceof CancellationException) {
195
196
197
198 channel.eventLoop().schedule(new Runnable() {
199 @Override
200 public void run() {
201 removeFromContextManager(nameServerAddr);
202 }
203 }, ID_REUSE_ON_TIMEOUT_DELAY_MILLIS, TimeUnit.MILLISECONDS);
204 } else {
205
206
207 removeFromContextManager(nameServerAddr);
208 }
209 });
210 final DnsQuestion question = question();
211 final DnsQuery query = newQuery(id, nameServerAddr);
212
213 query.setRecursionDesired(recursionDesired);
214
215 query.addRecord(DnsSection.QUESTION, question);
216
217 for (DnsRecord record: additionals) {
218 query.addRecord(DnsSection.ADDITIONAL, record);
219 }
220
221 if (optResource != null) {
222 query.addRecord(DnsSection.ADDITIONAL, optResource);
223 }
224
225 if (logger.isDebugEnabled()) {
226 logger.debug("{} WRITE: {}, [{}: {}], {}",
227 channel, protocol(), id, nameServerAddr, question);
228 }
229
230 ChannelFuture f = sendQuery(query, flush);
231 queryLifecycleObserver.queryWritten(nameServerAddr, f);
232 }
233
234 private void removeFromContextManager(InetSocketAddress nameServerAddr) {
235 DnsQueryContext self = queryContextManager.remove(nameServerAddr, id);
236
237 assert self == this : "Removed DnsQueryContext is not the correct instance";
238 }
239
240 private ChannelFuture sendQuery(final DnsQuery query, final boolean flush) {
241 final ChannelPromise writePromise = channel.newPromise();
242 writeQuery(query, flush, writePromise);
243 return writePromise;
244 }
245
246 private void writeQuery(final DnsQuery query,
247 final boolean flush, ChannelPromise promise) {
248 final ChannelFuture writeFuture = flush ? channel.writeAndFlush(query, promise) :
249 channel.write(query, promise);
250 if (writeFuture.isDone()) {
251 onQueryWriteCompletion(queryTimeoutMillis, writeFuture);
252 } else {
253 writeFuture.addListener((ChannelFutureListener) future ->
254 onQueryWriteCompletion(queryTimeoutMillis, future));
255 }
256 }
257
258 private void onQueryWriteCompletion(final long queryTimeoutMillis,
259 ChannelFuture writeFuture) {
260 if (!writeFuture.isSuccess()) {
261 finishFailure("failed to send a query '" + id + "' via " + protocol(), writeFuture.cause(), false);
262 return;
263 }
264
265
266 if (queryTimeoutMillis > 0) {
267 timeoutFuture = channel.eventLoop().schedule(new Runnable() {
268 @Override
269 public void run() {
270 if (promise.isDone()) {
271
272 return;
273 }
274
275 finishFailure("query '" + id + "' via " + protocol() + " timed out after " +
276 queryTimeoutMillis + " milliseconds", null, true);
277 }
278 }, queryTimeoutMillis, TimeUnit.MILLISECONDS);
279 }
280 }
281
282
283
284
285
286 void finishSuccess(AddressedEnvelope<? extends DnsResponse, InetSocketAddress> envelope, boolean truncated) {
287
288 if (!truncated || !retryWithTcp(envelope)) {
289 final DnsResponse res = envelope.content();
290 if (res.count(DnsSection.QUESTION) != 1) {
291 logger.warn("{} Received a DNS response with invalid number of questions. Expected: 1, found: {}",
292 channel, envelope);
293 } else if (!question().equals(res.recordAt(DnsSection.QUESTION))) {
294 logger.warn("{} Received a mismatching DNS response. Expected: [{}], found: {}",
295 channel, question(), envelope);
296 } else if (trySuccess(envelope)) {
297 return;
298 }
299 envelope.release();
300 }
301 }
302
303 @SuppressWarnings("unchecked")
304 private boolean trySuccess(AddressedEnvelope<? extends DnsResponse, InetSocketAddress> envelope) {
305 return promise.trySuccess((AddressedEnvelope<DnsResponse, InetSocketAddress>) envelope);
306 }
307
308
309
310
311 final boolean finishFailure(String message, Throwable cause, boolean timeout) {
312 if (promise.isDone()) {
313 return false;
314 }
315 final DnsQuestion question = question();
316
317 final StringBuilder buf = new StringBuilder(message.length() + 128);
318 buf.append('[')
319 .append(id)
320 .append(": ")
321 .append(nameServerAddr)
322 .append("] ")
323 .append(question)
324 .append(' ')
325 .append(message)
326 .append(" (no stack trace available)");
327
328 final DnsNameResolverException e;
329 if (timeout) {
330
331
332 e = new DnsNameResolverTimeoutException(nameServerAddr, question, buf.toString());
333 if (retryWithTcpOnTimeout && retryWithTcp(e)) {
334
335 return false;
336 }
337 } else {
338 e = new DnsNameResolverException(nameServerAddr, question, buf.toString(), cause);
339 }
340 return promise.tryFailure(e);
341 }
342
343
344
345
346
347
348
349
350 private boolean retryWithTcp(final Object originalResult) {
351 if (socketBootstrap == null) {
352 return false;
353 }
354
355 socketBootstrap.connect(nameServerAddr).addListener((ChannelFutureListener) future -> {
356 if (!future.isSuccess()) {
357 logger.debug("{} Unable to fallback to TCP [{}: {}]",
358 future.channel(), id, nameServerAddr, future.cause());
359
360 finishOriginal(originalResult, future);
361 return;
362 }
363 final Channel tcpCh = future.channel();
364 Promise<AddressedEnvelope<DnsResponse, InetSocketAddress>> promise =
365 tcpCh.eventLoop().newPromise();
366 final TcpDnsQueryContext tcpCtx = new TcpDnsQueryContext(tcpCh,
367 (InetSocketAddress) tcpCh.remoteAddress(), queryContextManager, queryLifecycleObserver, 0,
368 recursionDesired, queryTimeoutMillis, question(), additionals, promise);
369 tcpCh.pipeline().addLast(TCP_ENCODER);
370 tcpCh.pipeline().addLast(new TcpDnsResponseDecoder());
371 tcpCh.pipeline().addLast(new ChannelInboundHandlerAdapter() {
372 @Override
373 public void channelRead(ChannelHandlerContext ctx, Object msg) {
374 Channel tcpCh = ctx.channel();
375 DnsResponse response = (DnsResponse) msg;
376 int queryId = response.id();
377
378 if (logger.isDebugEnabled()) {
379 logger.debug("{} RECEIVED: TCP [{}: {}], {}", tcpCh, queryId,
380 tcpCh.remoteAddress(), response);
381 }
382
383 DnsQueryContext foundCtx = queryContextManager.get(nameServerAddr, queryId);
384 if (foundCtx != null && foundCtx.isDone()) {
385 logger.debug("{} Received a DNS response for a query that was timed out or cancelled " +
386 ": TCP [{}: {}]", tcpCh, queryId, nameServerAddr);
387 response.release();
388 } else if (foundCtx == tcpCtx) {
389 tcpCtx.finishSuccess(new AddressedEnvelopeAdapter(
390 (InetSocketAddress) ctx.channel().remoteAddress(),
391 (InetSocketAddress) ctx.channel().localAddress(),
392 response), false);
393 } else {
394 response.release();
395 tcpCtx.finishFailure("Received TCP DNS response with unexpected ID", null, false);
396 if (logger.isDebugEnabled()) {
397 logger.debug("{} Received a DNS response with an unexpected ID: TCP [{}: {}]",
398 tcpCh, queryId, tcpCh.remoteAddress());
399 }
400 }
401 }
402
403 @Override
404 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
405 if (tcpCtx.finishFailure(
406 "TCP fallback error", cause, false) && logger.isDebugEnabled()) {
407 logger.debug("{} Error during processing response: TCP [{}: {}]",
408 ctx.channel(), id,
409 ctx.channel().remoteAddress(), cause);
410 }
411 }
412 });
413
414 promise.addListener(
415 (FutureListener<AddressedEnvelope<DnsResponse, InetSocketAddress>>) future1 -> {
416 if (future1.isSuccess()) {
417 finishSuccess(future1.getNow(), false);
418
419 ReferenceCountUtil.release(originalResult);
420 } else {
421
422 finishOriginal(originalResult, future1);
423 }
424 tcpCh.close();
425 });
426 tcpCtx.writeQuery(true);
427 });
428 return true;
429 }
430
431 @SuppressWarnings("unchecked")
432 private void finishOriginal(Object originalResult, Future<?> future) {
433 if (originalResult instanceof Throwable) {
434 Throwable error = (Throwable) originalResult;
435 ThrowableUtil.addSuppressed(error, future.cause());
436 promise.tryFailure(error);
437 } else {
438 finishSuccess((AddressedEnvelope<? extends DnsResponse, InetSocketAddress>) originalResult, false);
439 }
440 }
441
442 private static final class AddressedEnvelopeAdapter implements AddressedEnvelope<DnsResponse, InetSocketAddress> {
443 private final InetSocketAddress sender;
444 private final InetSocketAddress recipient;
445 private final DnsResponse response;
446
447 AddressedEnvelopeAdapter(InetSocketAddress sender, InetSocketAddress recipient, DnsResponse response) {
448 this.sender = sender;
449 this.recipient = recipient;
450 this.response = response;
451 }
452
453 @Override
454 public DnsResponse content() {
455 return response;
456 }
457
458 @Override
459 public InetSocketAddress sender() {
460 return sender;
461 }
462
463 @Override
464 public InetSocketAddress recipient() {
465 return recipient;
466 }
467
468 @Override
469 public AddressedEnvelope<DnsResponse, InetSocketAddress> retain() {
470 response.retain();
471 return this;
472 }
473
474 @Override
475 public AddressedEnvelope<DnsResponse, InetSocketAddress> retain(int increment) {
476 response.retain(increment);
477 return this;
478 }
479
480 @Override
481 public AddressedEnvelope<DnsResponse, InetSocketAddress> touch() {
482 response.touch();
483 return this;
484 }
485
486 @Override
487 public AddressedEnvelope<DnsResponse, InetSocketAddress> touch(Object hint) {
488 response.touch(hint);
489 return this;
490 }
491
492 @Override
493 public int refCnt() {
494 return response.refCnt();
495 }
496
497 @Override
498 public boolean release() {
499 return response.release();
500 }
501
502 @Override
503 public boolean release(int decrement) {
504 return response.release(decrement);
505 }
506
507 @Override
508 public boolean equals(Object obj) {
509 if (this == obj) {
510 return true;
511 }
512
513 if (!(obj instanceof AddressedEnvelope)) {
514 return false;
515 }
516
517 @SuppressWarnings("unchecked")
518 final AddressedEnvelope<?, SocketAddress> that = (AddressedEnvelope<?, SocketAddress>) obj;
519 if (sender() == null) {
520 if (that.sender() != null) {
521 return false;
522 }
523 } else if (!sender().equals(that.sender())) {
524 return false;
525 }
526
527 if (recipient() == null) {
528 if (that.recipient() != null) {
529 return false;
530 }
531 } else if (!recipient().equals(that.recipient())) {
532 return false;
533 }
534
535 return response.equals(obj);
536 }
537
538 @Override
539 public int hashCode() {
540 int hashCode = response.hashCode();
541 if (sender() != null) {
542 hashCode = hashCode * 31 + sender().hashCode();
543 }
544 if (recipient() != null) {
545 hashCode = hashCode * 31 + recipient().hashCode();
546 }
547 return hashCode;
548 }
549 }
550 }