asynchronous API

make API calls to set configuration asynchronous

fixes #3678
fixes #3686
fixes #3771

Author: vladak

apiary.apib

@@ -8,6 +8,10 @@ Besides `/suggester`, `/search` and `/metrics` endpoints, everything is accessib
 unless authentication bearer tokens are configured in the web application and used via the 'Authorization' HTTP header
 (within HTTPS connection).
 
+Some of the APIs are asynchronous. They return status code 202 (Accepted) and a Location header that contains the URL
+for the status endpoint to check for the result of the API call. Once the status API returns result other than 202,
+the client should issue DELETE request to this URL to clean up server resources.
+
 ## Annotation [/annotation{?path}]
 
 ### Get annotation for a file [GET]
@@ -37,6 +41,8 @@ unless authentication bearer tokens are configured in the web application and us
 
 ### reloads authorization framework [POST]
 
+This is asynchronous API endpoint.
+
 + Request (application/text)
 
 + Response 204
@@ -61,6 +67,8 @@ unless authentication bearer tokens are configured in the web application and us
 
 ### sets configuration from XML representation [PUT]
 
+This is asynchronous API endpoint.
+
 + Request (application/xml)
   + Body
 
@@ -85,6 +93,8 @@ unless authentication bearer tokens are configured in the web application and us
 
 ### sets specific configuration field [PUT]
 
+This is asynchronous API endpoint.
+
 + Parameters
   + reindex (boolean) - specifies if the underlying data were also reindexed (refreshes some searchers and additional data structures)
 
@@ -577,3 +587,22 @@ This kicks off suggester data rebuild in the background, i.e. the rebuild will v
   + project - project name
 
 + Response 204
+
+## Status [/status/{uuid}]
+
+### Check the state of API request [GET]
+
++ Parameters
+  + uuid
+
++ Response 202
+
+### Delete state associated with API request tracking [DELETE]
+
+This should be done only after the API request is completed, i.e. after the GET request for the API request state
+returns appropriate status code (e.g. 201).
+
++ Parameters
+  + uuid
+
++ Response 200

opengrok-indexer/src/main/java/org/opengrok/indexer/configuration/Configuration.java

@@ -297,6 +297,9 @@ public final class Configuration {
 
     private String serverName;  // for reverse proxy environment
 
+    private int connectTimeout = -1;    // connect timeout in seconds
+    private int apiTimeout = -1;    // API timeout in seconds
+
     /*
      * types of handling history for remote SCM repositories:
      *  ON - index history and display it in webapp
@@ -577,6 +580,8 @@ public Configuration() {
         setWebappLAF("default");
         // webappCtags is default(boolean)
         setXrefTimeout(30);
+        setApiTimeout(300); // 5 minutes
+        setConnectTimeout(3);
     }
 
     public String getRepoCmd(String clazzName) {
@@ -1385,6 +1390,28 @@ public void setServerName(String serverName) {
         this.serverName = serverName;
     }
 
+    public int getConnectTimeout() {
+        return connectTimeout;
+    }
+
+    public void setConnectTimeout(int connectTimeout) {
+        if (connectTimeout < 0) {
+            throw new IllegalArgumentException(String.format(NEGATIVE_NUMBER_ERROR, "connectTimeout", connectTimeout));
+        }
+        this.connectTimeout = connectTimeout;
+    }
+
+    public int getApiTimeout() {
+        return apiTimeout;
+    }
+
+    public void setApiTimeout(int apiTimeout) {
+        if (apiTimeout < 0) {
+            throw new IllegalArgumentException(String.format(NEGATIVE_NUMBER_ERROR, "apiTimeout", apiTimeout));
+        }
+        this.apiTimeout = apiTimeout;
+    }
+
     /**
      * Write the current configuration to a file.
      *

opengrok-indexer/src/main/java/org/opengrok/indexer/configuration/RuntimeEnvironment.java

@@ -53,6 +53,7 @@
 
 import jakarta.ws.rs.client.ClientBuilder;
 import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.Response;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.MultiReader;
@@ -61,6 +62,7 @@
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
 import org.apache.lucene.util.NamedThreadFactory;
+import org.jetbrains.annotations.NotNull;
 import org.opengrok.indexer.authorization.AuthorizationFramework;
 import org.opengrok.indexer.authorization.AuthorizationStack;
 import org.opengrok.indexer.history.HistoryGuru;
@@ -1399,6 +1401,22 @@ public void setServerName(String serverName) {
         syncWriteConfiguration(serverName, Configuration::setServerName);
     }
 
+    public int getApiTimeout() {
+        return syncReadConfiguration(Configuration::getApiTimeout);
+    }
+
+    public void setApiTimeout(int apiTimeout) {
+        syncWriteConfiguration(apiTimeout, Configuration::setApiTimeout);
+    }
+
+    public int getConnectTimeout() {
+        return syncReadConfiguration(Configuration::getConnectTimeout);
+    }
+
+    public void setConnectTimeout(int connectTimeout) {
+        syncWriteConfiguration(connectTimeout, Configuration::setConnectTimeout);
+    }
+
     /**
      * Read an configuration file and set it as the current configuration.
      *
@@ -1440,15 +1458,62 @@ public String getConfigurationXML() {
     }
 
     /**
-     * Write the current configuration to a socket.
+     * Busy waits for API call to complete by repeatedly querying the status API endpoint passed
+     * in the {@code Location} header in the response parameter. The overall time is governed
+     * by the {@link #getApiTimeout()}, however each individual status check
+     * uses {@link #getConnectTimeout()} so in the worst case the total time can be
+     * {@code getApiTimeout() * getConnectTimeout()}.
+     * @param response response returned from the server upon asynchronous API request
+     * @return response from the status API call
+     * @throws InterruptedException on sleep interruption
+     * @throws IllegalArgumentException on invalid request (no {@code Location} header)
+     */
+    private @NotNull Response waitForAsyncApi(@NotNull Response response)
+            throws InterruptedException, IllegalArgumentException {
+
+        String location = response.getHeaderString(HttpHeaders.LOCATION);
+        if (location == null) {
+            throw new IllegalArgumentException(String.format("no %s header in %s", HttpHeaders.LOCATION, response));
+        }
+
+        LOGGER.log(Level.FINER, "checking asynchronous API result on {0}", location);
+        for (int i = 0; i < getApiTimeout(); i++) {
+            response = ClientBuilder.newBuilder().
+                    connectTimeout(RuntimeEnvironment.getInstance().getConnectTimeout(), TimeUnit.SECONDS).build().
+                    target(location).request().get();
+            if (response.getStatus() == Response.Status.ACCEPTED.getStatusCode()) {
+                Thread.sleep(1000);
+            } else {
+                break;
+            }
+        }
+
+        if (response.getStatus() == Response.Status.ACCEPTED.getStatusCode()) {
+            LOGGER.log(Level.WARNING, "API request still not completed: {0}", response);
+            return response;
+        }
+
+        LOGGER.log(Level.FINER, "making DELETE API request to {0}", location);
+        Response deleteResponse = ClientBuilder.newBuilder().connectTimeout(3, TimeUnit.SECONDS).build().
+                target(location).request().delete();
+        if (deleteResponse.getStatusInfo().getFamily() != Response.Status.Family.SUCCESSFUL) {
+            LOGGER.log(Level.WARNING, "DELETE API call to {0} failed with HTTP error {1}",
+                    new Object[]{location, response.getStatusInfo()});
+        }
+
+        return response;
+    }
+
+    /**
+     * Write the current configuration to a socket and waits for the result.
      *
      * @param host the host address to receive the configuration
      * @throws IOException if an error occurs
      */
-    public void writeConfiguration(String host) throws IOException {
+    public void writeConfiguration(String host) throws IOException, InterruptedException, IllegalArgumentException {
         String configXML = syncReadConfiguration(Configuration::getXMLRepresentationAsString);
 
-        Response r = ClientBuilder.newClient()
+        Response response = ClientBuilder.newClient()
                 .target(host)
                 .path("api")
                 .path("v1")
@@ -1458,8 +1523,12 @@ public void writeConfiguration(String host) throws IOException {
                 .headers(getWebAppHeaders())
                 .put(Entity.xml(configXML));
 
-        if (r.getStatusInfo().getFamily() != Response.Status.Family.SUCCESSFUL) {
-            throw new IOException(r.toString());
+        if (response.getStatus() == Response.Status.ACCEPTED.getStatusCode()) {
+            response = waitForAsyncApi(response);
+        }
+
+        if (response.getStatusInfo().getFamily() != Response.Status.Family.SUCCESSFUL) {
+            throw new IOException(response.toString());
         }
     }
 

opengrok-indexer/src/main/java/org/opengrok/indexer/index/IndexDatabase.java

@@ -48,6 +48,7 @@
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -378,7 +379,7 @@ private void markProjectIndexed(Project project) {
 
         Response r;
         try {
-            r = ClientBuilder.newClient()
+            r = ClientBuilder.newBuilder().connectTimeout(env.getConnectTimeout(), TimeUnit.SECONDS).build()
                     .target(env.getConfigURI())
                     .path("api")
                     .path("v1")

opengrok-indexer/src/main/java/org/opengrok/indexer/index/Indexer.java

@@ -468,6 +468,12 @@ public static String[] parseOptions(String[] argv) throws ParseException {
                         }
             });
 
+            parser.on("--apiTimeout", "=number", Integer.class,
+                    "Set timeout for asynchronous API requests.").execute(v -> cfg.setApiTimeout((Integer) v));
+
+            parser.on("--connectTimeout", "=number", Integer.class,
+                    "Set connect timeout. Used for API requests.").execute(v -> cfg.setConnectTimeout((Integer) v));
+
             parser.on(
                 "-A (.ext|prefix.):(-|analyzer)", "--analyzer",
                     "/(\\.\\w+|\\w+\\.):(-|[a-zA-Z_0-9.]+)/",
@@ -1154,10 +1160,12 @@ public void sendToConfigHost(RuntimeEnvironment env, String host) {
         LOGGER.log(Level.INFO, "Sending configuration to: {0}", host);
         try {
             env.writeConfiguration(host);
-        } catch (IOException ex) {
+        } catch (IOException | IllegalArgumentException ex) {
             LOGGER.log(Level.SEVERE, String.format(
                     "Failed to send configuration to %s "
                     + "(is web application server running with opengrok deployed?)", host), ex);
+        } catch (InterruptedException e) {
+            LOGGER.log(Level.WARNING, "interrupted while sending configuration");
         }
         LOGGER.info("Configuration update routine done, check log output for errors.");
     }

opengrok-web/src/main/java/org/opengrok/web/WebappListener.java

@@ -40,6 +40,9 @@
 import org.opengrok.indexer.index.IndexDatabase;
 import org.opengrok.indexer.logger.LoggerFactory;
 import org.opengrok.indexer.web.SearchHelper;
+import org.opengrok.web.api.ApiTaskManager;
+import org.opengrok.web.api.v1.controller.ConfigurationController;
+import org.opengrok.web.api.v1.controller.ProjectsController;
 import org.opengrok.web.api.v1.suggester.provider.service.SuggesterServiceFactory;
 
 import java.io.File;
@@ -108,6 +111,14 @@ public void contextInitialized(final ServletContextEvent servletContextEvent) {
         checkIndex(env);
 
         env.startExpirationTimer();
+
+        ApiTaskManager.getInstance().setContextPath(context.getContextPath());
+        // register API task queues
+        ApiTaskManager.getInstance().addPool(ProjectsController.PROJECTS_PATH, 1);
+        // Used by ConfigurationController#reloadAuthorization()
+        ApiTaskManager.getInstance().addPool("authorization", 1);
+        ApiTaskManager.getInstance().addPool(ConfigurationController.PATH, 1);
+
         startupTimer.record(Duration.between(start, Instant.now()));
     }
 
@@ -163,6 +174,13 @@ public void contextDestroyed(final ServletContextEvent servletContextEvent) {
         // need to explicitly close the suggester service because it might have scheduled rebuild which could prevent
         // the web application from closing
         SuggesterServiceFactory.getDefault().close();
+
+        // destroy queue(s) of API tasks
+        try {
+            ApiTaskManager.getInstance().shutdown();
+        } catch (InterruptedException e) {
+            LOGGER.log(Level.WARNING, "could not shutdown API task manager cleanly", e);
+        }
     }
 
     /**