report progress when initializing/rebuilding suggester (#4343)

vladak · web-flow · commit 057e8ad0517b · 2023-05-10T13:45:29.000+02:00
diff --git a/opengrok-web/src/main/java/org/opengrok/web/api/v1/suggester/provider/service/impl/SuggesterServiceImpl.java b/opengrok-web/src/main/java/org/opengrok/web/api/v1/suggester/provider/service/impl/SuggesterServiceImpl.java
@@ -320,7 +320,8 @@ private void initSuggester() {
                 suggesterConfig.getAllowedFields(),
                 suggesterConfig.getTimeThreshold(),
                 rebuildParalleismLevel,
-                Metrics.getRegistry());
+                Metrics.getRegistry(),
+                env.isPrintProgress());
 
         new Thread(() -> {
             suggester.init(getAllProjectIndexDirs());
diff --git a/suggester/src/main/java/org/opengrok/suggest/Suggester.java b/suggester/src/main/java/org/opengrok/suggest/Suggester.java
@@ -18,7 +18,7 @@
  */
 
 /*
- * Copyright (c) 2018, 2022, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2018, 2023, Oracle and/or its affiliates. All rights reserved.
  */
 package org.opengrok.suggest;
 
@@ -34,6 +34,7 @@
 import org.apache.lucene.util.BytesRef;
 import org.opengrok.suggest.query.SuggesterPrefixQuery;
 import org.opengrok.suggest.query.SuggesterQuery;
+import org.opengrok.suggest.util.Progress;
 
 import java.io.Closeable;
 import java.io.File;
@@ -93,6 +94,7 @@ public final class Suggester implements Closeable {
     private final int timeThreshold;
 
     private final int rebuildParallelismLevel;
+    private final boolean isPrintProgress;
 
     private volatile boolean rebuilding;
     private volatile boolean terminating;
@@ -123,7 +125,9 @@ public final class Suggester implements Closeable {
      * @param allowedFields fields for which should the suggester be enabled,
      * if {@code null} then enabled for all fields
      * @param timeThreshold time in milliseconds after which the suggestions requests should time out
+     * @param rebuildParallelismLevel parallelism level for rebuild
      * @param registry meter registry
+     * @param isPrintProgress whether to report progress for initialization and rebuild
      */
     public Suggester(
             final File suggesterDir,
@@ -134,7 +138,8 @@ public Suggester(
             final Set<String> allowedFields,
             final int timeThreshold,
             final int rebuildParallelismLevel,
-            MeterRegistry registry) {
+            MeterRegistry registry,
+            boolean isPrintProgress) {
         if (suggesterDir == null) {
             throw new IllegalArgumentException("Suggester needs to have directory specified");
         }
@@ -152,6 +157,7 @@ public Suggester(
         this.allowedFields = new HashSet<>(allowedFields);
         this.timeThreshold = timeThreshold;
         this.rebuildParallelismLevel = rebuildParallelismLevel;
+        this.isPrintProgress = isPrintProgress;
 
         suggesterRebuildTimer = Timer.builder("suggester.rebuild.latency").
                 description("suggester rebuild latency").
@@ -180,17 +186,20 @@ public void init(final Collection<NamedIndexDir> luceneIndexes) {
 
             ExecutorService executor = Executors.newWorkStealingPool(rebuildParallelismLevel);
 
-            for (NamedIndexDir indexDir : luceneIndexes) {
-                if (terminating) {
-                    LOGGER.log(Level.INFO, "Terminating suggester initialization");
-                    return;
+            try (Progress progress = new Progress(LOGGER, "suggester initialization", luceneIndexes.size(),
+                    Level.INFO, isPrintProgress)) {
+                for (NamedIndexDir indexDir : luceneIndexes) {
+                    if (terminating) {
+                        LOGGER.log(Level.INFO, "Terminating suggester initialization");
+                        return;
+                    }
+                    submitInitIfIndexExists(executor, indexDir, progress);
                 }
-                submitInitIfIndexExists(executor, indexDir);
-            }
 
-            shutdownAndAwaitTermination(executor, start, suggesterInitTimer,
-                    "Suggester successfully initialized");
-            initDone.countDown();
+                shutdownAndAwaitTermination(executor, start, suggesterInitTimer,
+                        "Suggester successfully initialized");
+                initDone.countDown();
+            }
         }
     }
 
@@ -206,10 +215,11 @@ public void waitForInit(long timeout, TimeUnit unit) throws InterruptedException
         }
     }
 
-    private void submitInitIfIndexExists(final ExecutorService executorService, final NamedIndexDir indexDir) {
+    private void submitInitIfIndexExists(final ExecutorService executorService, final NamedIndexDir indexDir,
+                                         Progress progress) {
         try {
             if (indexExists(indexDir.path)) {
-                executorService.submit(getInitRunnable(indexDir));
+                executorService.submit(getInitRunnable(indexDir, progress));
             } else {
                 LOGGER.log(Level.FINE, "Index in {0} directory does not exist, skipping...", indexDir);
             }
@@ -218,7 +228,7 @@ private void submitInitIfIndexExists(final ExecutorService executorService, fina
         }
     }
 
-    private Runnable getInitRunnable(final NamedIndexDir indexDir) {
+    private Runnable getInitRunnable(final NamedIndexDir indexDir, Progress progress) {
         return () -> {
             try {
                 if (terminating) {
@@ -239,6 +249,7 @@ private Runnable getInitRunnable(final NamedIndexDir indexDir) {
 
                 Duration d = Duration.between(start, Instant.now());
                 LOGGER.log(Level.FINE, "Finished initialization of {0}, took {1}", new Object[] {indexDir, d});
+                progress.increment();
             } catch (Exception e) {
                 LOGGER.log(Level.SEVERE, String.format("Could not initialize suggester data for %s", indexDir), e);
             }
@@ -300,17 +311,20 @@ public void rebuild(final Collection<NamedIndexDir> indexDirs) {
 
             ExecutorService executor = Executors.newWorkStealingPool(rebuildParallelismLevel);
 
-            for (NamedIndexDir indexDir : indexDirs) {
-                SuggesterProjectData data = this.projectData.get(indexDir.name);
-                if (data != null) {
-                    executor.submit(getRebuildRunnable(data));
-                } else {
-                    submitInitIfIndexExists(executor, indexDir);
+            try (Progress progress = new Progress(LOGGER, "suggester rebuild", indexDirs.size(),
+                    Level.INFO, isPrintProgress)) {
+                for (NamedIndexDir indexDir : indexDirs) {
+                    SuggesterProjectData data = this.projectData.get(indexDir.name);
+                    if (data != null) {
+                        executor.submit(getRebuildRunnable(data, progress));
+                    } else {
+                        submitInitIfIndexExists(executor, indexDir, progress);
+                    }
                 }
-            }
 
-            shutdownAndAwaitTermination(executor, start, suggesterRebuildTimer,
-                    "Suggesters for " + indexDirs + " were successfully rebuilt");
+                shutdownAndAwaitTermination(executor, start, suggesterRebuildTimer,
+                        "Suggesters for " + indexDirs + " were successfully rebuilt");
+            }
         }
 
         rebuildLock.lock();
@@ -341,7 +355,7 @@ public void waitForRebuild(long timeout, TimeUnit unit) throws InterruptedExcept
         }
     }
 
-    private Runnable getRebuildRunnable(final SuggesterProjectData data) {
+    private Runnable getRebuildRunnable(final SuggesterProjectData data, Progress progress) {
         return () -> {
             try {
                 if (terminating) {
@@ -354,6 +368,7 @@ private Runnable getRebuildRunnable(final SuggesterProjectData data) {
 
                 Duration d = Duration.between(start, Instant.now());
                 LOGGER.log(Level.FINE, "Rebuild of {0} finished, took {1}", new Object[] {data, d});
+                progress.increment();
             } catch (Exception e) {
                 LOGGER.log(Level.SEVERE, "Could not rebuild suggester", e);
             }
diff --git a/suggester/src/main/java/org/opengrok/suggest/util/Progress.java b/suggester/src/main/java/org/opengrok/suggest/util/Progress.java
@@ -0,0 +1,214 @@
+/*
+ * CDDL HEADER START
+ *
+ * The contents of this file are subject to the terms of the
+ * Common Development and Distribution License (the "License").
+ * You may not use this file except in compliance with the License.
+ *
+ * See LICENSE.txt included in this distribution for the specific
+ * language governing permissions and limitations under the License.
+ *
+ * When distributing Covered Code, include this CDDL HEADER in each
+ * file and include the License file at LICENSE.txt.
+ * If applicable, add the following below this CDDL HEADER, with the
+ * fields enclosed by brackets "[]" replaced with your own identifying
+ * information: Portions Copyright [yyyy] [name of copyright owner]
+ *
+ * CDDL HEADER END
+ */
+
+/*
+ * Copyright (c) 2007, 2023, Oracle and/or its affiliates. All rights reserved.
+ * Portions Copyright (c) 2017, 2020, Chris Fraire <cfraire@me.com>.
+ */
+package org.opengrok.suggest.util;
+
+import org.jetbrains.annotations.VisibleForTesting;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Copy of {@code org.opengrok.indexer.util.Progress}.
+ * <p>
+ * Progress reporting via logging. The idea is that for anything that has a set of items
+ * to go through, it will ping an instance of this class for each item completed.
+ * This class will then log based on the number of pings. The bigger the progress,
+ * the higher log level ({@link Level} value) will be used. The default base level is {@code Level.INFO}.
+ * Regardless of the base level, maximum 4 log levels will be used.
+ * </p>
+ */
+public class Progress implements AutoCloseable {
+    private final Logger logger;
+    private final Long totalCount;
+    private final String suffix;
+
+    private final AtomicLong currentCount = new AtomicLong();
+    private final Map<Level, Integer> levelCountMap = new TreeMap<>(Comparator.comparingInt(Level::intValue).reversed());
+    private Thread loggerThread = null;
+    private volatile boolean run;
+
+    private final Level baseLogLevel;
+
+    private final Object sync = new Object();
+
+    /**
+     * @param logger logger instance
+     * @param suffix string suffix to identify the operation
+     * @param totalCount total count
+     * @param logLevel base log level
+     * @param isPrintProgress whether to print the progress
+     */
+    public Progress(Logger logger, String suffix, long totalCount, Level logLevel, boolean isPrintProgress) {
+        this.logger = logger;
+        this.suffix = suffix;
+        this.baseLogLevel = logLevel;
+
+        if (totalCount < 0) {
+            this.totalCount = null;
+        } else {
+            this.totalCount = totalCount;
+        }
+
+        // Note: Level.CONFIG is missing as it does not make too much sense for progress reporting semantically.
+        final List<Level> standardLevels = Arrays.asList(Level.OFF, Level.SEVERE, Level.WARNING, Level.INFO,
+                Level.FINE, Level.FINER, Level.FINEST, Level.ALL);
+        int i = standardLevels.indexOf(baseLogLevel);
+        for (int num : new int[]{100, 50, 10, 1}) {
+            if (i >= standardLevels.size()) {
+                break;
+            }
+
+            Level level = standardLevels.get(i);
+            if (level == null) {
+                break;
+            }
+            levelCountMap.put(level, num);
+            if (num == 1) {
+                break;
+            }
+            i++;
+        }
+
+        // Assuming the printProgress configuration setting cannot be changed on the fly.
+        if (!baseLogLevel.equals(Level.OFF) && isPrintProgress) {
+            spawnLogThread();
+        }
+    }
+
+    private void spawnLogThread() {
+        // spawn a logger thread.
+        run = true;
+        loggerThread = new Thread(this::logLoop,
+                "progress-thread-" + suffix.replaceAll(" ", "_"));
+        loggerThread.start();
+    }
+
+    /**
+     * Increment counter. The actual logging will be done eventually.
+     */
+    public void increment() {
+        this.currentCount.incrementAndGet();
+
+        if (loggerThread != null) {
+            // nag the thread.
+            synchronized (sync) {
+                sync.notifyAll();
+            }
+        }
+    }
+
+    private void logLoop() {
+        long cachedCount = 0;
+        Map<Level, Long> lastLoggedChunk = new HashMap<>();
+
+        while (true) {
+            long currentCount = this.currentCount.get();
+            Level currentLevel = Level.FINEST;
+
+            // Do not log if there was no progress.
+            if (cachedCount < currentCount) {
+                currentLevel = getLevel(lastLoggedChunk, currentCount, currentLevel);
+                logIt(lastLoggedChunk, currentCount, currentLevel);
+            }
+
+            if (!run) {
+                return;
+            }
+
+            cachedCount = currentCount;
+
+            // wait for event
+            try {
+                synchronized (sync) {
+                    if (!run) {
+                        // Loop once more to do the final logging.
+                        continue;
+                    }
+                    sync.wait();
+                }
+            } catch (InterruptedException e) {
+                logger.log(Level.WARNING, "logger thread interrupted");
+            }
+        }
+    }
+
+    @VisibleForTesting
+    Level getLevel(Map<Level, Long> lastLoggedChunk, long currentCount, Level currentLevel) {
+        // The intention is to log the initial and final count at the base log level.
+        if (currentCount <= 1 || (totalCount != null && currentCount == totalCount)) {
+            currentLevel = baseLogLevel;
+        } else {
+            // Set the log level based on the "buckets".
+            for (Level level : levelCountMap.keySet()) {
+                if (lastLoggedChunk.getOrDefault(level, -1L) <
+                        currentCount / levelCountMap.get(level)) {
+                    currentLevel = level;
+                    break;
+                }
+            }
+        }
+        return currentLevel;
+    }
+
+    private void logIt(Map<Level, Long> lastLoggedChunk, long currentCount, Level currentLevel) {
+        if (logger.isLoggable(currentLevel)) {
+            lastLoggedChunk.put(currentLevel, currentCount / levelCountMap.get(currentLevel));
+            StringBuilder stringBuilder = new StringBuilder();
+            stringBuilder.append("Progress: ");
+            stringBuilder.append(currentCount);
+            stringBuilder.append(" ");
+            if (totalCount != null) {
+                stringBuilder.append("(");
+                stringBuilder.append(String.format("%.2f", currentCount * 100.0f / totalCount));
+                stringBuilder.append("%) ");
+            }
+            stringBuilder.append(suffix);
+            logger.log(currentLevel, stringBuilder.toString());
+        }
+    }
+
+    @Override
+    public void close() {
+        if (loggerThread == null) {
+            return;
+        }
+
+        try {
+            run = false;
+            synchronized (sync) {
+                sync.notifyAll();
+            }
+            loggerThread.join();
+        } catch (InterruptedException e) {
+            logger.log(Level.WARNING, "logger thread interrupted");
+        }
+    }
+}
diff --git a/suggester/src/test/java/org/opengrok/suggest/SuggesterTest.java b/suggester/src/test/java/org/opengrok/suggest/SuggesterTest.java
