Switch to OkHttp. attempt to workaround oref api brokenness

If an API call fails for any reason, we retry a different server

Author: Andrew Pikler

oref-integration/pom.xml

@@ -72,6 +72,16 @@
             <artifactId>guava</artifactId>
             <version>33.4.8-jre</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-compress</artifactId>
+            <version>1.27.1</version>
+        </dependency>
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
+            <version>4.12.0</version>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>

oref-integration/src/main/java/com/github/pyckle/oref/cli/Main.java

@@ -1,5 +1,6 @@
 package com.github.pyckle.oref.cli;
 
+import com.github.pyckle.oref.integration.DnsUtil;
 import com.github.pyckle.oref.integration.activealerts.ActiveAlert;
 import com.github.pyckle.oref.integration.caching.CachedApiResult;
 import com.github.pyckle.oref.integration.caching.OrefApiCachingService;
@@ -15,6 +16,7 @@
  */
 public class Main {
     public static void main(String[] args) throws InterruptedException {
+        DnsUtil.disableDnsCaching();
         OrefApiCachingService orefApiCachingService = new OrefApiCachingService(new OrefConfig(new Properties()));
         orefApiCachingService.waitForInitialization();
         System.out.println("Caches initialized");

oref-integration/src/main/java/com/github/pyckle/oref/integration/DnsUtil.java

@@ -0,0 +1,15 @@
+package com.github.pyckle.oref.integration;
+
+public class DnsUtil
+{
+    /**
+     * Disable JVM dns caching to ensure we get fresh hosts whenever we resolve oref servers.
+     * <br>
+     * This MUST be called on startup before any networking is done. See also: <a
+     * href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/net/doc-files/net-properties.html">...</a>
+     */
+    public static void disableDnsCaching() {
+        java.security.Security.setProperty("networkaddress.cache.ttl", "0");
+        java.security.Security.setProperty("networkaddress.cache.negative.ttl", "0");
+    }
+}

oref-integration/src/main/java/com/github/pyckle/oref/integration/caching/ApiResponse.java

@@ -1,8 +1,8 @@
 package com.github.pyckle.oref.integration.caching;
 
-import java.net.http.HttpResponse;
+import okhttp3.Response;
 
 public record ApiResponse<T>(
-        HttpResponse<?> response,
+        Response response,
         T responseObj) {
 }

oref-integration/src/main/java/com/github/pyckle/oref/integration/caching/CachedApiResult.java

@@ -1,15 +1,10 @@
 package com.github.pyckle.oref.integration.caching;
 
-import com.github.pyckle.oref.integration.translationstores.DistrictStore;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import okhttp3.Headers;
+import okhttp3.Response;
 
-import java.net.http.HttpHeaders;
-import java.net.http.HttpResponse;
 import java.time.Instant;
-import java.time.ZonedDateTime;
-import java.time.format.DateTimeFormatter;
-import java.time.format.DateTimeParseException;
+import java.util.Date;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.regex.Pattern;
@@ -21,27 +16,26 @@ public record CachedApiResult<T>(Instant localTimestamp,
                                  long maxAge,
                                  T retrievedValue) {
 
-    private static final Logger logger = LoggerFactory.getLogger(DistrictStore.class);
     private static final String LAST_MODIFIED_HEADER = "Last-Modified";
     private static final String SERVER_TIMESTAMP = "Date";
     private static final String CACHE_CONTROL = "Cache-Control";
     private static final Pattern MAX_AGE_PATTERN = Pattern.compile("max-age=(\\d{1,18})");
 
     public static <T> CachedApiResult<T> buildCachedApiResult(Instant localTime,
-                                                              HttpResponse<?> response,
+                                                              Response response,
                                                               T retrievedValue) {
 
-        int statusCode = response.statusCode();
+        int statusCode = response.code();
         Instant serverTimestamp = parseDate(response.headers(), SERVER_TIMESTAMP);
         Instant lastModified = parseDate(response.headers(), LAST_MODIFIED_HEADER);
         long maxAge = maxAge(response.headers());
 
         return new CachedApiResult<>(localTime, statusCode, serverTimestamp, lastModified, maxAge, retrievedValue);
     }
 
-    static long maxAge(HttpHeaders headers) {
-        var header = headers.firstValue(CACHE_CONTROL);
-        return maxAge(header);
+    static long maxAge(Headers headers) {
+        var header = headers.get(CACHE_CONTROL);
+        return maxAge(Optional.ofNullable(header));
     }
 
     static long maxAge(Optional<String> header) {
@@ -55,18 +49,9 @@ static long maxAge(Optional<String> header) {
     }
 
 
-    private static Instant parseDate(HttpHeaders headers, String header) {
-        var headerDateStr = headers.firstValue(header);
-        if (headerDateStr.isPresent()) {
-            try {
-                ZonedDateTime zdt = ZonedDateTime.parse(headerDateStr.get(), DateTimeFormatter.RFC_1123_DATE_TIME);
-                return zdt.toInstant();
-            } catch (DateTimeParseException ex) {
-                logger.debug("Could not parse date: {}", headerDateStr.get());
-                logger.trace("Could not parse date", ex);
-            }
-        }
-        return null;
+    private static Instant parseDate(Headers headers, String header) {
+        Date date = headers.getDate(header);
+        return date == null ? null : date.toInstant();
     }
 
     /**

oref-integration/src/main/java/com/github/pyckle/oref/integration/caching/OrefApiClient.java

@@ -4,60 +4,163 @@
 import com.google.common.io.CountingInputStream;
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
+import okhttp3.CacheControl;
+import okhttp3.ConnectionPool;
+import okhttp3.Dns;
+import okhttp3.Headers;
+import okhttp3.HttpUrl;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
+import org.jetbrains.annotations.NotNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
-import java.net.http.HttpClient;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
-import java.util.zip.GZIPInputStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 public class OrefApiClient {
     private static final Logger logger = LoggerFactory.getLogger(OrefApiClient.class);
-    private static final HttpClient client = HttpClient.newHttpClient();
+    private static OkHttpClient client = null;
+    private static ConnectionPool connectionPool;
     private static final int SUCCESS_STATUS_CODE = 200;
 
     public static <T> ApiResponse<T> get(HttpRequest req, TypeToken<T> typeToken)
             throws IOException, InterruptedException {
         Gson gson = new Gson();
 
         Stopwatch stopwatch = Stopwatch.createStarted();
-        HttpResponse<InputStream> httpResponse = client.send(req, HttpResponse.BodyHandlers.ofInputStream());
-
-        // Body stream *MUST* be listed separately because we always want to close it even with there is an error
-        // creating the gzipReader. Not closing the body stream is a memory leak..
-        try (CountingInputStream is = new CountingInputStream(httpResponse.body());
-             InputStream gzipReader = wrapWithGzip(httpResponse, is);
-             InputStreamReader isr = new InputStreamReader(gzipReader, StandardCharsets.UTF_8)) {
-            if (httpResponse.statusCode() != SUCCESS_STATUS_CODE) {
-                logger.warn("Failed http request: {} {} {}", req.uri(), httpResponse.statusCode(), httpResponse.headers());
-                throw new RuntimeException("Unexpected Status Code " + httpResponse.statusCode());
+
+        List<String> allheaders = convertHeadersToOkHttp(req);
+        var okHttpReq = new Request.Builder()
+                .cacheControl(new CacheControl.Builder().noCache().build())
+                .headers(Headers.of(allheaders.toArray(new String[0])))
+                .url(req.uri().toURL())
+                .build();
+        var httpResponse = getHttpClient().newCall(okHttpReq).execute();
+        logger.debug("Received resp - content-length {} bytes in Request to {}: Response: {} in {}",
+                httpResponse.body() == null ? -1 : httpResponse.body().contentLength(), req.uri(),
+                httpResponse.code(), stopwatch.elapsed());
+
+        try (var is = httpResponse.body().byteStream();
+             var wrappedReader = new CountingInputStream(wrapWithGzip(httpResponse.request().url(), httpResponse.headers(), is));
+             var isr = new BufferedReader(new InputStreamReader(wrappedReader, StandardCharsets.UTF_8))) {
+            if (httpResponse.code() != SUCCESS_STATUS_CODE) {
+                logger.warn("Failed http request: {} {} {}", req.uri(), httpResponse.code(),
+                        httpResponse.headers());
+                throw new RuntimeException("Unexpected Status Code " + httpResponse.code());
             } else {
                 T ret = gson.fromJson(isr, typeToken);
 
-                logger.debug("Fetched {} bytes in Request to {}: Response: {} in {}", is.getCount(), req.uri(),
-                        httpResponse.statusCode(), stopwatch.elapsed());
-                logger.trace("Request to {}: Response: {} {} {}", req, httpResponse.statusCode(),
+                logger.debug("Decoded {} bytes uncompressed: {} bytes in Request to {}: Response: {} in {}",
+                        httpResponse.body().contentLength(), wrappedReader.getCount(), req.uri(), httpResponse.code(),
+                        stopwatch.elapsed());
+                logger.trace("Request to {}: Response: {} {} {}", req, httpResponse.code(),
                         httpResponse.headers(), ret);
 
                 return new ApiResponse<>(httpResponse, ret);
             }
+        } catch (Exception ex) {
+            logger.info("Failed to decode resp {}", httpResponse.request().url());
+            // evicting from the pool forces a reconnect, hopefully to a different host that is not broken.
+            connectionPool.evictAll();
+            throw ex;
+        }
+    }
+
+    @NotNull
+    private static List<String> convertHeadersToOkHttp(HttpRequest req)
+    {
+        List<String> allheaders = new ArrayList<>();
+        for (var h : req.headers().map().entrySet())
+        {
+            for (var val : h.getValue())
+            {
+                allheaders.add(h.getKey());
+                allheaders.add(val);
+            }
+        }
+        return allheaders;
+    }
+
+    private static synchronized OkHttpClient getHttpClient() {
+        if (client == null)
+        {
+            connectionPool = new ConnectionPool();
+            client = new OkHttpClient.Builder()
+                    .callTimeout(Duration.ofSeconds(8))
+                    .connectionPool(connectionPool)
+                    .dns(new RotatingDns())
+                    .build();
         }
+        return client;
     }
 
-    private static InputStream wrapWithGzip(HttpResponse<InputStream> httpResponse, InputStream is) throws IOException {
-        var contentEncoding = httpResponse.headers().allValues("Content-Encoding");
+    private static InputStream wrapWithGzip(HttpUrl url,  Headers responseHeaders, InputStream is) throws IOException {
+        var contentEncoding = responseHeaders.toMultimap().get("Content-Encoding");
         if (!contentEncoding.isEmpty()) {
             if (contentEncoding.size() == 1 && contentEncoding.get(0).equalsIgnoreCase("gzip")) {
-                is = new GZIPInputStream(is);
+                logger.debug("response is gzipped encoded {}", url);
+                return new GzipCompressorInputStream(is);
             } else {
                 throw new IllegalStateException("Unknown content encoding " + contentEncoding);
             }
         }
+        logger.debug("response is not gzipped encoded {}", url);
         return is;
     }
+
+    /**
+     * A dns resolver that ensures that the returned hosts are shuffled and the same host is not returned at the start of the list twice
+     */
+    private static class RotatingDns implements Dns
+    {
+        private final Map<String, InetAddress> cachedIps = new HashMap<>();
+
+        @NotNull
+        @Override
+        public synchronized List<InetAddress> lookup(@NotNull String hostname) throws UnknownHostException
+        {
+            try
+            {
+                var ips = InetAddress.getAllByName(hostname);
+                if (ips.length > 1 && cachedIps.containsKey(hostname))
+                {
+                    var oldFirstAddr = cachedIps.get(hostname);
+                    Collections.shuffle(Arrays.asList(ips));
+
+                    // make sure that the last first IP we returned is not the first IP again, since that is likely what was used.
+                    if (oldFirstAddr.equals(ips[0]))
+                    {
+                        ips[0] = ips[ips.length - 1];
+                        ips[ips.length - 1] = oldFirstAddr;
+                    }
+                }
+                if (ips.length > 0)
+                {
+                    cachedIps.put(hostname, ips[0]);
+                }
+                var ret = List.of(ips);
+                logger.info("resolved {} to {}", hostname, ret);
+                return ret;
+            }
+            catch (NullPointerException e)
+            {
+                throw new UnknownHostException("Broken system behaviour for dns lookup of " + hostname);
+            }
+        }
+    }
 }

oref-integration/src/main/java/com/github/pyckle/oref/integration/caching/OrefHttpRequestFactory.java

@@ -1,6 +1,7 @@
 package com.github.pyckle.oref.integration.caching;
 
 import java.net.URI;
+import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.time.Duration;
 import java.util.Arrays;
@@ -30,18 +31,23 @@ static HttpRequest buildRequest(URI uri, String... headers) {
     }
 
     private static String[] addAcceptEncodingGzip(String[] headers) {
-        String[] allHeaders = null;
+        String[] allHeaders = addHeader(headers, acceptEncodingHeader, acceptEncodingGzip);
+        return allHeaders;
+    }
 
+    private static String[] addHeader(String[] headers, String headerToAdd, String headerVal)
+    {
+        String[] allHeaders = null;
         for (int i = 0; i < headers.length; i += 2) {
-            if (headers[i].equalsIgnoreCase(acceptEncodingHeader)) {
+            if (headers[i].equalsIgnoreCase(headerToAdd)) {
                 allHeaders = headers;
                 break;
             }
         }
         if (allHeaders == null) {
             allHeaders = Arrays.copyOf(headers, headers.length + 2);
-            allHeaders[allHeaders.length - 2] = acceptEncodingHeader;
-            allHeaders[allHeaders.length - 1] = acceptEncodingGzip;
+            allHeaders[allHeaders.length - 2] = headerToAdd;
+            allHeaders[allHeaders.length - 1] = headerVal;
         }
         return allHeaders;
     }

oref-swingui/src/main/java/com/github/pyckle/oref/ui/Main.java

@@ -1,5 +1,6 @@
 package com.github.pyckle.oref.ui;
 
+import com.github.pyckle.oref.integration.DnsUtil;
 import com.github.pyckle.oref.integration.config.OrefConfig;
 
 import java.util.Arrays;
@@ -9,6 +10,7 @@
 public class Main {
 
     public static void main(String[] args) {
+        DnsUtil.disableDnsCaching();
         configSlf4j(args);
         Properties properties = parseProperties(args);
         javax.swing.SwingUtilities.invokeLater(() -> new PekudeiOrefGui(properties).createAndShowGUI());