Revert "Revert "Create local session based on remote session and distributed applicat…""

5 files changed, 99 insertions, 29 deletions

diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/application/TenantApplications.java b/configserver/src/main/java/com/yahoo/vespa/config/server/application/TenantApplications.java
index 22de36e98aa..a4dfec708d6 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/application/TenantApplications.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/application/TenantApplications.java
@@ -18,6 +18,7 @@ import com.yahoo.vespa.config.server.NotFoundException;
 import com.yahoo.vespa.config.server.ReloadHandler;
 import com.yahoo.vespa.config.server.ReloadListener;
 import com.yahoo.vespa.config.server.RequestHandler;
+import com.yahoo.vespa.config.server.deploy.TenantFileSystemDirs;
 import com.yahoo.vespa.config.server.host.HostRegistry;
 import com.yahoo.vespa.config.server.host.HostValidator;
 import com.yahoo.vespa.config.server.monitoring.MetricUpdater;
@@ -31,6 +32,8 @@ import com.yahoo.vespa.curator.transaction.CuratorTransaction;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
 
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.time.Clock;
 import java.time.Duration;
 import java.util.Collection;
@@ -73,10 +76,12 @@ public class TenantApplications implements RequestHandler, ReloadHandler, HostVa
     private final ApplicationMapper applicationMapper = new ApplicationMapper();
     private final MetricUpdater tenantMetricUpdater;
     private final Clock clock = Clock.systemUTC();
+    private final TenantFileSystemDirs tenantFileSystemDirs;
 
     public TenantApplications(TenantName tenant, Curator curator, StripedExecutor<TenantName> zkWatcherExecutor,
                               ExecutorService zkCacheExecutor, Metrics metrics, ReloadListener reloadListener,
-                              ConfigserverConfig configserverConfig, HostRegistry<ApplicationId> hostRegistry) {
+                              ConfigserverConfig configserverConfig, HostRegistry<ApplicationId> hostRegistry,
+                              TenantFileSystemDirs tenantFileSystemDirs) {
         this.curator = curator;
         this.applicationsPath = TenantRepository.getApplicationsPath(tenant);
         this.locksPath = TenantRepository.getLocksPath(tenant);
@@ -90,6 +95,7 @@ public class TenantApplications implements RequestHandler, ReloadHandler, HostVa
         this.responseFactory = ConfigResponseFactory.create(configserverConfig);
         this.tenantMetricUpdater = metrics.getOrCreateMetricUpdater(Metrics.createDimensions(tenant));
         this.hostRegistry = hostRegistry;
+        this.tenantFileSystemDirs = tenantFileSystemDirs;
     }
 
     // For testing only
@@ -101,7 +107,8 @@ public class TenantApplications implements RequestHandler, ReloadHandler, HostVa
                                       componentRegistry.getMetrics(),
                                       componentRegistry.getReloadListener(),
                                       componentRegistry.getConfigserverConfig(),
-                                      componentRegistry.getHostRegistries().createApplicationHostRegistry(tenantName));
+                                      componentRegistry.getHostRegistries().createApplicationHostRegistry(tenantName),
+                                      new TenantFileSystemDirs(componentRegistry.getConfigServerDB(), tenantName));
     }
 
     /**
@@ -128,6 +135,10 @@ public class TenantApplications implements RequestHandler, ReloadHandler, HostVa
         return data.isEmpty() ? Optional.empty() : Optional.of(Long.parseLong(data));
     }
 
+    public boolean hasLocalSession(long sessionId) {
+        return Files.exists(Paths.get(tenantFileSystemDirs.sessionsPath().getAbsolutePath(), String.valueOf(sessionId)));
+    }
+
     /**
      * Returns a transaction which writes the given session id as the currently active for the given application.
      *
@@ -237,12 +248,6 @@ public class TenantApplications implements RequestHandler, ReloadHandler, HostVa
         return application.resolveConfig(req, responseFactory);
     }
 
-    // For testing only
-    long getApplicationGeneration(ApplicationId appId, Optional<Version> vespaVersion) {
-        Application application = getApplication(appId, vespaVersion);
-        return application.getApplicationGeneration();
-    }
-
     private void notifyReloadListeners(ApplicationSet applicationSet) {
         reloadListener.hostsUpdated(tenant, hostRegistry.getAllHosts());
         reloadListener.configActivated(applicationSet);
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileDirectory.java b/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileDirectory.java
index ab7509a49aa..0865b72dbbf 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileDirectory.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/filedistribution/FileDirectory.java
@@ -55,7 +55,7 @@ public class FileDirectory  {
         return root.getAbsolutePath() + "/" + ref.value();
     }
 
-    File getFile(FileReference reference) {
+    public File getFile(FileReference reference) {
         ensureRootExist();
         File dir = new File(getPath(reference));
         if (!dir.exists()) {
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/session/RemoteSessionRepo.java b/configserver/src/main/java/com/yahoo/vespa/config/server/session/RemoteSessionRepo.java
index 316f7f7778d..0e538b05931 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/session/RemoteSessionRepo.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/session/RemoteSessionRepo.java
@@ -18,6 +18,9 @@ import com.yahoo.vespa.config.server.monitoring.Metrics;
 import com.yahoo.vespa.config.server.tenant.TenantRepository;
 import com.yahoo.vespa.config.server.zookeeper.ConfigCurator;
 import com.yahoo.vespa.curator.Curator;
+import com.yahoo.vespa.flags.BooleanFlag;
+import com.yahoo.vespa.flags.FlagSource;
+import com.yahoo.vespa.flags.Flags;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.recipes.cache.ChildData;
 import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
@@ -53,6 +56,7 @@ public class RemoteSessionRepo {
     private final ReloadHandler reloadHandler;
     private final TenantName tenantName;
     private final MetricUpdater metrics;
+    private final BooleanFlag distributeApplicationPackage;
     private final Curator.DirectoryCache directoryCache;
     private final TenantApplications applicationRepo;
     private final Executor zkWatcherExecutor;
@@ -62,7 +66,8 @@ public class RemoteSessionRepo {
                              SessionFactory sessionFactory,
                              ReloadHandler reloadHandler,
                              TenantName tenantName,
-                             TenantApplications applicationRepo) {
+                             TenantApplications applicationRepo,
+                             FlagSource flagSource) {
         this.sessionCache = new SessionCache<>();
         this.curator = componentRegistry.getCurator();
         this.sessionsPath = TenantRepository.getSessionsPath(tenantName);
@@ -71,6 +76,7 @@ public class RemoteSessionRepo {
         this.reloadHandler = reloadHandler;
         this.tenantName = tenantName;
         this.metrics = componentRegistry.getMetrics().getOrCreateMetricUpdater(Metrics.createDimensions(tenantName));
+        this.distributeApplicationPackage = Flags.CONFIGSERVER_DISTRIBUTE_APPLICATION_PACKAGE.bindTo(flagSource);
         StripedExecutor<TenantName> zkWatcherExecutor = componentRegistry.getZkWatcherExecutor();
         this.zkWatcherExecutor = command -> zkWatcherExecutor.execute(tenantName, command);
         initializeSessions();
@@ -89,6 +95,7 @@ public class RemoteSessionRepo {
 
     public void addSession(RemoteSession session) {
         sessionCache.addSession(session);
+        metrics.incAddedSessions();
     }
 
     public int deleteExpiredSessions(Clock clock, Duration expiryTime) {
@@ -155,8 +162,9 @@ public class RemoteSessionRepo {
         fileCache.addListener(this::nodeChanged);
         loadSessionIfActive(session);
         addSession(session);
-        metrics.incAddedSessions();
         sessionStateWatchers.put(sessionId, new RemoteSessionStateWatcher(fileCache, reloadHandler, session, metrics, zkWatcherExecutor));
+        if (distributeApplicationPackage.value())
+            sessionFactory.createLocalSessionUsingDistributedApplicationPackage(sessionId);
     }
 
     private void sessionRemoved(long sessionId) {
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionFactory.java b/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionFactory.java
index fc4071916ed..337e9b3e99c 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionFactory.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/session/SessionFactory.java
@@ -1,6 +1,7 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.config.server.session;
 
+import com.yahoo.config.FileReference;
 import com.yahoo.config.application.api.ApplicationPackage;
 import com.yahoo.config.application.api.DeployLogger;
 import com.yahoo.config.model.application.provider.DeployData;
@@ -14,15 +15,18 @@ import com.yahoo.vespa.config.server.GlobalComponentRegistry;
 import com.yahoo.vespa.config.server.TimeoutBudget;
 import com.yahoo.vespa.config.server.application.TenantApplications;
 import com.yahoo.vespa.config.server.deploy.TenantFileSystemDirs;
+import com.yahoo.vespa.config.server.filedistribution.FileDirectory;
 import com.yahoo.vespa.config.server.host.HostValidator;
 import com.yahoo.vespa.config.server.tenant.TenantRepository;
 import com.yahoo.vespa.config.server.zookeeper.ConfigCurator;
 import com.yahoo.vespa.config.server.zookeeper.SessionCounter;
 import com.yahoo.vespa.curator.Curator;
+import com.yahoo.vespa.defaults.Defaults;
 import com.yahoo.vespa.flags.BooleanFlag;
 import com.yahoo.vespa.flags.Flags;
 
 import java.io.File;
+import java.io.IOException;
 import java.time.Clock;
 import java.util.List;
 import java.util.Optional;
@@ -31,7 +35,7 @@ import java.util.logging.Logger;
 
 /**
  * Serves as the factory of sessions. Takes care of copying files to the correct folder and initializing the
- * session state.
+ * session state. There is one SessionFactory per tenant.
  *
  * @author Ulf Lilleengen
  */
@@ -86,8 +90,7 @@ public class SessionFactory {
     }
 
     public RemoteSession createRemoteSession(long sessionId) {
-        Path sessionPath = sessionsPath.append(String.valueOf(sessionId));
-        SessionZooKeeperClient sessionZKClient = createSessionZooKeeperClient(sessionPath);
+        SessionZooKeeperClient sessionZKClient = createSessionZooKeeperClient(getSessionPath(sessionId));
         return new RemoteSession(tenant, sessionId, componentRegistry, sessionZKClient);
     }
 
@@ -115,10 +118,10 @@ public class SessionFactory {
 
     private LocalSession createSessionFromApplication(ApplicationPackage applicationPackage,
                                                       long sessionId,
-                                                      SessionZooKeeperClient sessionZKClient,
                                                       TimeoutBudget timeoutBudget,
                                                       Clock clock) {
         log.log(Level.FINE, TenantRepository.logPre(tenant) + "Creating session " + sessionId + " in ZooKeeper");
+        SessionZooKeeperClient sessionZKClient = createSessionZooKeeperClient(getSessionPath(sessionId));
         sessionZKClient.createNewSession(clock.instant());
         Curator.CompletionWaiter waiter = sessionZKClient.getUploadWaiter();
         LocalSession session = new LocalSession(tenant, sessionId, sessionPreparer, applicationPackage, sessionZKClient,
@@ -161,23 +164,47 @@ public class SessionFactory {
                                 boolean internalRedeploy, TimeoutBudget timeoutBudget) {
         long sessionId = getNextSessionId();
         try {
-            ensureSessionPathDoesNotExist(sessionId);
+            ApplicationPackage app = createApplicationPackage(applicationFile, applicationId,
+                                                              sessionId, currentlyActiveSessionId, internalRedeploy);
+            return createSessionFromApplication(app, sessionId, timeoutBudget, clock);
+        } catch (Exception e) {
+            throw new RuntimeException("Error creating session " + sessionId, e);
+        }
+    }
+
+    /**
+     * This method is used when creating a session based on a remote session and the distributed application package
+     * It does not wait for session being created on other servers
+     */
+    private LocalSession createLocalSession(File applicationFile, ApplicationId applicationId,
+                                            long sessionId, long currentlyActiveSessionId) {
+        try {
+            ApplicationPackage applicationPackage = createApplicationPackage(applicationFile, applicationId,
+                                                                             sessionId, currentlyActiveSessionId, false);
             SessionZooKeeperClient sessionZooKeeperClient = createSessionZooKeeperClient(getSessionPath(sessionId));
-            File userApplicationDir = getSessionAppDir(sessionId);
-            IOUtils.copyDirectory(applicationFile, userApplicationDir);
-            ApplicationPackage applicationPackage = createApplication(applicationFile,
-                                                                      userApplicationDir,
-                                                                      applicationId,
-                                                                      sessionId,
-                                                                      currentlyActiveSessionId,
-                                                                      internalRedeploy);
-            applicationPackage.writeMetaData();
-            return createSessionFromApplication(applicationPackage, sessionId, sessionZooKeeperClient, timeoutBudget, clock);
+            return new LocalSession(tenant, sessionId, sessionPreparer, applicationPackage, sessionZooKeeperClient,
+                                    getSessionAppDir(sessionId), applicationRepo, hostRegistry);
         } catch (Exception e) {
             throw new RuntimeException("Error creating session " + sessionId, e);
         }
     }
 
+    private ApplicationPackage createApplicationPackage(File applicationFile, ApplicationId applicationId,
+                                                        long sessionId, long currentlyActiveSessionId,
+                                                        boolean internalRedeploy) throws IOException {
+        ensureSessionPathDoesNotExist(sessionId);
+        File userApplicationDir = getSessionAppDir(sessionId);
+        IOUtils.copyDirectory(applicationFile, userApplicationDir);
+        ApplicationPackage applicationPackage = createApplication(applicationFile,
+                                                                  userApplicationDir,
+                                                                  applicationId,
+                                                                  sessionId,
+                                                                  currentlyActiveSessionId,
+                                                                  internalRedeploy);
+        applicationPackage.writeMetaData();
+        return applicationPackage;
+    }
+
     /**
      * Returns a new session instance for the given session id.
      */
@@ -190,6 +217,34 @@ public class SessionFactory {
                                 getSessionAppDir(sessionId), applicationRepo, hostRegistry);
     }
 
+    /**
+     * Returns a new session instance for the given session id.
+     */
+    LocalSession createLocalSessionUsingDistributedApplicationPackage(long sessionId) {
+        if (applicationRepo.hasLocalSession(sessionId)) {
+            log.log(Level.FINE, "Local session for session id " + sessionId + " already exists");
+            return createSessionFromId(sessionId);
+        }
+
+        log.log(Level.INFO, "Creating local session for session id " + sessionId);
+        SessionZooKeeperClient sessionZKClient = createSessionZooKeeperClient(getSessionPath(sessionId));
+        FileReference fileReference = sessionZKClient.readApplicationPackageReference();
+        log.log(Level.FINE, "File reference for session id " + sessionId + ": " + fileReference);
+        if (fileReference != null) {
+            File rootDir = new File(Defaults.getDefaults().underVespaHome(componentRegistry.getConfigserverConfig().fileReferencesDir()));
+            File sessionDir = new FileDirectory(rootDir).getFile(fileReference);
+            if (!sessionDir.exists())
+                throw new RuntimeException("File reference for session " + sessionId + " not found (" + sessionDir.getAbsolutePath() + ")");
+            ApplicationId applicationId = sessionZKClient.readApplicationId();
+            return createLocalSession(sessionDir,
+                                      applicationId,
+                                      sessionId,
+                                      applicationRepo.activeSessionOf(applicationId).orElse(nonExistingActiveSession));
+        }
+        return null;
+    }
+
+    // Return Optional instead of faking it with nonExistingActiveSession
     private long getActiveSessionId(ApplicationId applicationId) {
         List<ApplicationId> applicationIds = applicationRepo.activeApplications();
         if (applicationIds.contains(applicationId)) {
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java b/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java
index 90a03153d30..d34f89a179b 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/tenant/TenantRepository.java
@@ -12,6 +12,7 @@ import com.yahoo.vespa.config.server.GlobalComponentRegistry;
 import com.yahoo.vespa.config.server.ReloadHandler;
 import com.yahoo.vespa.config.server.RequestHandler;
 import com.yahoo.vespa.config.server.application.TenantApplications;
+import com.yahoo.vespa.config.server.deploy.TenantFileSystemDirs;
 import com.yahoo.vespa.config.server.monitoring.MetricUpdater;
 import com.yahoo.vespa.config.server.session.LocalSessionRepo;
 import com.yahoo.vespa.config.server.session.RemoteSessionRepo;
@@ -216,7 +217,8 @@ public class TenantRepository {
                                        componentRegistry.getMetrics(),
                                        componentRegistry.getReloadListener(),
                                        componentRegistry.getConfigserverConfig(),
-                                       componentRegistry.getHostRegistries().createApplicationHostRegistry(tenantName));
+                                       componentRegistry.getHostRegistries().createApplicationHostRegistry(tenantName),
+                                       new TenantFileSystemDirs(componentRegistry.getConfigServerDB(), tenantName));
         if (requestHandler == null)
             requestHandler = applicationRepo;
         if (reloadHandler == null)
@@ -227,7 +229,8 @@ public class TenantRepository {
                                                                     sessionFactory,
                                                                     reloadHandler,
                                                                     tenantName,
-                                                                    applicationRepo);
+                                                                    applicationRepo,
+                                                                    componentRegistry.getFlagSource());
         log.log(Level.INFO, "Creating tenant '" + tenantName + "'");
         Tenant tenant = new Tenant(tenantName, sessionFactory, localSessionRepo, remoteSessionRepo, requestHandler,
                                    reloadHandler, applicationRepo, componentRegistry.getCurator());
@@ -244,7 +247,6 @@ public class TenantRepository {
         return tenants.get(DEFAULT_TENANT);
     }
 
-
     private void removeUnusedApplications() {
         getAllTenants().forEach(tenant -> tenant.getApplicationRepo().removeUnusedApplications());
     }