Merge pull request #19580 from vespa-engine/hmusum/improve-file-reference-download

Reduce file distribution download timeout [run-systemtest]

7 files changed, 50 insertions, 55 deletions

diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileDistributionAndUrlDownload.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileDistributionAndUrlDownload.java
index 4d03522e980..f2f52dca9fa 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileDistributionAndUrlDownload.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileDistributionAndUrlDownload.java
@@ -28,7 +28,7 @@ public class FileDistributionAndUrlDownload {
     public FileDistributionAndUrlDownload(Supervisor supervisor, ConfigSourceSet source) {
         fileDistributionRpcServer =
                 new FileDistributionRpcServer(supervisor,
-                                              new FileDownloader(new JRTConnectionPool(source, supervisor), supervisor));
+                                              new FileDownloader(new JRTConnectionPool(source, supervisor), supervisor, Duration.ofMinutes(5)));
         urlDownloadRpcServer = new UrlDownloadRpcServer(supervisor);
         cleanupExecutor.scheduleAtFixedRate(new CachedFilesMaintainer(), delay.toSeconds(), delay.toSeconds(), TimeUnit.SECONDS);
     }
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java b/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java
index 5d9a9c7b836..f34beae9c46 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileServer.java
@@ -5,11 +5,14 @@ import com.google.inject.Inject;
 import com.yahoo.cloud.config.ConfigserverConfig;
 import com.yahoo.concurrent.DaemonThreadFactory;
 import com.yahoo.config.FileReference;
+import com.yahoo.config.subscription.ConfigSourceSet;
 import com.yahoo.jrt.Int32Value;
 import com.yahoo.jrt.Request;
 import com.yahoo.jrt.StringValue;
 import com.yahoo.jrt.Supervisor;
 import com.yahoo.jrt.Transport;
+import com.yahoo.vespa.config.ConnectionPool;
+import com.yahoo.vespa.config.JRTConnectionPool;
 import com.yahoo.vespa.defaults.Defaults;
 import com.yahoo.vespa.filedistribution.CompressedFileReference;
 import com.yahoo.vespa.filedistribution.EmptyFileReferenceData;
@@ -24,6 +27,7 @@ import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.logging.Level;
@@ -71,16 +75,12 @@ public class FileServer {
     @Inject
     public FileServer(ConfigserverConfig configserverConfig) {
         this(new File(Defaults.getDefaults().underVespaHome(configserverConfig.fileReferencesDir())),
-             new FileDownloader(getOtherConfigServersInCluster(configserverConfig),
-                                new Supervisor(new Transport("filedistribution-pool"))
-                                        .setDropEmptyBuffers(true)));
+             createFileDownloader(getOtherConfigServersInCluster(configserverConfig)));
     }
 
     // For testing only
     public FileServer(File rootDir) {
-        this(rootDir, new FileDownloader(FileDownloader.emptyConnectionPool(),
-                                         new Supervisor(new Transport("fileserver-for-testing"))
-                                                 .setDropEmptyBuffers(true)));
+        this(rootDir, createFileDownloader(List.of()));
     }
 
     public FileServer(File rootDir, FileDownloader fileDownloader) {
@@ -205,4 +205,18 @@ public class FileServer {
         executor.shutdown();
     }
 
+    private static FileDownloader createFileDownloader(List<String> configServers) {
+        Supervisor supervisor = new Supervisor(new Transport("filedistribution-pool")).setDropEmptyBuffers(true);
+        return new FileDownloader(configServers.isEmpty()
+                                          ? FileDownloader.emptyConnectionPool()
+                                          : getConnectionPool(configServers, supervisor),
+                                  supervisor);
+    }
+
+    private static ConnectionPool getConnectionPool(List<String> configServers, Supervisor supervisor) {
+        return configServers.size() > 0
+                ? new JRTConnectionPool(new ConfigSourceSet(configServers), supervisor)
+                : FileDownloader.emptyConnectionPool();
+    }
+
 }
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/maintenance/ApplicationPackageMaintainer.java b/configserver/src/main/java/com/yahoo/vespa/config/server/maintenance/ApplicationPackageMaintainer.java
index ee775fa7afb..d83636d08d6 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/maintenance/ApplicationPackageMaintainer.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/maintenance/ApplicationPackageMaintainer.java
@@ -14,7 +14,6 @@ import com.yahoo.vespa.config.server.session.SessionRepository;
 import com.yahoo.vespa.config.server.tenant.Tenant;
 import com.yahoo.vespa.curator.Curator;
 import com.yahoo.vespa.defaults.Defaults;
-import com.yahoo.vespa.filedistribution.Downloads;
 import com.yahoo.vespa.filedistribution.FileDownloader;
 import com.yahoo.vespa.filedistribution.FileReferenceDownload;
 import com.yahoo.vespa.flags.FlagSource;
@@ -94,8 +93,7 @@ public class ApplicationPackageMaintainer extends ConfigServerMaintainer {
     private FileDownloader createFileDownloader() {
         return new FileDownloader(new JRTConnectionPool(new ConfigSourceSet(getOtherConfigServersInCluster(configserverConfig)), supervisor),
                                   supervisor,
-                                  downloadDirectory,
-                                  new Downloads());
+                                  downloadDirectory);
     }
 
     @Override
diff --git a/configserver/src/test/java/com/yahoo/vespa/config/server/filedistribution/FileServerTest.java b/configserver/src/test/java/com/yahoo/vespa/config/server/filedistribution/FileServerTest.java
index 1cb6b2a13cd..29ec11bad26 100644
--- a/configserver/src/test/java/com/yahoo/vespa/config/server/filedistribution/FileServerTest.java
+++ b/configserver/src/test/java/com/yahoo/vespa/config/server/filedistribution/FileServerTest.java
@@ -7,7 +7,6 @@ import com.yahoo.io.IOUtils;
 import com.yahoo.jrt.Supervisor;
 import com.yahoo.jrt.Transport;
 import com.yahoo.net.HostName;
-import com.yahoo.vespa.filedistribution.Downloads;
 import com.yahoo.vespa.filedistribution.FileDownloader;
 import com.yahoo.vespa.filedistribution.FileReferenceData;
 import com.yahoo.vespa.filedistribution.FileReferenceDownload;
@@ -142,7 +141,6 @@ public class FileServerTest {
             super(FileDownloader.emptyConnectionPool(),
                   new Supervisor(new Transport("mock")).setDropEmptyBuffers(true),
                   downloadDirectory,
-                  new Downloads(),
                   Duration.ofMillis(100),
                   Duration.ofMillis(100));
         }
diff --git a/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/CompressedFileReference.java b/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/CompressedFileReference.java
index e8fe09a89e3..cbe810fe9b6 100644
--- a/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/CompressedFileReference.java
+++ b/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/CompressedFileReference.java
@@ -2,7 +2,6 @@
 package com.yahoo.vespa.filedistribution;
 
 import com.google.common.io.ByteStreams;
-import java.util.logging.Level;
 import org.apache.commons.compress.archivers.ArchiveEntry;
 import org.apache.commons.compress.archivers.ArchiveInputStream;
 import org.apache.commons.compress.archivers.ArchiveOutputStream;
@@ -18,6 +17,7 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.List;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import java.util.zip.GZIPInputStream;
@@ -106,7 +106,7 @@ public class CompressedFileReference {
     }
 
     private static void writeFileToTar(ArchiveOutputStream taos, File baseDir, File file) throws IOException {
-        log.log(Level.FINE, () -> "Adding file to tar: " + baseDir.toPath().relativize(file.toPath()).toString());
+        log.log(Level.FINEST, () -> "Adding file to tar: " + baseDir.toPath().relativize(file.toPath()).toString());
         taos.putArchiveEntry(taos.createArchiveEntry(file, baseDir.toPath().relativize(file.toPath()).toString()));
         ByteStreams.copy(new FileInputStream(file), taos);
         taos.closeArchiveEntry();
diff --git a/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java b/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java
index 32b7a0a167e..1d638a427f9 100644
--- a/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java
+++ b/filedistribution/src/main/java/com/yahoo/vespa/filedistribution/FileDownloader.java
@@ -2,17 +2,14 @@
 package com.yahoo.vespa.filedistribution;
 
 import com.yahoo.config.FileReference;
-import com.yahoo.config.subscription.ConfigSourceSet;
 import com.yahoo.jrt.Supervisor;
 import com.yahoo.vespa.config.Connection;
 import com.yahoo.vespa.config.ConnectionPool;
-import com.yahoo.vespa.config.JRTConnectionPool;
 import com.yahoo.vespa.defaults.Defaults;
 import com.yahoo.yolean.Exceptions;
 
 import java.io.File;
 import java.time.Duration;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -31,31 +28,35 @@ import java.util.logging.Logger;
  */
 public class FileDownloader implements AutoCloseable {
 
-    private final static Logger log = Logger.getLogger(FileDownloader.class.getName());
-    public static File defaultDownloadDirectory = new File(Defaults.getDefaults().underVespaHome("var/db/vespa/filedistribution"));
+    private static final Logger log = Logger.getLogger(FileDownloader.class.getName());
+    private static final Duration defaultTimeout = Duration.ofMinutes(3);
+    private static final Duration defaultSleepBetweenRetries = Duration.ofSeconds(10);
+    public static final File defaultDownloadDirectory = new File(Defaults.getDefaults().underVespaHome("var/db/vespa/filedistribution"));
 
     private final ConnectionPool connectionPool;
     private final Supervisor supervisor;
     private final File downloadDirectory;
     private final Duration timeout;
     private final FileReferenceDownloader fileReferenceDownloader;
-    private final Downloads downloads;
+    private final Downloads downloads = new Downloads();
 
-    public FileDownloader(List<String> configServers, Supervisor supervisor) {
-        this(getConnectionPool(configServers, supervisor), supervisor);
+    public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor) {
+        this(connectionPool, supervisor, defaultDownloadDirectory, defaultTimeout, defaultSleepBetweenRetries);
     }
 
-    public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor) {
-        this(connectionPool, supervisor, defaultDownloadDirectory, new Downloads());
+    public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor, Duration timeout) {
+        this(connectionPool, supervisor, defaultDownloadDirectory, timeout, defaultSleepBetweenRetries);
     }
 
-    public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor, File downloadDirectory, Downloads downloads) {
-        // TODO: Reduce timeout even more, timeout is so long that we might get starvation
-        this(connectionPool, supervisor, downloadDirectory, downloads, Duration.ofMinutes(5), Duration.ofSeconds(10));
+    public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor, File downloadDirectory) {
+        this(connectionPool, supervisor, downloadDirectory, defaultTimeout, defaultSleepBetweenRetries);
     }
 
-    public FileDownloader(ConnectionPool connectionPool, Supervisor supervisor, File downloadDirectory, Downloads downloads,
-                          Duration timeout, Duration sleepBetweenRetries) {
+    public FileDownloader(ConnectionPool connectionPool,
+                          Supervisor supervisor,
+                          File downloadDirectory,
+                          Duration timeout,
+                          Duration sleepBetweenRetries) {
         this.connectionPool = connectionPool;
         this.supervisor = supervisor;
         this.downloadDirectory = downloadDirectory;
@@ -63,7 +64,6 @@ public class FileDownloader implements AutoCloseable {
         // Needed to receive RPC receiveFile* calls from server after asking for files
         new FileReceiver(supervisor, downloads, downloadDirectory);
         this.fileReferenceDownloader = new FileReferenceDownloader(connectionPool, downloads, timeout, sleepBetweenRetries);
-        this.downloads = downloads;
     }
 
     public Optional<File> getFile(FileReference fileReference) {
@@ -95,6 +95,8 @@ public class FileDownloader implements AutoCloseable {
 
     public ConnectionPool connectionPool() { return connectionPool; }
 
+    public Downloads downloads() { return downloads; }
+
     File downloadDirectory() {
         return downloadDirectory;
     }
@@ -121,19 +123,9 @@ public class FileDownloader implements AutoCloseable {
         return downloads.get(fileReference).isPresent();
     }
 
-    private boolean alreadyDownloaded(FileReferenceDownload fileReferenceDownload) {
-        try {
-            return getFileFromFileSystem(fileReferenceDownload.fileReference()).isPresent();
-        } catch (RuntimeException e) {
-            return false;
-        }
-    }
-
     /** Start a download, don't wait for result */
     public void downloadIfNeeded(FileReferenceDownload fileReferenceDownload) {
-        if (alreadyDownloaded(fileReferenceDownload)) return;
-
-        download(fileReferenceDownload);
+        getFutureFile(fileReferenceDownload);
     }
 
     /** Download, the future returned will be complete()d by receiving method in {@link FileReceiver} */
@@ -146,12 +138,6 @@ public class FileDownloader implements AutoCloseable {
         supervisor.transport().shutdown().join();
     }
 
-    private static ConnectionPool getConnectionPool(List<String> configServers, Supervisor supervisor) {
-        return configServers.size() > 0
-                ? new JRTConnectionPool(new ConfigSourceSet(configServers), supervisor)
-                : emptyConnectionPool();
-    }
-
     public static ConnectionPool emptyConnectionPool() {
         return new EmptyConnectionPool();
     }
diff --git a/filedistribution/src/test/java/com/yahoo/vespa/filedistribution/FileDownloaderTest.java b/filedistribution/src/test/java/com/yahoo/vespa/filedistribution/FileDownloaderTest.java
index 6855f7f818c..79530c39ad7 100644
--- a/filedistribution/src/test/java/com/yahoo/vespa/filedistribution/FileDownloaderTest.java
+++ b/filedistribution/src/test/java/com/yahoo/vespa/filedistribution/FileDownloaderTest.java
@@ -41,7 +41,6 @@ public class FileDownloaderTest {
     private static final Duration sleepBetweenRetries = Duration.ofMillis(10);
 
     private MockConnection connection;
-    private Downloads downloads;
     private FileDownloader fileDownloader;
     private File downloadDir;
     private Supervisor supervisor;
@@ -51,9 +50,8 @@ public class FileDownloaderTest {
         try {
             downloadDir = Files.createTempDirectory("filedistribution").toFile();
             connection = new MockConnection();
-            downloads = new Downloads();
             supervisor = new Supervisor(new Transport()).setDropEmptyBuffers(true);
-            fileDownloader = new FileDownloader(connection, supervisor, downloadDir, downloads, Duration.ofSeconds(1), sleepBetweenRetries);
+            fileDownloader = new FileDownloader(connection, supervisor, downloadDir, Duration.ofSeconds(1), sleepBetweenRetries);
         } catch (IOException e) {
             e.printStackTrace();
             fail(e.getMessage());
@@ -124,7 +122,7 @@ public class FileDownloaderTest {
             assertEquals("some other content", IOUtils.readFile(downloadedFile.get()));
 
             // Verify download status when downloaded
-            System.out.println(downloads.downloadStatuses());
+            System.out.println(fileDownloader.downloads().downloadStatuses());
             assertDownloadStatus(fileReference, 1.0);
         }
 
@@ -166,7 +164,7 @@ public class FileDownloaderTest {
 
     @Test
     public void getFileWhenConnectionError() throws IOException {
-        fileDownloader = new FileDownloader(connection, supervisor, downloadDir, downloads, Duration.ofSeconds(2), sleepBetweenRetries);
+        fileDownloader = new FileDownloader(connection, supervisor, downloadDir, Duration.ofSeconds(2), sleepBetweenRetries);
         File downloadDir = fileDownloader.downloadDirectory();
 
         int timesToFail = 2;
@@ -200,7 +198,7 @@ public class FileDownloaderTest {
     public void getFileWhenDownloadInProgress() throws IOException, ExecutionException, InterruptedException {
         ExecutorService executor = Executors.newFixedThreadPool(2);
         String filename = "abc.jar";
-        fileDownloader = new FileDownloader(connection, supervisor, downloadDir, downloads, Duration.ofSeconds(3), sleepBetweenRetries);
+        fileDownloader = new FileDownloader(connection, supervisor, downloadDir, Duration.ofSeconds(3), sleepBetweenRetries);
         File downloadDir = fileDownloader.downloadDirectory();
 
         // Delay response so that we can make a second request while downloading the file from the first request
@@ -240,7 +238,7 @@ public class FileDownloaderTest {
         Duration timeout = Duration.ofMillis(200);
         MockConnection connectionPool = new MockConnection();
         connectionPool.setResponseHandler(new MockConnection.WaitResponseHandler(timeout.plus(Duration.ofMillis(1000))));
-        FileDownloader fileDownloader = new FileDownloader(connectionPool, supervisor, downloadDir, downloads, timeout, sleepBetweenRetries);
+        FileDownloader fileDownloader = new FileDownloader(connectionPool, supervisor, downloadDir, timeout, sleepBetweenRetries);
         FileReference xyzzy = new FileReference("xyzzy");
         // Should download since we do not have the file on disk
         fileDownloader.downloadIfNeeded(new FileReferenceDownload(xyzzy));
@@ -275,6 +273,7 @@ public class FileDownloaderTest {
     }
 
     private void assertDownloadStatus(FileReference fileReference, double expectedDownloadStatus) {
+        Downloads downloads = fileDownloader.downloads();
         double downloadStatus = downloads.downloadStatus(fileReference);
         assertEquals("Download statuses: " + downloads.downloadStatuses().toString(),
                      expectedDownloadStatus,
@@ -293,7 +292,7 @@ public class FileDownloaderTest {
                 new FileReceiver.Session(downloadDir, 1, fileReference, type, filename, content.length);
         session.addPart(0, content);
         File file = session.close(hasher.hash(ByteBuffer.wrap(content), 0));
-        downloads.completedDownloading(fileReference, file);
+        fileDownloader.downloads().completedDownloading(fileReference, file);
     }
 
     private static class MockConnection implements ConnectionPool, com.yahoo.vespa.config.Connection {