Track time spent in DNS resolution by OKHTTP protocol, fixes #878

Author: jnioche

core/src/main/java/com/digitalpebble/stormcrawler/bolt/FetcherBolt.java

@@ -629,6 +629,14 @@ fit.t, new Values(fit.url,
                     long timeFetching = System.currentTimeMillis() - start;
 
                     final int byteLength = response.getContent().length;
+                    
+                    // get any metrics from the protocol metadata
+                    // expect Longs
+                    response.getMetadata().keySet().stream()
+                            .filter(s -> s.startsWith("metrics."))
+                            .forEach(s -> averagedMetrics.scope(s.substring(8))
+                                    .update(Long.parseLong(response.getMetadata()
+                                            .getFirstValue(s))));
 
                     averagedMetrics.scope("fetch_time").update(timeFetching);
                     averagedMetrics.scope("time_in_queues")

core/src/main/java/com/digitalpebble/stormcrawler/bolt/SimpleFetcherBolt.java

@@ -433,6 +433,13 @@ input, new Values(urlString, metadata,
             long timeFetching = System.currentTimeMillis() - start;
 
             final int byteLength = response.getContent().length;
+            
+            // get any metrics from the protocol metadata
+            response.getMetadata().keySet().stream()
+                    .filter(s -> s.startsWith("metrics."))
+                    .forEach(s -> averagedMetrics.scope(s.substring(8))
+                            .update(Long.parseLong(
+                                    response.getMetadata().getFirstValue(s))));
 
             averagedMetrics.scope("wait_time").update(timeWaiting);
             averagedMetrics.scope("fetch_time").update(timeFetching);

core/src/main/java/com/digitalpebble/stormcrawler/protocol/okhttp/DNSResolutionListener.java

@@ -0,0 +1,56 @@
+/**
+ * Licensed to DigitalPebble Ltd under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * DigitalPebble licenses this file to You 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.
+ */
+
+package com.digitalpebble.stormcrawler.protocol.okhttp;
+
+import java.net.InetAddress;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.LoggerFactory;
+
+import okhttp3.Call;
+import okhttp3.EventListener;
+
+public class DNSResolutionListener extends EventListener {
+
+    private static final org.slf4j.Logger LOG = LoggerFactory
+            .getLogger(DNSResolutionListener.class);
+
+    private long dnsStartMillis;
+
+    final Map<String, Long> times;
+
+    public DNSResolutionListener(final Map<String, Long> times) {
+        this.times = times;
+    }
+
+    @Override
+    public void dnsEnd(Call call, String domainName,
+            List<InetAddress> inetAddressList) {
+        long timeSpent = System.currentTimeMillis() - dnsStartMillis;
+        LOG.debug("DNS resolution for {} took {} millisecs", domainName,
+                timeSpent);
+        times.put(call.toString(), timeSpent);
+    }
+
+    @Override
+    public void dnsStart(Call call, String domainName) {
+        dnsStartMillis = System.currentTimeMillis();
+    }
+
+}

core/src/main/java/com/digitalpebble/stormcrawler/protocol/okhttp/HttpProtocol.java

@@ -25,9 +25,11 @@
 import java.security.cert.CertificateException;
 import java.util.ArrayList;
 import java.util.Base64;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import javax.net.ssl.HostnameVerifier;
@@ -54,6 +56,7 @@
 import okhttp3.Call;
 import okhttp3.Connection;
 import okhttp3.Credentials;
+import okhttp3.EventListener;
 import okhttp3.Headers;
 import okhttp3.Interceptor;
 import okhttp3.MediaType;
@@ -65,6 +68,7 @@
 import okhttp3.Response;
 import okhttp3.ResponseBody;
 import okhttp3.Route;
+import okhttp3.EventListener.Factory;
 import okio.BufferedSource;
 
 public class HttpProtocol extends AbstractHttpProtocol {
@@ -86,6 +90,9 @@ public class HttpProtocol extends AbstractHttpProtocol {
 
     private final List<String[]> customRequestHeaders = new LinkedList<>();
 
+    // track the time spent for each URL in DNS resolution
+    private final Map<String, Long> DNStimes = new HashMap<>();
+
     private static final TrustManager[] trustAllCerts = new TrustManager[] { new X509TrustManager() {
         @Override
         public void checkClientTrusted(
@@ -237,6 +244,13 @@ public boolean verify(String hostname, SSLSession session) {
             });
         }
 
+        builder.eventListenerFactory(new Factory() {
+            @Override
+            public EventListener create(Call call) {
+                return new DNSResolutionListener(DNStimes);
+            }
+        });
+
         client = builder.build();
     }
 
@@ -332,7 +346,13 @@ public ProtocolResponse getProtocolOutput(String url, final Metadata metadata) t
                         trimmed.getValue().toString().toLowerCase(Locale.ROOT));
                 LOG.warn("HTTP content trimmed to {}", bytes.length);
             }
-
+            
+            Long DNSResolution = DNStimes.remove(call.toString());
+            if (DNSResolution != null) {
+                responsemetadata.setValue("metrics.dns.resolution.msec",
+                        DNSResolution.toString());
+            }
+            
             return new ProtocolResponse(bytes, response.code(), responsemetadata);
         }
     }
@@ -358,7 +378,7 @@ private final byte[] toByteArray(final ResponseBody responseBody,
         int bytesRequested = 0;
         int bufferGrowStepBytes = 8192;
 
-        while (source.buffer().size() <= maxContentBytes) {
+        while (source.getBuffer().size() <= maxContentBytes) {
             bytesRequested += Math.min(bufferGrowStepBytes,
             /*
              * request one byte more than required to reliably detect truncated
@@ -371,7 +391,7 @@ private final byte[] toByteArray(final ResponseBody responseBody,
                 success = source.request(bytesRequested);
             } catch (IOException e) {
                 // requesting more content failed, e.g. by a socket timeout
-                if (partialContentAsTrimmed && source.buffer().size() > 0) {
+                if (partialContentAsTrimmed && source.getBuffer().size() > 0) {
                     // treat already fetched content as trimmed
                     trimmed.setValue(TrimmedContentReason.DISCONNECT);
                     LOG.debug("Exception while fetching {}", e);
@@ -392,17 +412,17 @@ private final byte[] toByteArray(final ResponseBody responseBody,
 
             // okhttp may fetch more content than requested, quickly "increment"
             // bytes
-            bytesRequested = (int) source.buffer().size();
+            bytesRequested = (int) source.getBuffer().size();
         }
-        int bytesBuffered = (int) source.buffer().size();
+        int bytesBuffered = (int) source.getBuffer().size();
         int bytesToCopy = bytesBuffered;
         if (maxContent != -1 && bytesToCopy > maxContent) {
             // okhttp's internal buffer is larger than maxContent
             trimmed.setValue(TrimmedContentReason.LENGTH);
             bytesToCopy = maxContentBytes;
         }
         byte[] arr = new byte[bytesToCopy];
-        source.buffer().readFully(arr);
+        source.getBuffer().readFully(arr);
         return arr;
     }