Download files specified in url config field

3 files changed, 130 insertions, 5 deletions

diff --git a/config/src/main/java/com/yahoo/vespa/config/ConfigPayloadApplier.java b/config/src/main/java/com/yahoo/vespa/config/ConfigPayloadApplier.java
index 8a12405d505..4415ac21948 100644
--- a/config/src/main/java/com/yahoo/vespa/config/ConfigPayloadApplier.java
+++ b/config/src/main/java/com/yahoo/vespa/config/ConfigPayloadApplier.java
@@ -20,7 +20,11 @@ import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.nio.file.Path;
-import java.util.*;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.Stack;
 import java.util.logging.Logger;
 
 /**
@@ -36,15 +40,17 @@ public class ConfigPayloadApplier<T extends ConfigInstance.Builder> {
 
     private final ConfigInstance.Builder rootBuilder;
     private final ConfigTransformer.PathAcquirer pathAcquirer;
+    private final UrlDownloader urlDownloader;
     private final Stack<NamedBuilder> stack = new Stack<>();
 
     public ConfigPayloadApplier(T builder) {
-        this(builder, new IdentityPathAcquirer());
+        this(builder, new IdentityPathAcquirer(), null);
     }
 
-    public ConfigPayloadApplier(T builder, ConfigTransformer.PathAcquirer pathAcquirer) {
+    public ConfigPayloadApplier(T builder, ConfigTransformer.PathAcquirer pathAcquirer, UrlDownloader urlDownloader) {
         this.rootBuilder = builder;
         this.pathAcquirer = pathAcquirer;
+        this.urlDownloader = urlDownloader;
         debug("rootBuilder=" + rootBuilder);
     }
 
@@ -207,6 +213,12 @@ public class ConfigPayloadApplier<T extends ConfigInstance.Builder> {
             if (isPathField(builder, methodName)) {
                 FileReference wrappedPath = resolvePath((String)value);
                 invokeSetter(builder, methodName, key, wrappedPath);
+
+            // Need to convert url into actual file if 'url' type is used
+            } else if (isUrlField(builder, methodName)) {
+                UrlReference url = resolveUrl((String)value);
+                invokeSetter(builder, methodName, key, url);
+
             } else {
                 invokeSetter(builder, methodName, key, value);
             }
@@ -258,7 +270,8 @@ public class ConfigPayloadApplier<T extends ConfigInstance.Builder> {
 
             // Need to convert url into actual file if 'url' type is used
             } else if (isUrlField(builder, methodName)) {
-                throw new UnsupportedOperationException("'url' type is not yet implemented");
+                UrlReference url = resolveUrl(Utf8.toString(value.asUtf8()));
+                invokeSetter(builder, methodName, url);
 
             } else {
                 Object object = getValueFromInspector(value);
@@ -276,6 +289,14 @@ public class ConfigPayloadApplier<T extends ConfigInstance.Builder> {
         return newFileReference(path.toString());
     }
 
+    private UrlReference resolveUrl(String url) {
+        if (urlDownloader == null || !urlDownloader.isValid()) {
+            throw new RuntimeException("Resolving url field failed due to missing or invalid URL downloader.");
+        }
+        File file = urlDownloader.waitFor(new UrlReference(url), 10 * 60);
+        return new UrlReference(file.getAbsolutePath());
+    }
+
     private FileReference newFileReference(String fileReference) {
         try {
             Constructor<FileReference> constructor = FileReference.class.getDeclaredConstructor(String.class);
@@ -515,4 +536,5 @@ public class ConfigPayloadApplier<T extends ConfigInstance.Builder> {
             return new File(fileReference.value()).toPath();
         }
     }
+
 }
diff --git a/config/src/main/java/com/yahoo/vespa/config/ConfigTransformer.java b/config/src/main/java/com/yahoo/vespa/config/ConfigTransformer.java
index ac3f490182a..163e010cdd6 100644
--- a/config/src/main/java/com/yahoo/vespa/config/ConfigTransformer.java
+++ b/config/src/main/java/com/yahoo/vespa/config/ConfigTransformer.java
@@ -26,6 +26,7 @@ public class ConfigTransformer<T extends ConfigInstance> {
     private final Class<T> clazz;
 
     private static volatile PathAcquirer pathAcquirer = new IdentityPathAcquirer();
+    private static volatile UrlDownloader urlDownloader;
 
     /**
      * For internal use only *
@@ -36,6 +37,10 @@ public class ConfigTransformer<T extends ConfigInstance> {
                 pathAcquirer;
     }
 
+    public static void setUrlDownloader(UrlDownloader urlDownloader) {
+        ConfigTransformer.urlDownloader = urlDownloader;
+    }
+
     /**
      * Create a transformer capable of converting payloads to clazz
      *
@@ -53,7 +58,7 @@ public class ConfigTransformer<T extends ConfigInstance> {
      */
     public ConfigInstance.Builder toConfigBuilder(ConfigPayload payload) {
         ConfigInstance.Builder builder = getRootBuilder();
-        ConfigPayloadApplier<?> creator = new ConfigPayloadApplier<>(builder, pathAcquirer);
+        ConfigPayloadApplier<?> creator = new ConfigPayloadApplier<>(builder, pathAcquirer, urlDownloader);
         creator.applyPayload(payload);
         return builder;
     }
diff --git a/config/src/main/java/com/yahoo/vespa/config/UrlDownloader.java b/config/src/main/java/com/yahoo/vespa/config/UrlDownloader.java
new file mode 100644
index 00000000000..cd72c83d100
--- /dev/null
+++ b/config/src/main/java/com/yahoo/vespa/config/UrlDownloader.java
@@ -0,0 +1,98 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.config;
+
+import com.yahoo.config.UrlReference;
+import com.yahoo.jrt.Request;
+import com.yahoo.jrt.Spec;
+import com.yahoo.jrt.StringValue;
+import com.yahoo.jrt.Supervisor;
+import com.yahoo.jrt.Target;
+import com.yahoo.jrt.Transport;
+import com.yahoo.log.LogLevel;
+import com.yahoo.vespa.defaults.Defaults;
+
+import java.io.File;
+import java.util.logging.Logger;
+
+/**
+ * @author lesters
+ */
+public class UrlDownloader {
+
+    private static final Logger log = Logger.getLogger(UrlDownloader.class.getName());
+
+    private static final int BASE_ERROR_CODE = 0x10000;
+    public  static final int DOES_NOT_EXIST = BASE_ERROR_CODE + 1;
+    public  static final int INTERNAL_ERROR = BASE_ERROR_CODE + 2;
+    public  static final int HTTP_ERROR = BASE_ERROR_CODE + 3;
+
+    private final Supervisor supervisor = new Supervisor(new Transport());
+    private final Spec spec;
+    private Target target;
+
+    public UrlDownloader() {
+        spec = new Spec(Defaults.getDefaults().vespaHostname(), Defaults.getDefaults().vespaConfigProxyRpcPort());
+        connect();
+    }
+
+    public void shutdown() {
+        supervisor.transport().shutdown().join();
+    }
+
+    private void connect() {
+        int timeRemaining = 5000;
+        try {
+            while (timeRemaining > 0) {
+                target = supervisor.connectSync(spec);
+                if (target.isValid()) {
+                    log.log(LogLevel.DEBUG, "Successfully connected to '" + spec + "', this = " + System.identityHashCode(this));
+                    return;
+                }
+                Thread.sleep(500);
+                timeRemaining -= 500;
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public boolean isValid() {
+        return target.isValid();
+    }
+
+    private boolean temporaryError(Request req) {
+        return false;  // Currently, none of the errors are considered temporary
+    }
+
+    public File waitFor(UrlReference urlReference, long timeout) {
+        long start = System.currentTimeMillis() / 1000;
+        long timeLeft = timeout;
+        do {
+            Request request = new Request("url.waitFor");
+            request.parameters().add(new StringValue(urlReference.value()));
+
+            double rpcTimeout = Math.min(timeLeft, 60.0);
+            log.log(LogLevel.DEBUG, "InvokeSync waitFor " + urlReference + " with " + rpcTimeout + " seconds timeout");
+            target.invokeSync(request, rpcTimeout);
+
+            if (request.checkReturnTypes("s")) {
+                return new File(request.returnValues().get(0).asString());
+            } else if (!request.isError()) {
+                throw new RuntimeException("Invalid response: " + request.returnValues());
+            } else if (temporaryError(request)) {
+                log.log(LogLevel.INFO, "Retrying waitFor for " + urlReference + ": " + request.errorCode() + " -- " + request.errorMessage());
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException("Interrupted sleep between retries of waitFor", e);
+                }
+            } else {
+                 throw new RuntimeException("Wait for " + urlReference + " failed:" + request.errorMessage() + " (" + request.errorCode() + ")");
+            }
+            timeLeft = start + timeout - System.currentTimeMillis() / 1000;
+        } while (timeLeft > 0);
+
+        throw new RuntimeException("Timed out waiting for " + urlReference + " after " + timeout);
+    }
+
+}