Run synchronous fetch as a single FetchOp for all refs in a batch Following the introduction of the batch-fetch api [1], the work on the synchronous fetch side was outstanding. The payload of the batch fetch contains a list of all the refs in the batch, for which we want to execute a single git fetch operation. With regards to the async operations, all the scheduling would end up in the same replication task as they have in common the same URI and repository. [1] https://gerrit-review.googlesource.com/c/plugins/pull-replication/+/378496 Bug: Issue 303112557 Change-Id: Ifd2a28a6337f762d89d744934f44b7f2b47e6e6a
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java index 218592b..0a55043 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java
@@ -537,21 +537,23 @@ public Optional<FetchOne> fetchSync( Project.NameKey project, - String ref, + Set<String> refs, URIish uri, - ReplicationState state, Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) { - if (shouldReplicate(project, ref) - && (config.replicatePermissions() || !ref.equals(RefNames.REFS_CONFIG))) { + Set<String> refsToReplicate = + refs.stream() + .filter(ref -> shouldReplicate(project, ref)) + .filter(ref -> config.replicatePermissions() || !ref.equals(RefNames.REFS_CONFIG)) + .collect(Collectors.toUnmodifiableSet()); - FetchOne e = opFactory.create(project, uri, apiRequestMetrics); - e.addRef(ref); - e.addState(ref, state); - e.runSync(); - return Optional.of(e); + if (refsToReplicate.isEmpty()) { + return Optional.empty(); } - return Optional.empty(); + FetchOne e = opFactory.create(project, uri, apiRequestMetrics); + e.addRefs(refsToReplicate); + e.runSync(); + return Optional.of(e); } void scheduleDeleteProject(String uri, Project.NameKey project) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchAction.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchAction.java index c6ad47d..b7b1ab1 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchAction.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchAction.java
@@ -20,12 +20,11 @@ import com.google.gerrit.server.project.ProjectResource; import com.google.inject.Inject; import com.google.inject.Singleton; -import com.googlesource.gerrit.plugins.replication.pull.api.FetchAction.Input; -import java.util.ArrayList; +import com.googlesource.gerrit.plugins.replication.pull.api.FetchAction.BatchInput; import java.util.List; @Singleton -public class BatchFetchAction implements RestModifyView<ProjectResource, List<Input>> { +public class BatchFetchAction implements RestModifyView<ProjectResource, List<FetchAction.Input>> { private final FetchAction fetchAction; @Inject @@ -34,14 +33,10 @@ } @Override - public Response<?> apply(ProjectResource resource, List<Input> inputs) throws RestApiException { - - List<Response<?>> allResponses = new ArrayList<>(); - for (Input input : inputs) { - Response<?> res = fetchAction.apply(resource, input); - allResponses.add(res); - } - - return Response.ok(allResponses); + public Response<?> apply(ProjectResource resource, List<FetchAction.Input> inputs) + throws RestApiException { + return Response.ok( + fetchAction.apply( + resource, BatchInput.fromInput(inputs.toArray(new FetchAction.Input[0])))); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java index 9e69f8d..d57ae0c 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java
@@ -27,6 +27,7 @@ import com.google.gerrit.extensions.restapi.UnprocessableEntityException; import com.google.gerrit.server.config.UrlFormatter; import com.google.gerrit.server.git.WorkQueue; +import com.google.gerrit.server.git.WorkQueue.Task; import com.google.gerrit.server.ioutil.HexFormat; import com.google.gerrit.server.project.ProjectResource; import com.google.inject.Inject; @@ -35,8 +36,11 @@ import com.googlesource.gerrit.plugins.replication.pull.api.FetchJob.Factory; import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException; import java.util.Optional; +import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.eclipse.jgit.errors.TransportException; @Singleton @@ -67,25 +71,50 @@ public boolean async; } + public static class BatchInput { + public String label; + public Set<String> refsNames; + public boolean async; + + static BatchInput fromInput(Input... input) { + BatchInput batchInput = new BatchInput(); + batchInput.async = input[0].async; + batchInput.label = input[0].label; + batchInput.refsNames = Stream.of(input).map(i -> i.refName).collect(Collectors.toSet()); + return batchInput; + } + } + @Override public Response<?> apply(ProjectResource resource, Input input) throws RestApiException { + return apply(resource, BatchInput.fromInput(input)); + } + + public Response<?> apply(ProjectResource resource, BatchInput batchInput) + throws RestApiException { if (!preConditions.canCallFetchApi()) { throw new AuthException("not allowed to call fetch command"); } try { - if (Strings.isNullOrEmpty(input.label)) { + if (Strings.isNullOrEmpty(batchInput.label)) { throw new BadRequestException("Source label cannot be null or empty"); } - if (Strings.isNullOrEmpty(input.refName)) { + if (batchInput.refsNames.isEmpty()) { throw new BadRequestException("Ref-update refname cannot be null or empty"); } - if (input.async) { - return applyAsync(resource.getNameKey(), input); + for (String refName : batchInput.refsNames) { + if (Strings.isNullOrEmpty(refName)) { + throw new BadRequestException("Ref-update refname cannot be null or empty"); + } } - return applySync(resource.getNameKey(), input); + + if (batchInput.async) { + return applyAsync(resource.getNameKey(), batchInput); + } + return applySync(resource.getNameKey(), batchInput); } catch (InterruptedException | ExecutionException | IllegalStateException @@ -97,22 +126,32 @@ } } - private Response<?> applySync(Project.NameKey project, Input input) + private Response<?> applySync(Project.NameKey project, BatchInput input) throws InterruptedException, ExecutionException, RemoteConfigurationMissingException, TimeoutException, TransportException { - command.fetchSync(project, input.label, input.refName); + command.fetchSync(project, input.label, input.refsNames); return Response.created(input); } - private Response.Accepted applyAsync(Project.NameKey project, Input input) { - @SuppressWarnings("unchecked") - WorkQueue.Task<Void> task = - (WorkQueue.Task<Void>) - workQueue - .getDefaultQueue() - .submit( - fetchJobFactory.create(project, input, PullReplicationApiRequestMetrics.get())); - Optional<String> url = + @SuppressWarnings("unchecked") + private Response.Accepted applyAsync(Project.NameKey project, BatchInput batchInput) { + WorkQueue.Task<Void> task = null; + Optional<String> url; + + for (String refName : batchInput.refsNames) { + Input input = new Input(); + input.label = batchInput.label; + input.async = batchInput.async; + input.refName = refName; + task = + (Task<Void>) + workQueue + .getDefaultQueue() + .submit( + fetchJobFactory.create( + project, input, PullReplicationApiRequestMetrics.get())); + } + url = urlFormatter .get() .getRestUrl("a/config/server/tasks/" + HexFormat.fromInt(task.getTaskId()));
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java index 5323425..8b86965 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java
@@ -30,8 +30,10 @@ import com.googlesource.gerrit.plugins.replication.pull.Source; import com.googlesource.gerrit.plugins.replication.pull.SourcesCollection; import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException; +import java.util.ArrayList; import java.util.List; import java.util.Optional; +import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; @@ -66,19 +68,19 @@ PullReplicationApiRequestMetrics apiRequestMetrics) throws InterruptedException, ExecutionException, RemoteConfigurationMissingException, TimeoutException, TransportException { - fetch(name, label, refName, ASYNC, Optional.of(apiRequestMetrics)); + fetch(name, label, Set.of(refName), ASYNC, Optional.of(apiRequestMetrics)); } - public void fetchSync(Project.NameKey name, String label, String refName) + public void fetchSync(Project.NameKey name, String label, Set<String> refsNames) throws InterruptedException, ExecutionException, RemoteConfigurationMissingException, TimeoutException, TransportException { - fetch(name, label, refName, SYNC, Optional.empty()); + fetch(name, label, refsNames, SYNC, Optional.empty()); } private void fetch( Project.NameKey name, String label, - String refName, + Set<String> refsNames, ReplicationType fetchType, Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) throws InterruptedException, ExecutionException, RemoteConfigurationMissingException, @@ -96,18 +98,21 @@ try { if (fetchType == ReplicationType.ASYNC) { state.markAllFetchTasksScheduled(); - Future<?> future = source.get().schedule(name, refName, state, apiRequestMetrics); + List<Future<?>> futures = new ArrayList<>(); + for (String refName : refsNames) { + futures.add(source.get().schedule(name, refName, state, apiRequestMetrics)); + } int timeout = source.get().getTimeout(); - if (timeout == 0) { - future.get(); - } else { - future.get(timeout, TimeUnit.SECONDS); + for (Future future : futures) { + if (timeout == 0) { + future.get(); + } else { + future.get(timeout, TimeUnit.SECONDS); + } } } else { Optional<FetchOne> maybeFetch = - source - .get() - .fetchSync(name, refName, source.get().getURI(name), state, apiRequestMetrics); + source.get().fetchSync(name, refsNames, source.get().getURI(name), apiRequestMetrics); if (maybeFetch.map(FetchOne::getFetchRefSpecs).filter(List::isEmpty).isPresent()) { fetchStateLog.warn( String.format(
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchActionTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchActionTest.java index 738815a..e2e4cd3 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchActionTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/BatchFetchActionTest.java
@@ -19,6 +19,7 @@ import static com.google.common.truth.Truth.assertThat; import static org.apache.http.HttpStatus.SC_OK; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.eq; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -51,14 +52,14 @@ } @Test - public void shouldDelegateToFetchActionForEveryFetchInput() throws RestApiException { + public void shouldDelegateToFetchActionWithBatchInputForListOfFetchInput() + throws RestApiException { FetchAction.Input first = createInput(master); FetchAction.Input second = createInput(test); batchFetchAction.apply(projectResource, List.of(first, second)); - verify(fetchAction).apply(projectResource, first); - verify(fetchAction).apply(projectResource, second); + verify(fetchAction).apply(eq(projectResource), any(FetchAction.BatchInput.class)); } @Test @@ -67,7 +68,7 @@ FetchAction.Input first = createInput(master); FetchAction.Input second = createInput(test); - when(fetchAction.apply(any(), any())) + when(fetchAction.apply(any(ProjectResource.class), any(FetchAction.BatchInput.class))) .thenAnswer((Answer<Response<?>>) invocation -> Response.accepted("some-url")); Response<?> response = batchFetchAction.apply(projectResource, List.of(first, second)); @@ -75,40 +76,18 @@ } @Test - public void shouldReturnAListWithAllResponsesOnSuccess() throws RestApiException { + public void shouldReturnAResponsesOnSuccess() throws RestApiException { FetchAction.Input first = createInput(master); FetchAction.Input second = createInput(test); String masterUrl = "master-url"; String testUrl = "test-url"; - Response.Accepted firstResponse = Response.accepted(masterUrl); - Response.Accepted secondResponse = Response.accepted(testUrl); + Response.Accepted batchResponse = Response.accepted(masterUrl); - when(fetchAction.apply(projectResource, first)) - .thenAnswer((Answer<Response<?>>) invocation -> firstResponse); - when(fetchAction.apply(projectResource, second)) - .thenAnswer((Answer<Response<?>>) invocation -> secondResponse); + when(fetchAction.apply(eq(projectResource), any(FetchAction.BatchInput.class))) + .thenAnswer((Answer<Response<?>>) invocation -> batchResponse); Response<?> response = batchFetchAction.apply(projectResource, List.of(first, second)); - assertThat((List<Response<?>>) response.value()) - .isEqualTo(List.of(firstResponse, secondResponse)); - } - - @Test - public void shouldReturnAMixOfSyncAndAsyncResponses() throws RestApiException { - FetchAction.Input async = createInput(master); - FetchAction.Input sync = createInput(test); - String masterUrl = "master-url"; - Response.Accepted asyncResponse = Response.accepted(masterUrl); - Response<?> syncResponse = Response.created(sync); - - when(fetchAction.apply(projectResource, async)) - .thenAnswer((Answer<Response<?>>) invocation -> asyncResponse); - when(fetchAction.apply(projectResource, sync)) - .thenAnswer((Answer<Response<?>>) invocation -> syncResponse); - Response<?> response = batchFetchAction.apply(projectResource, List.of(async, sync)); - - assertThat((List<Response<?>>) response.value()) - .isEqualTo(List.of(asyncResponse, syncResponse)); + assertThat(response.value()).isEqualTo(batchResponse); } @Test(expected = RestApiException.class) @@ -117,9 +96,8 @@ FetchAction.Input second = createInput(test); String masterUrl = "master-url"; - when(fetchAction.apply(projectResource, first)) - .thenAnswer((Answer<Response<?>>) invocation -> Response.accepted(masterUrl)); - when(fetchAction.apply(projectResource, second)).thenThrow(new MergeConflictException("BOOM")); + when(fetchAction.apply(eq(projectResource), any(FetchAction.BatchInput.class))) + .thenThrow(new MergeConflictException("BOOM")); batchFetchAction.apply(projectResource, List.of(first, second)); }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java index 024a712..9653209 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java
@@ -34,6 +34,7 @@ import com.google.gerrit.server.project.ProjectResource; import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException; import java.util.Optional; +import java.util.Set; import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeoutException; @@ -51,6 +52,7 @@ String label = "instance-2-label"; String url = "file:///gerrit-host/instance-1/git/${name}.git"; String refName = "refs/heads/master"; + String altRefName = "refs/heads/alt"; String location = "http://gerrit-host/a/config/server/tasks/08d173e9"; int taskId = 1234; @@ -88,7 +90,7 @@ } @Test - public void shouldReturnCreatedResponseCode() throws Exception { + public void shouldReturnCreatedResponseCodeForSingleRefFetchAction() throws Exception { FetchAction.Input inputParams = new FetchAction.Input(); inputParams.label = label; inputParams.refName = refName; @@ -98,6 +100,17 @@ assertThat(response.statusCode()).isEqualTo(SC_CREATED); } + @Test + public void shouldReturnCreatedResponseCodeForBatchRefFetchAction() throws Exception { + FetchAction.BatchInput batchInputParams = new FetchAction.BatchInput(); + batchInputParams.label = label; + batchInputParams.refsNames = Set.of(refName, altRefName); + + Response<?> response = fetchAction.apply(projectResource, batchInputParams); + + assertThat(response.statusCode()).isEqualTo(SC_CREATED); + } + @SuppressWarnings("cast") @Test public void shouldReturnSourceUrlAndrefNameAsAResponseBody() throws Exception { @@ -107,7 +120,11 @@ Response<?> response = fetchAction.apply(projectResource, inputParams); - assertThat((FetchAction.Input) response.value()).isEqualTo(inputParams); + FetchAction.BatchInput responseBatchInput = (FetchAction.BatchInput) response.value(); + + assertThat(responseBatchInput.label).isEqualTo(inputParams.label); + assertThat(responseBatchInput.async).isEqualTo(inputParams.async); + assertThat(responseBatchInput.refsNames).containsExactly(inputParams.refName); } @Test(expected = BadRequestException.class)
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java index 777350b..f8d12a9 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java
@@ -18,7 +18,6 @@ import static org.mockito.Mockito.any; import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.eq; -import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -32,6 +31,7 @@ import com.googlesource.gerrit.plugins.replication.pull.SourcesCollection; import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException; import java.util.Optional; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Future; import org.eclipse.jgit.transport.URIish; @@ -44,6 +44,9 @@ @RunWith(MockitoJUnitRunner.class) public class FetchCommandTest { private static final String REF_NAME_TO_FETCH = "refs/heads/master"; + private static final String ALT_REF_NAME_TO_FETCH = "refs/heads/alt"; + private static final Set<String> REFS_NAMES_TO_FETCH = + Set.of(REF_NAME_TO_FETCH, ALT_REF_NAME_TO_FETCH); @Mock ReplicationState state; @Mock ReplicationState.Factory fetchReplicationStateFactory; @Mock PullReplicationStateLogger fetchStateLog; @@ -77,6 +80,11 @@ } @Test + public void shouldScheduleRefFetch() throws Exception { + objectUnderTest.fetchSync(projectName, label, REFS_NAMES_TO_FETCH); + } + + @Test public void shouldScheduleRefFetchWithDelay() throws Exception { objectUnderTest.fetchAsync(projectName, label, REF_NAME_TO_FETCH, apiRequestMetrics); @@ -87,18 +95,24 @@ @Test public void shouldNotScheduleAsyncTaskWhenFetchSync() throws Exception { - objectUnderTest.fetchSync(projectName, label, REF_NAME_TO_FETCH); + objectUnderTest.fetchSync(projectName, label, REFS_NAMES_TO_FETCH); + verify(source, times(1)).fetchSync(projectName, REFS_NAMES_TO_FETCH, null, Optional.empty()); + } - verify(source, never()) - .schedule( - eq(projectName), eq(REF_NAME_TO_FETCH), eq(state), eq(Optional.of(apiRequestMetrics))); + @Test + public void shouldMarkAllFetchTasksScheduled() throws Exception { + objectUnderTest.fetchAsync(projectName, label, REF_NAME_TO_FETCH, apiRequestMetrics); + + verify(source, times(1)) + .schedule(projectName, REF_NAME_TO_FETCH, state, Optional.of(apiRequestMetrics)); + verify(state, times(1)).markAllFetchTasksScheduled(); } @Test public void shouldUpdateStateWhenRemoteConfigNameIsMissing() { assertThrows( RemoteConfigurationMissingException.class, - () -> objectUnderTest.fetchSync(projectName, "unknownLabel", REF_NAME_TO_FETCH)); + () -> objectUnderTest.fetchSync(projectName, "unknownLabel", REFS_NAMES_TO_FETCH)); verify(fetchStateLog, times(1)).error(anyString(), eq(state)); } }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilterTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilterTest.java index 34885b2..bba638c 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilterTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilterTest.java
@@ -130,13 +130,14 @@ .getBytes(StandardCharsets.UTF_8); defineBehaviours(payloadFetch, FETCH_URI); - when(fetchAction.apply(any(), any())).thenReturn(OK_RESPONSE); + when(fetchAction.apply(any(ProjectResource.class), any(FetchAction.Input.class))) + .thenReturn(OK_RESPONSE); PullReplicationFilter pullReplicationFilter = createPullReplicationFilter(); pullReplicationFilter.doFilter(request, response, filterChain); verifyBehaviours(); - verify(fetchAction).apply(any(ProjectResource.class), any()); + verify(fetchAction).apply(any(ProjectResource.class), any(FetchAction.Input.class)); } @Test @@ -314,7 +315,7 @@ .getBytes(StandardCharsets.UTF_8); defineBehaviours(payloadFetchAction, FETCH_URI); - when(fetchAction.apply(any(), any())) + when(fetchAction.apply(any(ProjectResource.class), any(FetchAction.Input.class))) .thenThrow(new AuthException("The user is not authorised")); when(response.getOutputStream()).thenReturn(outputStream); @@ -348,7 +349,7 @@ .getBytes(StandardCharsets.UTF_8); defineBehaviours(payloadFetchAction, FETCH_URI); - when(fetchAction.apply(any(), any())) + when(fetchAction.apply(any(), any(FetchAction.Input.class))) .thenThrow(new UnprocessableEntityException("Entity cannot be processed")); when(response.getOutputStream()).thenReturn(outputStream);