Merge branch 'stable-3.3'
* stable-3.3:
Update global-refdb to 3.3.2.1 and consume java8 flavor
Add batchThreadPoolSize to high-availability.config
Use dedicated Executor Service for batch indexing
Use separate endpoint for interactive indexing
Use separate FlusherRunner to fix NullPointerExceptions
Reuse repository object when calculating target and meta SHA
Check meta ref sha1 during the index propagation
Upgrade bazlets to latest stable-3.0 to build with 3.0.15 API
Use Gerrit 3.1.12 in Dockerfile
Upgrade bazlets to latest stable-2.16 to build with 2.16.26 API
Explain how to set index.numStripedLocks
Document last index update timestamp storage
Use global-refdb version 3.3.1
Upgrade bazlets to latest stable-3.2 to build with 3.2.6 API
Upgrade bazlets to latest stable-3.1 to build with 3.1.11 API
Use Gerrit 3.1.11 in Dockerfile
HttpSession: Align project name encoding with gerrit core
Change-Id: Id93016c2f2b13c33f0068aedd97784b9ebe509d0
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index cb3f043..9af89f0 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -15,6 +15,6 @@
maven_jar(
name = "global-refdb",
- artifact = "com.gerritforge:global-refdb:3.3.0-rc1",
- sha1 = "1b005b31c27a30ff10de97f903fa2834051bcadf",
+ artifact = "com.gerritforge:global-refdb:3.3.2.1:jdk8",
+ sha1 = "7a293d577665dfc6f4d36371af21c4a3f7177b23",
)
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java
index 7ff8f2b..eb9722d 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/Configuration.java
@@ -51,13 +51,15 @@
public static final int DEFAULT_NUM_STRIPED_LOCKS = 10;
public static final int DEFAULT_TIMEOUT_MS = 5000;
- public static final String PLUGIN_CONFIG_FILE = "high-availability.config";
+ public static final String PLUGIN_NAME = "high-availability";
+ public static final String PLUGIN_CONFIG_FILE = PLUGIN_NAME + ".config";
// common parameter to peerInfo section
static final String PEER_INFO_SECTION = "peerInfo";
// common parameters to cache and index sections
static final String THREAD_POOL_SIZE_KEY = "threadPoolSize";
+ static final String BATCH_THREAD_POOL_SIZE_KEY = "batchThreadPoolSize";
static final int DEFAULT_INDEX_MAX_TRIES = 2;
static final int DEFAULT_INDEX_RETRY_INTERVAL = 30000;
static final int DEFAULT_THREAD_POOL_SIZE = 4;
@@ -109,7 +111,7 @@
index = new Index(cfg);
websession = new Websession(cfg);
healthCheck = new HealthCheck(cfg);
- sharedRefDb = new SharedRefDbConfiguration(cfg);
+ sharedRefDb = new SharedRefDbConfiguration(cfg, PLUGIN_NAME);
}
private static FileBasedConfig getConfigFile(SitePaths sitePaths, String configFileName) {
@@ -480,6 +482,7 @@
static final boolean DEFAULT_SYNCHRONIZE_FORCED = true;
private final int threadPoolSize;
+ private final int batchThreadPoolSize;
private final int retryInterval;
private final int maxTries;
private final int numStripedLocks;
@@ -488,6 +491,7 @@
private Index(Config cfg) {
super(cfg, INDEX_SECTION);
threadPoolSize = getInt(cfg, INDEX_SECTION, THREAD_POOL_SIZE_KEY, DEFAULT_THREAD_POOL_SIZE);
+ batchThreadPoolSize = getInt(cfg, INDEX_SECTION, BATCH_THREAD_POOL_SIZE_KEY, threadPoolSize);
numStripedLocks = getInt(cfg, INDEX_SECTION, NUM_STRIPED_LOCKS, DEFAULT_NUM_STRIPED_LOCKS);
retryInterval = getInt(cfg, INDEX_SECTION, RETRY_INTERVAL_KEY, DEFAULT_INDEX_RETRY_INTERVAL);
maxTries = getInt(cfg, INDEX_SECTION, MAX_TRIES_KEY, DEFAULT_INDEX_MAX_TRIES);
@@ -499,6 +503,10 @@
return threadPoolSize;
}
+ public int batchThreadPoolSize() {
+ return batchThreadPoolSize;
+ }
+
public int numStripedLocks() {
return numStripedLocks;
}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/autoreindex/IndexTs.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/autoreindex/IndexTs.java
index ff42a4e..62d9c10 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/autoreindex/IndexTs.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/autoreindex/IndexTs.java
@@ -48,7 +48,10 @@
private final Path dataDir;
private final ScheduledExecutorService exec;
- private final FlusherRunner flusher;
+ private final FlusherRunner changeFlusher;
+ private final FlusherRunner accountFlusher;
+ private final FlusherRunner groupFlusher;
+ private final FlusherRunner projectFlusher;
private final ChangeFinder changeFinder;
private final CurrentRequestContext currCtx;
@@ -58,16 +61,11 @@
private volatile LocalDateTime projectTs;
class FlusherRunner implements Runnable {
+ private final AbstractIndexRestApiServlet.IndexName index;
@Override
public void run() {
- store(AbstractIndexRestApiServlet.IndexName.CHANGE, changeTs);
- store(AbstractIndexRestApiServlet.IndexName.ACCOUNT, accountTs);
- store(AbstractIndexRestApiServlet.IndexName.GROUP, groupTs);
- store(AbstractIndexRestApiServlet.IndexName.PROJECT, projectTs);
- }
-
- private void store(AbstractIndexRestApiServlet.IndexName index, LocalDateTime latestTs) {
+ LocalDateTime latestTs = getIndexTimeStamp();
Optional<LocalDateTime> currTs = getUpdateTs(index);
if (!currTs.isPresent() || latestTs.isAfter(currTs.get())) {
Path indexTsFile = dataDir.resolve(index.name().toLowerCase());
@@ -78,6 +76,25 @@
}
}
}
+
+ FlusherRunner(AbstractIndexRestApiServlet.IndexName index) {
+ this.index = index;
+ }
+
+ private LocalDateTime getIndexTimeStamp() {
+ switch (index) {
+ case CHANGE:
+ return changeTs;
+ case GROUP:
+ return groupTs;
+ case ACCOUNT:
+ return accountTs;
+ case PROJECT:
+ return projectTs;
+ default:
+ throw new IllegalArgumentException("Unsupported index " + index);
+ }
+ }
}
@Inject
@@ -88,7 +105,10 @@
CurrentRequestContext currCtx) {
this.dataDir = dataDir;
this.exec = queue.getDefaultQueue();
- this.flusher = new FlusherRunner();
+ this.changeFlusher = new FlusherRunner(AbstractIndexRestApiServlet.IndexName.CHANGE);
+ this.accountFlusher = new FlusherRunner(AbstractIndexRestApiServlet.IndexName.ACCOUNT);
+ this.groupFlusher = new FlusherRunner(AbstractIndexRestApiServlet.IndexName.GROUP);
+ this.projectFlusher = new FlusherRunner(AbstractIndexRestApiServlet.IndexName.PROJECT);
this.changeFinder = changeFinder;
this.currCtx = currCtx;
}
@@ -147,19 +167,22 @@
switch (index) {
case CHANGE:
changeTs = dateTime;
+ exec.execute(changeFlusher);
break;
case ACCOUNT:
accountTs = dateTime;
+ exec.execute(accountFlusher);
break;
case GROUP:
groupTs = dateTime;
+ exec.execute(groupFlusher);
break;
case PROJECT:
projectTs = dateTime;
+ exec.execute(projectFlusher);
break;
default:
throw new IllegalArgumentException("Unsupported index " + index);
}
- exec.execute(flusher);
}
}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java
new file mode 100644
index 0000000..dee8876
--- /dev/null
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/ForwardedIndexBatchChangeHandler.java
@@ -0,0 +1,38 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.ericsson.gerrit.plugins.highavailability.forwarder;
+
+import com.ericsson.gerrit.plugins.highavailability.Configuration;
+import com.ericsson.gerrit.plugins.highavailability.index.ChangeCheckerImpl.Factory;
+import com.ericsson.gerrit.plugins.highavailability.index.ForwardedBatchIndexExecutor;
+import com.google.gerrit.server.index.change.ChangeIndexer;
+import com.google.gerrit.server.util.OneOffRequestContext;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import java.util.concurrent.ScheduledExecutorService;
+
+@Singleton
+public class ForwardedIndexBatchChangeHandler extends ForwardedIndexChangeHandler {
+
+ @Inject
+ ForwardedIndexBatchChangeHandler(
+ ChangeIndexer indexer,
+ Configuration config,
+ @ForwardedBatchIndexExecutor ScheduledExecutorService indexExecutor,
+ OneOffRequestContext oneOffCtx,
+ Factory changeCheckerFactory) {
+ super(indexer, config, indexExecutor, oneOffCtx, changeCheckerFactory);
+ }
+}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java
index bb47f11..b73b676 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/Forwarder.java
@@ -42,6 +42,18 @@
CompletableFuture<Boolean> indexChange(String projectName, int changeId, IndexEvent indexEvent);
/**
+ * Forward a change indexing event to the other master using batch index endpoint.
+ *
+ * @param projectName the project of the change to index.
+ * @param changeId the change to index.
+ * @param indexEvent the details of the index event.
+ * @return {@link CompletableFuture} of true if successful, otherwise {@link CompletableFuture} of
+ * false.
+ */
+ CompletableFuture<Boolean> batchIndexChange(
+ String projectName, int changeId, IndexEvent indexEvent);
+
+ /**
* Forward a delete change from index event to the other master.
*
* @param changeId the change to remove from the index.
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/IndexEvent.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/IndexEvent.java
index 037c1c6..71cf044 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/IndexEvent.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/IndexEvent.java
@@ -21,10 +21,14 @@
public class IndexEvent {
public long eventCreatedOn = System.currentTimeMillis() / 1000;
public String targetSha;
+ public String metaSha;
@Override
public String toString() {
- return "IndexEvent@" + format(eventCreatedOn) + ((targetSha != null) ? "/" + targetSha : "");
+ return "IndexEvent@"
+ + format(eventCreatedOn)
+ + ((targetSha != null) ? "/target:" + targetSha : "")
+ + ((metaSha != null) ? "/meta:" + metaSha : "");
}
public static String format(long eventTs) {
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/IndexBatchChangeRestApiServlet.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/IndexBatchChangeRestApiServlet.java
new file mode 100644
index 0000000..bebfae9
--- /dev/null
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/IndexBatchChangeRestApiServlet.java
@@ -0,0 +1,37 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.ericsson.gerrit.plugins.highavailability.forwarder.rest;
+
+import com.ericsson.gerrit.plugins.highavailability.forwarder.ForwardedIndexBatchChangeHandler;
+import com.google.gerrit.extensions.restapi.Url;
+import com.google.gerrit.server.events.EventGson;
+import com.google.gson.Gson;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+@Singleton
+class IndexBatchChangeRestApiServlet extends AbstractIndexRestApiServlet<String> {
+ private static final long serialVersionUID = -1L;
+
+ @Inject
+ IndexBatchChangeRestApiServlet(ForwardedIndexBatchChangeHandler handler, @EventGson Gson gson) {
+ super(handler, IndexName.CHANGE, true, gson);
+ }
+
+ @Override
+ String parse(String id) {
+ return Url.decode(id);
+ }
+}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java
index d404ab2..ea4ad10 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarder.java
@@ -83,6 +83,16 @@
event);
}
+ public CompletableFuture<Boolean> batchIndexChange(
+ String projectName, int changeId, IndexEvent event) {
+ return execute(
+ RequestMethod.POST,
+ "index change",
+ "index/change/batch",
+ buildIndexEndpoint(projectName, changeId),
+ event);
+ }
+
@Override
public CompletableFuture<Boolean> deleteChangeFromIndex(final int changeId, IndexEvent event) {
return execute(
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderServletModule.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderServletModule.java
index 589bbef..4d3de37 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderServletModule.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderServletModule.java
@@ -20,6 +20,7 @@
@Override
protected void configureServlets() {
serveRegex("/index/account/\\d+$").with(IndexAccountRestApiServlet.class);
+ serveRegex("/index/change/batch/.*$").with(IndexBatchChangeRestApiServlet.class);
serveRegex("/index/change/.*$").with(IndexChangeRestApiServlet.class);
serveRegex("/index/group/\\w+$").with(IndexGroupRestApiServlet.class);
serveRegex("/index/project/.*$").with(IndexProjectRestApiServlet.class);
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutor.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutor.java
new file mode 100644
index 0000000..06da9f0
--- /dev/null
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutor.java
@@ -0,0 +1,24 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.ericsson.gerrit.plugins.highavailability.index;
+
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import com.google.inject.BindingAnnotation;
+import java.lang.annotation.Retention;
+
+@Retention(RUNTIME)
+@BindingAnnotation
+@interface BatchIndexExecutor {}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutorProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutorProvider.java
new file mode 100644
index 0000000..cfbd4fb
--- /dev/null
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/BatchIndexExecutorProvider.java
@@ -0,0 +1,30 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.ericsson.gerrit.plugins.highavailability.index;
+
+import com.ericsson.gerrit.plugins.highavailability.Configuration;
+import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+@Singleton
+class BatchIndexExecutorProvider extends ExecutorProvider {
+
+ @Inject
+ BatchIndexExecutorProvider(WorkQueue workQueue, Configuration config) {
+ super(workQueue, config.index().batchThreadPoolSize(), "Forward-BatchIndex-Event");
+ }
+}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ChangeCheckerImpl.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ChangeCheckerImpl.java
index 6587acf..1a2ebf7 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ChangeCheckerImpl.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ChangeCheckerImpl.java
@@ -18,6 +18,7 @@
import com.google.common.flogger.FluentLogger;
import com.google.gerrit.entities.Change;
import com.google.gerrit.entities.HumanComment;
+import com.google.gerrit.entities.RefNames;
import com.google.gerrit.server.CommentsUtil;
import com.google.gerrit.server.change.ChangeFinder;
import com.google.gerrit.server.git.GitRepositoryManager;
@@ -62,15 +63,25 @@
}
@Override
- public Optional<IndexEvent> newIndexEvent() {
- return getComputedChangeTs()
- .map(
- ts -> {
- IndexEvent event = new IndexEvent();
- event.eventCreatedOn = ts;
- event.targetSha = getBranchTargetSha();
- return event;
- });
+ public Optional<IndexEvent> newIndexEvent() throws IOException {
+ Optional<Long> changeTs = getComputedChangeTs();
+ if (!changeTs.isPresent()) {
+ return Optional.empty();
+ }
+
+ long ts = changeTs.get();
+
+ IndexEvent event = new IndexEvent();
+ event.eventCreatedOn = ts;
+ try (Repository repo = gitRepoMgr.openRepository(changeNotes.get().getProjectName())) {
+ event.targetSha = getBranchTargetSha();
+ event.metaSha = getMetaSha(repo);
+ return Optional.of(event);
+ } catch (IOException e) {
+ log.atSevere().withCause(e).log(
+ "Unable to create index event for project %s", changeNotes.get().getProjectName());
+ throw e;
+ }
}
@Override
@@ -82,25 +93,31 @@
}
@Override
- public boolean isChangeUpToDate(Optional<IndexEvent> indexEvent) {
+ public boolean isChangeUpToDate(Optional<IndexEvent> indexEventOption) throws IOException {
getComputedChangeTs();
- log.atFine().log("Checking change %s against index event %s", this, indexEvent);
+ log.atFine().log("Checking change %s against index event %s", this, indexEventOption);
if (!computedChangeTs.isPresent()) {
log.atWarning().log("Unable to compute last updated ts for change %s", changeId);
return false;
}
+ try {
+ if (indexEventOption.isPresent()) {
+ try (Repository repo = gitRepoMgr.openRepository(changeNotes.get().getProjectName())) {
+ IndexEvent indexEvent = indexEventOption.get();
+ return (computedChangeTs.get() > indexEvent.eventCreatedOn)
+ || (computedChangeTs.get() == indexEvent.eventCreatedOn)
+ && (Objects.isNull(indexEvent.targetSha)
+ || Objects.equals(getBranchTargetSha(), indexEvent.targetSha))
+ && (Objects.isNull(indexEvent.metaSha)
+ || Objects.equals(getMetaSha(repo), indexEvent.metaSha));
+ }
+ }
+ return true;
- if (indexEvent.isPresent() && indexEvent.get().targetSha == null) {
- return indexEvent.map(e -> (computedChangeTs.get() >= e.eventCreatedOn)).orElse(true);
+ } catch (IOException ex) {
+ log.atWarning().log("Unable to read meta sha for change %s", changeId);
+ return false;
}
-
- return indexEvent
- .map(
- e ->
- (computedChangeTs.get() > e.eventCreatedOn)
- || (computedChangeTs.get() == e.eventCreatedOn)
- && (Objects.equals(getBranchTargetSha(), e.targetSha)))
- .orElse(true);
}
@Override
@@ -113,12 +130,19 @@
@Override
public String toString() {
- return "change-id="
- + changeId
- + "@"
- + getComputedChangeTs().map(IndexEvent::format)
- + "/"
- + getBranchTargetSha();
+ try (Repository repo = gitRepoMgr.openRepository(changeNotes.get().getProjectName())) {
+ return "change-id="
+ + changeId
+ + "@"
+ + getComputedChangeTs().map(IndexEvent::format)
+ + "/target:"
+ + getBranchTargetSha()
+ + "/meta:"
+ + getMetaSha(repo);
+ } catch (IOException e) {
+ log.atSevere().withCause(e).log("Unable to render change %s", changeId);
+ return "change-id=" + changeId;
+ }
}
private String getBranchTargetSha() {
@@ -141,6 +165,16 @@
return getChangeNotes().map(this::getTsFromChangeAndDraftComments);
}
+ private String getMetaSha(Repository repo) throws IOException {
+ String refName = RefNames.changeMetaRef(changeNotes.get().getChange().getId());
+ Ref ref = repo.exactRef(refName);
+ if (ref == null) {
+ throw new IOException(
+ String.format("Unable to find meta ref %s for change %s", refName, changeId));
+ }
+ return ref.getTarget().getObjectId().getName();
+ }
+
private long getTsFromChangeAndDraftComments(ChangeNotes notes) {
Change change = notes.getChange();
Timestamp changeTs = change.getLastUpdatedOn();
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutor.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutor.java
new file mode 100644
index 0000000..b5b6aab
--- /dev/null
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutor.java
@@ -0,0 +1,24 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.ericsson.gerrit.plugins.highavailability.index;
+
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import com.google.inject.BindingAnnotation;
+import java.lang.annotation.Retention;
+
+@Retention(RUNTIME)
+@BindingAnnotation
+public @interface ForwardedBatchIndexExecutor {}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java
new file mode 100644
index 0000000..f5bc85e
--- /dev/null
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/ForwardedBatchIndexExecutorProvider.java
@@ -0,0 +1,30 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.ericsson.gerrit.plugins.highavailability.index;
+
+import com.ericsson.gerrit.plugins.highavailability.Configuration;
+import com.ericsson.gerrit.plugins.highavailability.ExecutorProvider;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+@Singleton
+class ForwardedBatchIndexExecutorProvider extends ExecutorProvider {
+
+ @Inject
+ ForwardedBatchIndexExecutorProvider(WorkQueue workQueue, Configuration config) {
+ super(workQueue, config.index().batchThreadPoolSize(), "Forwarded-BatchIndex-Event");
+ }
+}
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java
index 368581c..73e842f 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandler.java
@@ -40,6 +40,7 @@
ProjectIndexedListener {
private static final FluentLogger log = FluentLogger.forEnclosingClass();
private final ScheduledExecutorService executor;
+ private final ScheduledExecutorService batchExecutor;
private final Forwarder forwarder;
private final String pluginName;
private final Set<IndexTask> queuedTasks = Collections.newSetFromMap(new ConcurrentHashMap<>());
@@ -53,6 +54,7 @@
@Inject
IndexEventHandler(
@IndexExecutor ScheduledExecutorService executor,
+ @BatchIndexExecutor ScheduledExecutorService batchExecutor,
@PluginName String pluginName,
Forwarder forwarder,
ChangeCheckerImpl.Factory changeChecker,
@@ -61,6 +63,7 @@
IndexEventLocks locks) {
this.forwarder = forwarder;
this.executor = executor;
+ this.batchExecutor = batchExecutor;
this.pluginName = pluginName;
this.changeChecker = changeChecker;
this.currCtx = currCtx;
@@ -95,11 +98,21 @@
changeChecker
.create(changeId)
.newIndexEvent()
- .map(event -> new IndexChangeTask(projectName, id, event))
+ .map(
+ event -> {
+ if (Thread.currentThread().getName().contains("Batch")) {
+ return new BatchIndexChangeTask(projectName, id, event);
+ }
+ return new IndexChangeTask(projectName, id, event);
+ })
.ifPresent(
task -> {
if (queuedTasks.add(task)) {
- executor.execute(task);
+ if (task instanceof BatchIndexChangeTask) {
+ batchExecutor.execute(task);
+ } else {
+ executor.execute(task);
+ }
}
});
} catch (Exception e) {
@@ -215,6 +228,46 @@
}
}
+ class BatchIndexChangeTask extends IndexTask {
+ private final int changeId;
+ private final String projectName;
+
+ BatchIndexChangeTask(String projectName, int changeId, IndexEvent indexEvent) {
+ super(indexEvent);
+ this.projectName = projectName;
+ this.changeId = changeId;
+ }
+
+ @Override
+ public CompletableFuture<Boolean> execute() {
+ return forwarder.batchIndexChange(projectName, changeId, indexEvent);
+ }
+
+ @Override
+ String indexId() {
+ return "change/" + changeId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(IndexEventHandler.BatchIndexChangeTask.class, changeId);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof IndexEventHandler.BatchIndexChangeTask)) {
+ return false;
+ }
+ IndexEventHandler.BatchIndexChangeTask other = (IndexEventHandler.BatchIndexChangeTask) obj;
+ return changeId == other.changeId;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("[%s] Index change %s in target instance", pluginName, changeId);
+ }
+ }
+
class DeleteChangeTask extends IndexTask {
private final int changeId;
diff --git a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java
index 4a25fc8..3bcc34f 100644
--- a/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java
+++ b/src/main/java/com/ericsson/gerrit/plugins/highavailability/index/IndexModule.java
@@ -35,6 +35,12 @@
.annotatedWith(ForwardedIndexExecutor.class)
.toProvider(ForwardedIndexExecutorProvider.class);
bind(IndexEventLocks.class).in(Scopes.SINGLETON);
+ bind(ScheduledExecutorService.class)
+ .annotatedWith(BatchIndexExecutor.class)
+ .toProvider(BatchIndexExecutorProvider.class);
+ bind(ScheduledExecutorService.class)
+ .annotatedWith(ForwardedBatchIndexExecutor.class)
+ .toProvider(ForwardedBatchIndexExecutorProvider.class);
listener().to(IndexExecutorProvider.class);
DynamicSet.bind(binder(), ChangeIndexedListener.class)
.to(IndexEventHandler.class)
diff --git a/src/main/resources/Documentation/about.md b/src/main/resources/Documentation/about.md
index 851f584..80117e2 100644
--- a/src/main/resources/Documentation/about.md
+++ b/src/main/resources/Documentation/about.md
@@ -171,4 +171,17 @@
```
[ref-database "zookeeper"]
connectString = zookeeperhost:2181
-```
\ No newline at end of file
+```
+### Last index update timestamp storage
+
+The plugin keeps track of the timestamp when it lastly updated an index.
+When the [`autoReindex.enabled` option](config.md) is set to `true`, the timestamp
+is used to determine which changes to reindex when a node is temporarily out of
+sync with the primary, for example, after a node being offline for a long time.
+
+The HA plugin keeps the last update timestamp for each index in the following files:
+* `<gerrit_home>/data/high-availability/group`
+* `<gerrit_home>/data/high-availability/account`
+* `<gerrit_home>/data/high-availability/change`
+
+The timestamp is stored in this format `yyyy-mm-ddTHH:MM:SS.ss`, i.e.: `2020-12-18T12:17:53.25`.
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 67dd815..2f7d41a 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -180,7 +180,11 @@
Defaults to true.
```index.numStripedLocks```
-: Number of striped locks to use during reindexing.
+: Number of striped locks to use during reindexing. Should be of the same order
+ of magnitude of the open changes. I.e.: if one has 5000 changes, one might want
+ to have at least 1000 striped locks. The value has to be tuned empirically
+ by checking the number of failures in acquiring the locking. Checking the
+ `consider increasing the number of shards` warnings should help.
Defaults to 10.
```index.synchronize```
@@ -196,6 +200,11 @@
: Maximum number of threads used to send index events to the target instance.
Defaults to 4.
+```index.batchThreadPoolSize```
+: Maximum number of threads used to send batch index events to the target instance
+ and not associated to an interactive action performed by a user.
+ Defaults equal index.threadPoolSize.
+
```index.maxTries```
: Maximum number of times the plugin should attempt to reindex changes.
Setting this value to 0 will disable retries. After this number of failed tries,
diff --git a/src/test/docker/gerrit/Dockerfile b/src/test/docker/gerrit/Dockerfile
index 4a1dfbb..5f455d4 100644
--- a/src/test/docker/gerrit/Dockerfile
+++ b/src/test/docker/gerrit/Dockerfile
@@ -1,4 +1,4 @@
-FROM gerritcodereview/gerrit:3.1.8
+FROM gerritcodereview/gerrit:3.1.12
ENV GERRIT_BRANCH=stable-3.1
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/ConfigurationTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/ConfigurationTest.java
index 4bfac28..19cc48d 100644
--- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/ConfigurationTest.java
+++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/ConfigurationTest.java
@@ -14,6 +14,7 @@
package com.ericsson.gerrit.plugins.highavailability;
+import static com.ericsson.gerrit.plugins.highavailability.Configuration.BATCH_THREAD_POOL_SIZE_KEY;
import static com.ericsson.gerrit.plugins.highavailability.Configuration.Cache.CACHE_SECTION;
import static com.ericsson.gerrit.plugins.highavailability.Configuration.Cache.PATTERN_KEY;
import static com.ericsson.gerrit.plugins.highavailability.Configuration.DEFAULT_NUM_STRIPED_LOCKS;
@@ -264,6 +265,19 @@
}
@Test
+ public void testGetBatchIndexThreadPoolSize() throws Exception {
+ assertThat(getConfiguration().index().batchThreadPoolSize())
+ .isEqualTo(DEFAULT_THREAD_POOL_SIZE);
+
+ globalPluginConfig.setInt(INDEX_SECTION, null, BATCH_THREAD_POOL_SIZE_KEY, THREAD_POOL_SIZE);
+ assertThat(getConfiguration().index().batchThreadPoolSize()).isEqualTo(THREAD_POOL_SIZE);
+
+ globalPluginConfig.setString(INDEX_SECTION, null, BATCH_THREAD_POOL_SIZE_KEY, INVALID_INT);
+ assertThat(getConfiguration().index().batchThreadPoolSize())
+ .isEqualTo(DEFAULT_THREAD_POOL_SIZE);
+ }
+
+ @Test
public void testGetIndexSynchronize() throws Exception {
assertThat(getConfiguration().index().synchronize()).isEqualTo(DEFAULT_SYNCHRONIZE);
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java
index 046d6ad..7908669 100644
--- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java
+++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/forwarder/rest/RestForwarderTest.java
@@ -68,6 +68,14 @@
PLUGIN_NAME,
"index/change",
PROJECT_NAME_URL_END + "~" + CHANGE_NUMBER);
+ private static final String INDEX_BATCH_CHANGE_ENDPOINT =
+ Joiner.on("/")
+ .join(
+ URL,
+ PLUGINS,
+ PLUGIN_NAME,
+ "index/change/batch",
+ PROJECT_NAME_URL_END + "~" + CHANGE_NUMBER);
private static final String DELETE_CHANGE_ENDPOINT =
Joiner.on("/").join(URL, PLUGINS, PLUGIN_NAME, "index/change", "~" + CHANGE_NUMBER);
private static final int ACCOUNT_NUMBER = 2;
@@ -198,6 +206,29 @@
}
@Test
+ public void testIndexBatchChangeOK() throws Exception {
+ when(httpSessionMock.post(eq(INDEX_BATCH_CHANGE_ENDPOINT), any()))
+ .thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG));
+ assertThat(forwarder.batchIndexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent()).get())
+ .isTrue();
+ }
+
+ @Test
+ public void testIndexBatchChangeFailed() throws Exception {
+ when(httpSessionMock.post(eq(INDEX_BATCH_CHANGE_ENDPOINT), any()))
+ .thenReturn(new HttpResult(FAILED, EMPTY_MSG));
+ assertThat(forwarder.batchIndexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent()).get())
+ .isFalse();
+ }
+
+ @Test
+ public void testIndexBatchChangeThrowsException() throws Exception {
+ doThrow(new IOException()).when(httpSessionMock).post(eq(INDEX_BATCH_CHANGE_ENDPOINT), any());
+ assertThat(forwarder.batchIndexChange(PROJECT_NAME, CHANGE_NUMBER, new IndexEvent()).get())
+ .isFalse();
+ }
+
+ @Test
public void testChangeDeletedFromIndexOK() throws Exception {
when(httpSessionMock.delete(eq(DELETE_CHANGE_ENDPOINT)))
.thenReturn(new HttpResult(SUCCESSFUL, EMPTY_MSG));
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/ChangeCheckerIT.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/ChangeCheckerIT.java
new file mode 100644
index 0000000..febf8f0
--- /dev/null
+++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/ChangeCheckerIT.java
@@ -0,0 +1,127 @@
+// Copyright (C) 2021 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.ericsson.gerrit.plugins.highavailability.index;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.ericsson.gerrit.plugins.highavailability.forwarder.IndexEvent;
+import com.google.gerrit.acceptance.LightweightPluginDaemonTest;
+import com.google.gerrit.acceptance.PushOneCommit.Result;
+import com.google.gerrit.acceptance.TestPlugin;
+import com.google.gerrit.entities.RefNames;
+import java.io.IOException;
+import java.util.Optional;
+import org.eclipse.jgit.lib.Ref;
+import org.eclipse.jgit.lib.Repository;
+import org.junit.Test;
+
+@TestPlugin(
+ name = "high-availability",
+ sysModule = "com.ericsson.gerrit.plugins.highavailability.Module",
+ httpModule = "com.ericsson.gerrit.plugins.highavailability.HttpModule")
+public class ChangeCheckerIT extends LightweightPluginDaemonTest {
+
+ ChangeCheckerImpl.Factory changeCheckerFactory;
+
+ @Override
+ public void setUpTestPlugin() throws Exception {
+ super.setUpTestPlugin();
+ changeCheckerFactory = plugin.getSysInjector().getInstance(ChangeCheckerImpl.Factory.class);
+ }
+
+ @Test
+ public void shouldPopulateMetaSha() throws Exception {
+ Result change = createChange();
+ ChangeChecker changeChecker = changeCheckerFactory.create(change.getChangeId());
+ Optional<IndexEvent> eventOption = changeChecker.newIndexEvent();
+
+ assertThat(eventOption.isPresent()).isTrue();
+ IndexEvent event = eventOption.get();
+ assertThat(event.metaSha).isNotNull();
+ assertThat(event.metaSha).isEqualTo(readMetaSha(change));
+ }
+
+ @Test
+ public void shouldReturnIsUpToDateTrueWhenEventContainsCorrectMetaAndTargetSha()
+ throws Exception {
+ Result change = createChange();
+ ChangeChecker changeChecker = changeCheckerFactory.create(change.getChangeId());
+ Optional<IndexEvent> event = changeChecker.newIndexEvent();
+
+ assertThat(changeChecker.isChangeUpToDate(event)).isTrue();
+ }
+
+ @Test
+ public void shouldReturnIsUpToDateTrueWhenTargetShaIsNull() throws Exception {
+ Result change = createChange();
+ ChangeChecker changeChecker = changeCheckerFactory.create(change.getChangeId());
+ Optional<IndexEvent> event =
+ changeChecker
+ .newIndexEvent()
+ .map(
+ e -> {
+ e.targetSha = null;
+ return e;
+ });
+
+ assertThat(changeChecker.isChangeUpToDate(event)).isTrue();
+ }
+
+ @Test
+ public void shouldReturnFalseWhenMetaShaIsNotUpToDate() throws Exception {
+ String testMetaRefSha = "6212efebe6e8b9f439a8ad013243e602afab7441";
+ Result change = createChange();
+ ChangeChecker changeChecker = changeCheckerFactory.create(change.getChangeId());
+ Optional<IndexEvent> event =
+ changeChecker
+ .newIndexEvent()
+ .map(
+ e -> {
+ e.metaSha = testMetaRefSha;
+ return e;
+ });
+
+ assertThat(changeChecker.isChangeUpToDate(event)).isFalse();
+ }
+
+ @Test
+ public void shouldReturnFalseWhenTargetShaIsNotUpToDate() throws Exception {
+ String testTargetRefSha = "abed47baf2818a86b68cf712073a748a6b5b293e";
+ Result change = createChange();
+ ChangeChecker changeChecker = changeCheckerFactory.create(change.getChangeId());
+ Optional<IndexEvent> event =
+ changeChecker
+ .newIndexEvent()
+ .map(
+ e -> {
+ e.targetSha = testTargetRefSha;
+ return e;
+ });
+
+ assertThat(changeChecker.isChangeUpToDate(event)).isFalse();
+ }
+
+ private String readMetaSha(Result change) throws IOException {
+ try (Repository repo = repoManager.openRepository(change.getChange().change().getProject())) {
+ String refName = RefNames.changeMetaRef(change.getChange().getId());
+ Ref ref = repo.exactRef(refName);
+ if (ref == null) {
+ return null;
+ }
+
+ return ref.getTarget().getObjectId().getName();
+ }
+ }
+}
diff --git a/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java
index 875f203..578dc17 100644
--- a/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java
+++ b/src/test/java/com/ericsson/gerrit/plugins/highavailability/index/IndexEventHandlerTest.java
@@ -85,6 +85,7 @@
private Account.Id accountId;
private AccountGroup.UUID accountGroupUUID;
private ScheduledExecutorService executor = new CurrentThreadScheduledExecutorService();
+ private ScheduledExecutorService batchExecutor = new CurrentThreadScheduledExecutorService();
private ScheduledExecutorService testExecutor =
Executors.newScheduledThreadPool(MAX_TEST_PARALLELISM);
@Mock private RequestContext mockCtx;
@@ -142,6 +143,7 @@
indexEventHandler =
new IndexEventHandler(
executor,
+ batchExecutor,
PLUGIN_NAME,
forwarder,
changeCheckerFactoryMock,
@@ -387,9 +389,11 @@
@Test
public void duplicateChangeEventOfAQueuedEventShouldGetDiscarded() {
ScheduledThreadPoolExecutor poolMock = mock(ScheduledThreadPoolExecutor.class);
+ ScheduledThreadPoolExecutor poolBatchMock = mock(ScheduledThreadPoolExecutor.class);
indexEventHandler =
new IndexEventHandler(
poolMock,
+ poolBatchMock,
PLUGIN_NAME,
forwarder,
changeCheckerFactoryMock,
@@ -405,9 +409,11 @@
@Test
public void duplicateAccountEventOfAQueuedEventShouldGetDiscarded() {
ScheduledThreadPoolExecutor poolMock = mock(ScheduledThreadPoolExecutor.class);
+ ScheduledThreadPoolExecutor poolBatchMock = mock(ScheduledThreadPoolExecutor.class);
indexEventHandler =
new IndexEventHandler(
poolMock,
+ poolBatchMock,
PLUGIN_NAME,
forwarder,
changeCheckerFactoryMock,
@@ -422,9 +428,11 @@
@Test
public void duplicateGroupEventOfAQueuedEventShouldGetDiscarded() {
ScheduledThreadPoolExecutor poolMock = mock(ScheduledThreadPoolExecutor.class);
+ ScheduledThreadPoolExecutor poolBatchMock = mock(ScheduledThreadPoolExecutor.class);
indexEventHandler =
new IndexEventHandler(
poolMock,
+ poolBatchMock,
PLUGIN_NAME,
forwarder,
changeCheckerFactoryMock,
