Merge "ReplicationState: Streamline getRefStatus()"
diff --git a/BUILD b/BUILD
index 4c74bf2..ee97660 100644
--- a/BUILD
+++ b/BUILD
@@ -14,9 +14,6 @@
"Gerrit-SshModule: com.googlesource.gerrit.plugins.replication.SshModule",
],
resources = glob(["src/main/resources/**/*"]),
- deps = [
- "//lib/auto:auto-value-gson",
- ],
)
junit_tests(
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
index e0a9354..8ef21d0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
@@ -63,6 +63,7 @@
import com.google.inject.assistedinject.FactoryModuleBuilder;
import com.google.inject.servlet.RequestScoped;
import com.googlesource.gerrit.plugins.replication.ReplicationState.RefPushResult;
+import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionState;
import com.googlesource.gerrit.plugins.replication.events.RefReplicatedEvent;
import com.googlesource.gerrit.plugins.replication.events.ReplicationScheduledEvent;
@@ -70,7 +71,6 @@
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -463,7 +463,7 @@
}
void scheduleDeleteProject(URIish uri, Project.NameKey project, ProjectDeletionState state) {
- repLog.atFine().log("scheduling deletion of project {} at {}", project, uri);
+ repLog.atFine().log("scheduling deletion of project %s at %s", project, uri);
@SuppressWarnings("unused")
ScheduledFuture<?> ignored =
pool.schedule(deleteProjectFactory.create(uri, project, state), 0, TimeUnit.SECONDS);
@@ -608,15 +608,15 @@
}
}
- public Set<String> getPrunableTaskNames() {
- Set<String> names = new HashSet<>();
+ public Map<ReplicateRefUpdate, String> getTaskNamesByReplicateRefUpdate() {
+ Map<ReplicateRefUpdate, String> taskNameByReplicateRefUpdate = new HashMap<>();
for (PushOne push : pending.values()) {
- if (!replicationTasksStorage.get().isWaiting(push)) {
- repLog.atFine().log("No longer isWaiting, can prune %s", push.getURI());
- names.add(push.toString());
+ String taskName = push.toString();
+ for (ReplicateRefUpdate refUpdate : push.getReplicateRefUpdates()) {
+ taskNameByReplicateRefUpdate.put(refUpdate, taskName);
}
}
- return names;
+ return taskNameByReplicateRefUpdate;
}
boolean wouldPush(URIish uri, Project.NameKey project, String ref) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java b/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java
index d195aa3..0fa02ef 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java
@@ -68,7 +68,8 @@
if (exitCode == 1) {
logger.atInfo().log(
"DeleteProject plugin is not installed on %s;"
- + " will not try to forward this operation to that host");
+ + " will not try to forward this operation to that host",
+ uri);
withoutDeleteProjectPlugin.add(uri);
}
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
index 404d4bd..87c35ee 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
@@ -468,6 +468,7 @@
pool.reschedule(this, Destination.RetryReason.TRANSPORT_ERROR);
}
} else {
+ retryDone();
repLog.atSevere().log(
"Giving up after %d '%s' failures during replication to %s",
updateRefRetryCount, e.getMessage(), uri);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
index ed474ae..5310c14 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -41,6 +41,7 @@
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -69,6 +70,11 @@
private final Queue<ReferenceUpdatedEvent> beforeStartupEventsQueue;
private Distributor distributor;
+ protected enum Prune {
+ TRUE,
+ FALSE;
+ }
+
@Inject
ReplicationQueue(
ReplicationConfig rc,
@@ -94,7 +100,7 @@
destinations.get().startup(workQueue);
running = true;
replicationTasksStorage.recoverAll();
- firePendingEvents();
+ synchronizePendingEvents(Prune.FALSE);
fireBeforeStartupEvents();
distributor = new Distributor(workQueue);
}
@@ -193,8 +199,14 @@
}
}
- private void firePendingEvents() {
+ private void synchronizePendingEvents(Prune prune) {
if (replaying.compareAndSet(false, true)) {
+ final Map<ReplicateRefUpdate, String> taskNamesByReplicateRefUpdate = new ConcurrentHashMap<>();
+ if (Prune.TRUE.equals(prune)) {
+ for (Destination destination : destinations.get().getAll(FilterType.ALL)) {
+ taskNamesByReplicateRefUpdate.putAll(destination.getTaskNamesByReplicateRefUpdate());
+ }
+ }
new ChainedScheduler.StreamScheduler<>(
workQueue.getDefaultQueue(),
replicationTasksStorage.streamWaiting(),
@@ -203,6 +215,9 @@
public void run(ReplicationTasksStorage.ReplicateRefUpdate u) {
try {
fire(new URIish(u.uri()), Project.nameKey(u.project()), u.ref());
+ if (Prune.TRUE.equals(prune)) {
+ taskNamesByReplicateRefUpdate.remove(u);
+ }
} catch (URISyntaxException e) {
repLog.atSevere().withCause(e).log(
"Encountered malformed URI for persisted event %s", u);
@@ -213,6 +228,9 @@
@Override
public void onDone() {
+ if (Prune.TRUE.equals(prune)) {
+ pruneNoLongerPending(new HashSet<>(taskNamesByReplicateRefUpdate.values()));
+ }
replaying.set(false);
}
@@ -224,17 +242,12 @@
}
}
- private void pruneCompleted() {
+ private void pruneNoLongerPending(Set<String> prunableTaskNames) {
// Queue tasks have wrappers around them so workQueue.getTasks() does not return the PushOnes.
// We also cannot access them by taskId since PushOnes don't have a taskId, they do have
- // and Id, but it not the id assigned to the task in the queues. The tasks in the queue
- // do use the same name as returned by toString() though, so that be used to correlate
+ // an Id, but it is not the id assigned to the task in the queues. The tasks in the queue
+ // do use the same name as returned by toString() though, so that can be used to correlate
// PushOnes with queue tasks despite their wrappers.
- Set<String> prunableTaskNames = new HashSet<>();
- for (Destination destination : destinations.get().getAll(FilterType.ALL)) {
- prunableTaskNames.addAll(destination.getPrunableTaskNames());
- }
-
for (WorkQueue.Task<?> task : workQueue.getTasks()) {
WorkQueue.Task.State state = task.getState();
if (state == WorkQueue.Task.State.SLEEPING || state == WorkQueue.Task.State.READY) {
@@ -310,8 +323,7 @@
return;
}
try {
- firePendingEvents();
- pruneCompleted();
+ synchronizePendingEvents(Prune.TRUE);
} catch (Exception e) {
repLog.atSevere().withCause(e).log("error distributing tasks");
}
diff --git a/src/main/resources/Documentation/about.md b/src/main/resources/Documentation/about.md
index ded9d84..d216366 100644
--- a/src/main/resources/Documentation/about.md
+++ b/src/main/resources/Documentation/about.md
@@ -16,11 +16,9 @@
local path as replication target. This makes e.g. sense if a network
share is mounted to which the repositories should be replicated.
-In multi-primary scenario, any replication work which is already
-in-flight or completed by the other nodes is not performed to
-avoid extra work. This is because, the storage for replication
-events is shared between multiple primaries.(The storage location
-is specified in the config using: `replication.eventsDirectory`).
+It is possible to
+[configure](config.html#configuring-cluster-replication) the plugin so
+that multiple primaries share the replication work approximately evenly.
Replication of account data (NoteDb)
------------------------------------
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index f4ea9d6..8843671 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -46,6 +46,59 @@
To manually trigger replication at runtime, see
SSH command [start](cmd-start.md).
+<a name="configuring-cluster-replication"></a>
+Configuring Cluster Replication
+-------------------------------
+
+The replication plugin is designed to allow multiple primaries in a
+cluster to efficiently cooperate together. This cooperation is based on
+the replication event persistence subsystem and thus the directory
+pointed to by the replication.eventsDirectory config key must reside on
+a shared filesystem, such as NFS, to enable this cooperation. By
+default simply pointing multiple primaries to the same eventsDirectory
+will enable some cooperation by preventing the same replication push
+from being duplicated by more than one primary.
+
+To further improve cooperation across the cluster, the
+replication.distributionInterval config value can be set. With
+distribution enabled, the replication queues for all the nodes sharing
+the same eventsDirectory will reflect approximately the same outstanding
+replication work (i.e. tasks waiting in the queue). Replication pushes
+which are running will continue to only be visible in the queue of the
+node on which the push is actually happening. This feature not only
+helps administrators get a cluster wide view of outstanding replication
+tasks, it allows replication tasks triggered by one primary to be
+fullfilled by another node which is less busy.
+
+This enhanced replication work distribution allows the amount of
+replication work a cluster can handle to scale more evenly and linearly
+with the amount of primaries in the cluster. Adding more nodes to a
+cluster without distribution enabled will generally not allow the thread
+count per remote to be reduced without impacting service levels to those
+remotes. This is because without distribution, all events triggered by a
+node will only be fullfilled by the node which triggered the event, even
+if all the other nodes in the cluster are idle. This behavior implies
+that each node should be configured in a way that allows it alone to
+provide the level of service which each remote requires. However with
+distribution enabled, it becomes possible to reduce the amount of
+replication threads configured per remote proportionally to the amount
+of nodes in the cluster while maintaining the same approximate service
+level as before adding new nodes.
+
+Thread per remote reduction without service impacts is possible with
+distribution because when configuring a node it can be expected that
+other nodes will pick up some of the work it triggers and it no longer
+needs to be configured as if it were the only node in the cluster. For
+example, if a remote requires 6 threads with one node to achieve
+acceptable service, it should only take 2 threads on 3 equivalently
+powered nodes to provide the same service level with distribution
+enabled. Scaling down of the thread requirements per remote results in a
+reduced memory footprint per remote on each node in the cluster and this
+enables the nodes in the cluster to now scale to handle more remotes
+with the approximate same service level than without distribution. The
+amount of extra supported remotes then also scales approximately
+linearly with the extra nodes in a cluster.
+
File `replication.config`
-------------------------
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
index 202d77e..9606371 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
@@ -26,6 +26,7 @@
import com.google.gerrit.extensions.api.changes.NotifyHandling;
import com.google.gerrit.extensions.events.ProjectDeletedListener;
import com.google.gerrit.server.config.SitePaths;
+import com.google.gerrit.server.git.LocalDiskRepositoryManager;
import com.google.inject.Inject;
import java.io.IOException;
import java.nio.file.Path;
@@ -71,6 +72,18 @@
protected Path gitPath;
protected FileBasedConfig config;
+ protected void setDistributionInterval(int interval) throws IOException {
+ config.setInt("replication", null, "distributionInterval", interval);
+ config.save();
+ }
+
+ protected String getProjectUri(Project.NameKey project) throws Exception {
+ return ((LocalDiskRepositoryManager) repoManager)
+ .getBasePath(project)
+ .resolve(project.get() + ".git")
+ .toString();
+ }
+
protected void setReplicationDestination(
String remoteName, String replicaSuffix, Optional<String> project) throws IOException {
setReplicationDestination(
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java
new file mode 100644
index 0000000..3d0dfc1
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java
@@ -0,0 +1,126 @@
+// 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.googlesource.gerrit.plugins.replication;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.gerrit.acceptance.TestPlugin;
+import com.google.gerrit.acceptance.UseLocalDisk;
+import com.google.gerrit.acceptance.WaitUtil;
+import com.google.gerrit.entities.BranchNameKey;
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.server.git.WorkQueue;
+import java.time.Duration;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.eclipse.jgit.lib.Ref;
+import org.eclipse.jgit.lib.RefUpdate;
+import org.eclipse.jgit.lib.Repository;
+import org.eclipse.jgit.transport.URIish;
+import org.junit.Test;
+
+/**
+ * The tests in this class ensure the correctness of {@link
+ * com.googlesource.gerrit.plugins.replication.ReplicationQueue.Distributor}
+ */
+@UseLocalDisk
+@TestPlugin(
+ name = "replication",
+ sysModule = "com.googlesource.gerrit.plugins.replication.ReplicationModule")
+public class ReplicationDistributorIT extends ReplicationStorageDaemon {
+ private static final int TEST_DISTRIBUTION_INTERVAL_SECONDS = 3;
+ private static final int TEST_DISTRIBUTION_DURATION_SECONDS = 1;
+ private static final int TEST_DISTRIBUTION_CYCLE_SECONDS =
+ TEST_DISTRIBUTION_INTERVAL_SECONDS + TEST_DISTRIBUTION_DURATION_SECONDS;
+
+ @Override
+ public void setUpTestPlugin() throws Exception {
+ initConfig();
+ setDistributionInterval(TEST_DISTRIBUTION_INTERVAL_SECONDS);
+ super.setUpTestPlugin();
+ }
+
+ @Test
+ public void distributorAddingTaskFromStorage() throws Exception {
+ String remote = "foo";
+ String replica = "replica";
+ String master = "refs/heads/master";
+ String newBranch = "refs/heads/foo_branch";
+ Project.NameKey targetProject = createTestProject(project + replica);
+ ReplicationTasksStorage.ReplicateRefUpdate ref =
+ ReplicationTasksStorage.ReplicateRefUpdate.create(
+ project.get(), newBranch, new URIish(getProjectUri(targetProject)), remote);
+ createBranch(project, master, newBranch);
+ setReplicationDestination(remote, replica, ALL_PROJECTS);
+ reloadConfig();
+
+ tasksStorage.create(ref); // Mimics RefUpdate inserted into storage by other Primary
+ WaitUtil.waitUntil(
+ () -> getProjectTasks().size() != 0, Duration.ofSeconds(TEST_DISTRIBUTION_CYCLE_SECONDS));
+
+ List<WorkQueue.Task<?>> tasks = getProjectTasks();
+ assertThat(tasks).hasSize(1); // ReplicationTask for the created ref in queue
+ assertThat(waitForProjectTaskCount(0, TEST_PUSH_TIMEOUT)).isTrue();
+
+ try (Repository targetRepo = repoManager.openRepository(targetProject);
+ Repository sourceRepo = repoManager.openRepository(project)) {
+ Ref masterRef = getRef(sourceRepo, master);
+ Ref targetBranchRef = getRef(targetRepo, newBranch);
+ assertThat(targetBranchRef).isNotNull();
+ assertThat(targetBranchRef.getObjectId()).isEqualTo(masterRef.getObjectId());
+ }
+ }
+
+ @Test
+ public void distributorPrunesTaskFromWorkQueue() throws Exception {
+ createTestProject(project + "replica");
+ setReplicationDestination("foo", "replica", ALL_PROJECTS, Integer.MAX_VALUE);
+ reloadConfig();
+
+ String newBranch = "refs/heads/foo_branch";
+ createBranch(BranchNameKey.create(project, newBranch));
+
+ assertThat(listWaitingReplicationTasks(newBranch)).hasSize(1);
+ deleteWaitingReplicationTasks(newBranch); // This simulates the work being started by other node
+
+ assertThat(waitForProjectTaskCount(0, Duration.ofSeconds(TEST_DISTRIBUTION_CYCLE_SECONDS)))
+ .isTrue();
+ }
+
+ private List<WorkQueue.Task<?>> getProjectTasks() {
+ return getInstance(WorkQueue.class).getTasks().stream()
+ .filter(t -> t instanceof WorkQueue.ProjectTask)
+ .collect(Collectors.toList());
+ }
+
+ private void createBranch(Project.NameKey project, String fromRef, String refToCreate)
+ throws Exception {
+ try (Repository repo = repoManager.openRepository(project)) {
+ Ref from = repo.exactRef(fromRef);
+ RefUpdate createBranch = repo.updateRef(refToCreate);
+ createBranch.setNewObjectId(from.getObjectId());
+ createBranch.update();
+ }
+ }
+
+ private boolean waitForProjectTaskCount(int count, Duration duration) {
+ try {
+ WaitUtil.waitUntil(() -> getProjectTasks().size() == count, duration);
+ return true;
+ } catch (InterruptedException e) {
+ return false;
+ }
+ }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
index e508b32..e2e1e21 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
@@ -250,26 +250,6 @@
replicateBranchDeletion(false);
}
- private void replicateBranchDeletion(boolean mirror) throws Exception {
- setReplicationDestination("foo", "replica", ALL_PROJECTS);
- reloadConfig();
-
- Project.NameKey targetProject = createTestProject(project + "replica");
- String branchToDelete = "refs/heads/todelete";
- String master = "refs/heads/master";
- BranchInput input = new BranchInput();
- input.revision = master;
- gApi.projects().name(project.get()).branch(branchToDelete).create(input);
- isPushCompleted(targetProject, branchToDelete, TEST_PUSH_TIMEOUT);
-
- setReplicationDestination("foo", "replica", ALL_PROJECTS, Integer.MAX_VALUE, mirror);
- reloadConfig();
-
- gApi.projects().name(project.get()).branch(branchToDelete).delete();
-
- assertThat(listWaitingReplicationTasks(branchToDelete)).hasSize(1);
- }
-
@Test
public void shouldCleanupTasksAfterNewProjectReplication() throws Exception {
setReplicationDestination("task_cleanup_project", "replica", ALL_PROJECTS);
@@ -332,6 +312,26 @@
WaitUtil.waitUntil(() -> isTaskCleanedUp(), TEST_TASK_FINISH_TIMEOUT);
}
+ private void replicateBranchDeletion(boolean mirror) throws Exception {
+ setReplicationDestination("foo", "replica", ALL_PROJECTS);
+ reloadConfig();
+
+ Project.NameKey targetProject = createTestProject(project + "replica");
+ String branchToDelete = "refs/heads/todelete";
+ String master = "refs/heads/master";
+ BranchInput input = new BranchInput();
+ input.revision = master;
+ gApi.projects().name(project.get()).branch(branchToDelete).create(input);
+ isPushCompleted(targetProject, branchToDelete, TEST_PUSH_TIMEOUT);
+
+ setReplicationDestination("foo", "replica", ALL_PROJECTS, Integer.MAX_VALUE, mirror);
+ reloadConfig();
+
+ gApi.projects().name(project.get()).branch(branchToDelete).delete();
+
+ assertThat(listWaitingReplicationTasks(branchToDelete)).hasSize(1);
+ }
+
private boolean isTaskRescheduled(QueueInfo queue, URIish uri) {
PushOne pushOne = queue.pending.get(uri);
return pushOne == null ? false : pushOne.isRetrying();
