Introduce E2E fetch REST-API metrics
The E2E replication through the fetch REST-API
needs to be calculated from the client originating the call
where the initial ref-update happens.
Introduce a new end_2_end metric to track this latency.
When the pull replication is executed asynchronously,
the client originating the call propagates the timestamp
in an HTTP header (X-StartTimeNanos) and propagated
to the receiver side, which will use the initial time for
recording the final E2E replication time at the receiver
side.
Also add the pull-replication async integration tests
that were completely missing before: testing with async
replication is key for demonstrating that the E2E metrics
are working as expected.
Consequently, the TEST_REPLICATION_DELAY is reduced to
1 second for preventing the whole test suite to fail
for timeout.
TODO: Additional assertion on E2E metrics in sync and async
modes.
Change-Id: Ieb29e3475dc0938fe7700007b7d05a630c3c927f
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java
index bcd86d7..42310ff 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java
@@ -21,6 +21,7 @@
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
import com.googlesource.gerrit.plugins.replication.ReplicationFilter;
+import java.util.Optional;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.eclipse.jgit.transport.URIish;
@@ -91,7 +92,7 @@
for (Source cfg : sources.getAll()) {
if (cfg.wouldFetchProject(project)) {
for (URIish uri : cfg.getURIs(project, urlMatch)) {
- cfg.schedule(project, FetchOne.ALL_REFS, uri, state, replicationType);
+ cfg.schedule(project, FetchOne.ALL_REFS, uri, state, replicationType, Optional.empty());
}
}
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
index cee52b5..cdf573f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
@@ -32,6 +32,7 @@
import com.google.gerrit.server.util.IdGenerator;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
+import com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationApiRequestMetrics;
import com.googlesource.gerrit.plugins.replication.pull.fetch.Fetch;
import com.googlesource.gerrit.plugins.replication.pull.fetch.FetchFactory;
import com.googlesource.gerrit.plugins.replication.pull.fetch.RefUpdateState;
@@ -69,7 +70,8 @@
static final String ID_KEY = "fetchOneId";
interface Factory {
- FetchOne create(Project.NameKey d, URIish u);
+ FetchOne create(
+ Project.NameKey d, URIish u, Optional<PullReplicationApiRequestMetrics> apiRequestMetrics);
}
private final GitRepositoryManager gitManager;
@@ -94,6 +96,7 @@
private final FetchReplicationMetrics metrics;
private final AtomicBoolean canceledWhileRunning;
private final FetchFactory fetchFactory;
+ private final Optional<PullReplicationApiRequestMetrics> apiRequestMetrics;
@Inject
FetchOne(
@@ -106,7 +109,8 @@
FetchReplicationMetrics m,
FetchFactory fetchFactory,
@Assisted Project.NameKey d,
- @Assisted URIish u) {
+ @Assisted URIish u,
+ @Assisted Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
gitManager = grm;
pool = s;
config = c.getRemoteConfig();
@@ -122,6 +126,7 @@
canceledWhileRunning = new AtomicBoolean(false);
this.fetchFactory = fetchFactory;
maxRetries = s.getMaxRetries();
+ this.apiRequestMetrics = apiRequestMetrics;
}
@Override
@@ -299,12 +304,17 @@
git = gitManager.openRepository(projectName);
runImpl();
long elapsed = NANOSECONDS.toMillis(context.stop());
+ Optional<Long> elapsedEnd2End =
+ apiRequestMetrics
+ .flatMap(metrics -> metrics.stop(config.getName()))
+ .map(NANOSECONDS::toMillis);
repLog.info(
- "Replication from {} completed in {}ms, {}ms delay, {} retries",
+ "Replication from {} completed in {}ms, {}ms delay, {} retries{}",
uri,
elapsed,
delay,
- retryCount);
+ retryCount,
+ elapsedEnd2End.map(el -> String.format(", E2E %dms", el)).orElse(""));
} catch (RepositoryNotFoundException e) {
stateLog.error(
"Cannot replicate " + projectName + "; Local repository error: " + e.getMessage(),
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchReplicationMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchReplicationMetrics.java
index e952252..22bb073 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchReplicationMetrics.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchReplicationMetrics.java
@@ -23,10 +23,12 @@
import com.google.gerrit.server.logging.PluginMetadata;
import com.google.inject.Inject;
import com.google.inject.Singleton;
+import java.util.concurrent.TimeUnit;
@Singleton
public class FetchReplicationMetrics {
private final Timer1<String> executionTime;
+ private final Timer1<String> end2EndExecutionTime;
private final Histogram1<String> executionDelay;
private final Histogram1<String> executionRetries;
@@ -34,11 +36,11 @@
FetchReplicationMetrics(@PluginName String pluginName, MetricMaker metricMaker) {
Field<String> SOURCE_FIELD =
Field.ofString(
- "source",
+ "pull_replication",
(metadataBuilder, fieldValue) ->
metadataBuilder
.pluginName(pluginName)
- .addPluginMetadata(PluginMetadata.create("source", fieldValue)))
+ .addPluginMetadata(PluginMetadata.create("pull_replication", fieldValue)))
.build();
executionTime =
@@ -49,6 +51,14 @@
.setUnit(Description.Units.MILLISECONDS),
SOURCE_FIELD);
+ end2EndExecutionTime =
+ metricMaker.newTimer(
+ "replication_end_2_end_latency",
+ new Description("Time spent end-2-end fetching from remote source.")
+ .setCumulative()
+ .setUnit(Description.Units.MILLISECONDS),
+ SOURCE_FIELD);
+
executionDelay =
metricMaker.newHistogram(
"replication_delay",
@@ -77,6 +87,26 @@
}
/**
+ * Start the end-to-end replication latency timer from a source.
+ *
+ * @param name the source name.
+ * @return the timer context.
+ */
+ public Timer1.Context<String> startEnd2End(String name) {
+ return end2EndExecutionTime.start(name);
+ }
+
+ /**
+ * Record the end-to-end replication latency timer from a source.
+ *
+ * @param name the source name.
+ * @param metricNanos the timer value in nanos
+ */
+ public void recordEnd2End(String name, long metricNanos) {
+ end2EndExecutionTime.record(name, metricNanos, TimeUnit.NANOSECONDS);
+ }
+
+ /**
* Record the replication delay and retry metrics for a source.
*
* @param name the source name.
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java
index 6393c08..431255c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java
@@ -77,6 +77,7 @@
private ExcludedRefsFilter refsFilter;
private RevisionReader revisionReader;
private final ApplyObjectMetrics applyObjectMetrics;
+ private final FetchReplicationMetrics fetchMetrics;
@Inject
ReplicationQueue(
@@ -87,7 +88,8 @@
FetchApiClient.Factory fetchClientFactory,
ExcludedRefsFilter refsFilter,
RevisionReader revReader,
- ApplyObjectMetrics applyObjectMetrics) {
+ ApplyObjectMetrics applyObjectMetrics,
+ FetchReplicationMetrics fetchMetrics) {
workQueue = wq;
dispatcher = dis;
sources = rd;
@@ -97,6 +99,7 @@
this.refsFilter = refsFilter;
this.revisionReader = revReader;
this.applyObjectMetrics = applyObjectMetrics;
+ this.fetchMetrics = fetchMetrics;
}
@Override
@@ -349,7 +352,18 @@
try {
URIish uri = new URIish(apiUrl);
FetchApiClient fetchClient = fetchClientFactory.create(source);
- HttpResult result = fetchClient.callFetch(project, refName, uri);
+ repLog.info("Pull replication REST API fetch to {} for {}:{}", apiUrl, project, refName);
+ Context<String> timer = fetchMetrics.startEnd2End(source.getRemoteConfigName());
+ HttpResult result = fetchClient.callFetch(project, refName, uri, timer.getStartTime());
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(timer.stop());
+ repLog.info(
+ "Pull replication REST API fetch to {} COMPLETED for {}:{}, HTTP Result:"
+ + " {} - time:{} ms",
+ apiUrl,
+ project,
+ refName,
+ result,
+ elapsedMs);
if (isProjectMissing(result, project) && source.isCreateMissingRepositories()) {
result = initProject(project, uri, fetchClient, result);
}
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 ddefa47..4ae3f7d 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
@@ -61,6 +61,7 @@
import com.google.inject.servlet.RequestScoped;
import com.googlesource.gerrit.plugins.replication.RemoteSiteUser;
import com.googlesource.gerrit.plugins.replication.ReplicationFilter;
+import com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationApiRequestMetrics;
import com.googlesource.gerrit.plugins.replication.pull.fetch.BatchFetchClient;
import com.googlesource.gerrit.plugins.replication.pull.fetch.CGitFetch;
import com.googlesource.gerrit.plugins.replication.pull.fetch.CGitFetchValidator;
@@ -392,9 +393,10 @@
Project.NameKey project,
String ref,
ReplicationState state,
- ReplicationType replicationType) {
+ ReplicationType replicationType,
+ Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
URIish uri = getURI(project);
- return schedule(project, ref, uri, state, replicationType);
+ return schedule(project, ref, uri, state, replicationType, apiRequestMetrics);
}
public Future<?> schedule(
@@ -402,7 +404,8 @@
String ref,
URIish uri,
ReplicationState state,
- ReplicationType replicationType) {
+ ReplicationType replicationType,
+ Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
repLog.info("scheduling replication {}:{} => {}", uri, ref, project);
if (!shouldReplicate(project, ref, state)) {
@@ -438,7 +441,7 @@
FetchOne e = pending.get(uri);
Future<?> f = CompletableFuture.completedFuture(null);
if (e == null) {
- e = opFactory.create(project, uri);
+ e = opFactory.create(project, uri, apiRequestMetrics);
addRef(e, ref);
e.addState(ref, state);
pending.put(uri, e);
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 2767f1d..48964db 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
@@ -101,7 +101,10 @@
@SuppressWarnings("unchecked")
WorkQueue.Task<Void> task =
(WorkQueue.Task<Void>)
- workQueue.getDefaultQueue().submit(new FetchJob(command, project, input));
+ workQueue
+ .getDefaultQueue()
+ .submit(
+ new FetchJob(command, project, input, PullReplicationApiRequestMetrics.get()));
Optional<String> url =
urlFormatter
.get()
@@ -117,17 +120,23 @@
private FetchCommand command;
private Project.NameKey project;
private FetchAction.Input input;
+ private final PullReplicationApiRequestMetrics apiRequestMetrics;
- public FetchJob(FetchCommand command, Project.NameKey project, FetchAction.Input input) {
+ public FetchJob(
+ FetchCommand command,
+ Project.NameKey project,
+ FetchAction.Input input,
+ PullReplicationApiRequestMetrics apiRequestMetrics) {
this.command = command;
this.project = project;
this.input = input;
+ this.apiRequestMetrics = apiRequestMetrics;
}
@Override
public void run() {
try {
- command.fetchAsync(project, input.label, input.refName);
+ command.fetchAsync(project, input.label, input.refName, apiRequestMetrics);
} catch (InterruptedException
| ExecutionException
| RemoteConfigurationMissingException
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 e1ac9ac..3a502ef 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
@@ -54,19 +54,28 @@
this.eventDispatcher = eventDispatcher;
}
- public void fetchAsync(Project.NameKey name, String label, String refName)
+ public void fetchAsync(
+ Project.NameKey name,
+ String label,
+ String refName,
+ PullReplicationApiRequestMetrics apiRequestMetrics)
throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
TimeoutException {
- fetch(name, label, refName, ASYNC);
+ fetch(name, label, refName, ASYNC, Optional.of(apiRequestMetrics));
}
public void fetchSync(Project.NameKey name, String label, String refName)
throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
TimeoutException {
- fetch(name, label, refName, SYNC);
+ fetch(name, label, refName, SYNC, Optional.empty());
}
- private void fetch(Project.NameKey name, String label, String refName, ReplicationType fetchType)
+ private void fetch(
+ Project.NameKey name,
+ String label,
+ String refName,
+ ReplicationType fetchType,
+ Optional<PullReplicationApiRequestMetrics> apiRequestMetrics)
throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
TimeoutException {
ReplicationState state =
@@ -82,7 +91,7 @@
try {
state.markAllFetchTasksScheduled();
- Future<?> future = source.get().schedule(name, refName, state, fetchType);
+ Future<?> future = source.get().schedule(name, refName, state, fetchType, apiRequestMetrics);
future.get(source.get().getTimeout(), TimeUnit.SECONDS);
} catch (ExecutionException
| IllegalStateException
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java
index b2ef28d..b140cb4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java
@@ -38,5 +38,9 @@
} else {
serveRegex("/init-project/.*$").with(ProjectInitializationAction.class);
}
+
+ DynamicSet.bind(binder(), AllRequestFilter.class)
+ .to(PullReplicationApiMetricsFilter.class)
+ .in(Scopes.SINGLETON);
}
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiMetricsFilter.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiMetricsFilter.java
new file mode 100644
index 0000000..3858db2
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiMetricsFilter.java
@@ -0,0 +1,53 @@
+// Copyright (C) 2022 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.pull.api;
+
+import com.google.gerrit.httpd.AllRequestFilter;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.google.inject.Singleton;
+import java.io.IOException;
+import javax.servlet.FilterChain;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+@Singleton
+public class PullReplicationApiMetricsFilter extends AllRequestFilter {
+ private final Provider<PullReplicationApiRequestMetrics> apiRequestMetrics;
+
+ @Inject
+ public PullReplicationApiMetricsFilter(
+ Provider<PullReplicationApiRequestMetrics> apiRequestMetrics) {
+ this.apiRequestMetrics = apiRequestMetrics;
+ }
+
+ @Override
+ public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
+ throws IOException, ServletException {
+ if (!(request instanceof HttpServletRequest) || !(response instanceof HttpServletResponse)) {
+ chain.doFilter(request, response);
+ return;
+ }
+
+ PullReplicationApiRequestMetrics requestMetrics = apiRequestMetrics.get();
+ requestMetrics.start((HttpServletRequest) request);
+ PullReplicationApiRequestMetrics.set(requestMetrics);
+
+ chain.doFilter(request, response);
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiRequestMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiRequestMetrics.java
new file mode 100644
index 0000000..8a97901
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiRequestMetrics.java
@@ -0,0 +1,68 @@
+// Copyright (C) 2022 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.pull.api;
+
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.replication.pull.FetchReplicationMetrics;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.servlet.http.HttpServletRequest;
+
+public class PullReplicationApiRequestMetrics {
+ private static final ThreadLocal<PullReplicationApiRequestMetrics> localApiRequestMetrics =
+ new ThreadLocal<>();
+
+ public static final String HTTP_HEADER_X_START_TIME_NANOS = "X-StartTimeNanos";
+
+ private Optional<Long> startTimeNanos;
+ private final AtomicBoolean initialised = new AtomicBoolean();
+ private final FetchReplicationMetrics metrics;
+
+ public static PullReplicationApiRequestMetrics get() {
+ return localApiRequestMetrics.get();
+ }
+
+ public static void set(PullReplicationApiRequestMetrics metrics) {
+ localApiRequestMetrics.set(metrics);
+ }
+
+ @Inject
+ public PullReplicationApiRequestMetrics(FetchReplicationMetrics metrics) {
+ this.metrics = metrics;
+ }
+
+ public void start(HttpServletRequest req) {
+ if (!initialised.compareAndSet(false, true)) {
+ throw new IllegalStateException("PullReplicationApiRequestMetrics already initialised");
+ }
+
+ startTimeNanos =
+ Optional.ofNullable(req.getHeader(HTTP_HEADER_X_START_TIME_NANOS))
+ .map(Long::parseLong)
+ /* Adjust with System.nanoTime() for preventing negative execution times
+ * due to a clock skew between the client and the server timestamp.
+ */
+ .map(nanoTime -> Math.min(System.nanoTime(), nanoTime));
+ }
+
+ public Optional<Long> stop(String replicationSourceName) {
+ return startTimeNanos.map(
+ start -> {
+ long elapsed = System.nanoTime() - start;
+ metrics.recordEnd2End(replicationSourceName, elapsed);
+ return elapsed;
+ });
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchApiClient.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchApiClient.java
index 476a35b..01b6f1d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchApiClient.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchApiClient.java
@@ -27,9 +27,15 @@
FetchApiClient create(Source source);
}
- HttpResult callFetch(Project.NameKey project, String refName, URIish targetUri)
+ HttpResult callFetch(
+ Project.NameKey project, String refName, URIish targetUri, long startTimeNanos)
throws ClientProtocolException, IOException;
+ default HttpResult callFetch(Project.NameKey project, String refName, URIish targetUri)
+ throws ClientProtocolException, IOException {
+ return callFetch(project, refName, targetUri, System.nanoTime());
+ }
+
HttpResult initProject(Project.NameKey project, URIish uri) throws IOException;
HttpResult deleteProject(Project.NameKey project, URIish apiUri) throws IOException;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
index f2f57cd..4db4ceb 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
@@ -32,6 +32,7 @@
import com.googlesource.gerrit.plugins.replication.CredentialsFactory;
import com.googlesource.gerrit.plugins.replication.ReplicationConfig;
import com.googlesource.gerrit.plugins.replication.pull.Source;
+import com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationApiRequestMetrics;
import com.googlesource.gerrit.plugins.replication.pull.api.data.RevisionData;
import com.googlesource.gerrit.plugins.replication.pull.api.data.RevisionInput;
import com.googlesource.gerrit.plugins.replication.pull.filter.SyncRefsFilter;
@@ -95,7 +96,8 @@
* @see com.googlesource.gerrit.plugins.replication.pull.client.FetchApiClient#callFetch(com.google.gerrit.entities.Project.NameKey, java.lang.String, org.eclipse.jgit.transport.URIish)
*/
@Override
- public HttpResult callFetch(Project.NameKey project, String refName, URIish targetUri)
+ public HttpResult callFetch(
+ Project.NameKey project, String refName, URIish targetUri, long startTimeNanos)
throws ClientProtocolException, IOException {
String url =
String.format(
@@ -110,6 +112,9 @@
instanceLabel, refName, callAsync),
StandardCharsets.UTF_8));
post.addHeader(new BasicHeader("Content-Type", "application/json"));
+ post.addHeader(
+ PullReplicationApiRequestMetrics.HTTP_HEADER_X_START_TIME_NANOS,
+ Long.toString(startTimeNanos));
return httpClientFactory.create(source).execute(post, this, getContext(targetUri));
}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationAsyncIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationAsyncIT.java
new file mode 100644
index 0000000..58be58e
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationAsyncIT.java
@@ -0,0 +1,43 @@
+// Copyright (C) 2022 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.pull;
+
+import com.google.gerrit.acceptance.SkipProjectClone;
+import com.google.gerrit.acceptance.TestPlugin;
+import com.google.gerrit.acceptance.UseLocalDisk;
+import com.google.gerrit.server.config.SitePaths;
+import com.google.inject.Inject;
+import org.eclipse.jgit.storage.file.FileBasedConfig;
+import org.eclipse.jgit.util.FS;
+
+@SkipProjectClone
+@UseLocalDisk
+@TestPlugin(
+ name = "pull-replication",
+ sysModule = "com.googlesource.gerrit.plugins.replication.pull.PullReplicationModule",
+ httpModule = "com.googlesource.gerrit.plugins.replication.pull.api.HttpModule")
+public class PullReplicationAsyncIT extends PullReplicationIT {
+ @Inject private SitePaths sitePaths;
+
+ @Override
+ public void setUpTestPlugin() throws Exception {
+ FileBasedConfig config =
+ new FileBasedConfig(sitePaths.etc_dir.resolve("replication.config").toFile(), FS.DETECTED);
+ config.setString("replication", null, "syncRefs", "^$");
+ config.save();
+
+ super.setUpTestPlugin(true);
+ }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
index c6d32bf..7fcff10 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
@@ -50,6 +50,7 @@
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.googlesource.gerrit.plugins.replication.AutoReloadConfigDecorator;
+import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
import java.time.Duration;
@@ -78,11 +79,12 @@
@UseLocalDisk
@TestPlugin(
name = "pull-replication",
- sysModule = "com.googlesource.gerrit.plugins.replication.pull.PullReplicationModule")
+ sysModule = "com.googlesource.gerrit.plugins.replication.pull.PullReplicationModule",
+ httpModule = "com.googlesource.gerrit.plugins.replication.pull.api.HttpModule")
public class PullReplicationIT extends LightweightPluginDaemonTest {
private static final Optional<String> ALL_PROJECTS = Optional.empty();
private static final FluentLogger logger = FluentLogger.forEnclosingClass();
- private static final int TEST_REPLICATION_DELAY = 60;
+ private static final int TEST_REPLICATION_DELAY = 1;
private static final Duration TEST_TIMEOUT = Duration.ofSeconds(TEST_REPLICATION_DELAY * 2000);
private static final String TEST_REPLICATION_SUFFIX = "suffix1";
private static final String TEST_REPLICATION_REMOTE = "remote1";
@@ -96,10 +98,17 @@
@Override
public void setUpTestPlugin() throws Exception {
+ setUpTestPlugin(false);
+ }
+
+ protected void setUpTestPlugin(boolean loadExisting) throws Exception {
gitPath = sitePaths.site_path.resolve("git");
- config =
- new FileBasedConfig(sitePaths.etc_dir.resolve("replication.config").toFile(), FS.DETECTED);
+ File configFile = sitePaths.etc_dir.resolve("replication.config").toFile();
+ config = new FileBasedConfig(configFile, FS.DETECTED);
+ if (loadExisting && configFile.exists()) {
+ config.load();
+ }
setReplicationSource(
TEST_REPLICATION_REMOTE,
TEST_REPLICATION_SUFFIX,
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java
index 08d7e5a..56b6ad1 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java
@@ -19,6 +19,7 @@
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -81,6 +82,7 @@
@Mock RevisionData revisionData;
@Mock HttpResult httpResult;
ApplyObjectMetrics applyObjectMetrics;
+ FetchReplicationMetrics fetchMetrics;
@Captor ArgumentCaptor<String> stringCaptor;
@Captor ArgumentCaptor<Project.NameKey> projectNameKeyCaptor;
@@ -108,15 +110,24 @@
when(fetchClientFactory.create(any())).thenReturn(fetchRestApiClient);
when(fetchRestApiClient.callSendObject(any(), anyString(), anyBoolean(), any(), any()))
.thenReturn(httpResult);
- when(fetchRestApiClient.callFetch(any(), anyString(), any())).thenReturn(httpResult);
+ when(fetchRestApiClient.callFetch(any(), anyString(), any(), anyLong())).thenReturn(httpResult);
when(httpResult.isSuccessful()).thenReturn(true);
when(httpResult.isProjectMissing(any())).thenReturn(false);
applyObjectMetrics = new ApplyObjectMetrics("pull-replication", new DisabledMetricMaker());
+ fetchMetrics = new FetchReplicationMetrics("pull-replication", new DisabledMetricMaker());
objectUnderTest =
new ReplicationQueue(
- wq, rd, dis, sl, fetchClientFactory, refsFilter, revReader, applyObjectMetrics);
+ wq,
+ rd,
+ dis,
+ sl,
+ fetchClientFactory,
+ refsFilter,
+ revReader,
+ applyObjectMetrics,
+ fetchMetrics);
}
@Test
@@ -173,7 +184,7 @@
objectUnderTest.onGitReferenceUpdated(event);
- verify(fetchRestApiClient).callFetch(any(), anyString(), any());
+ verify(fetchRestApiClient).callFetch(any(), anyString(), any(), anyLong());
}
@Test
@@ -186,7 +197,7 @@
objectUnderTest.onGitReferenceUpdated(event);
- verify(fetchRestApiClient).callFetch(any(), anyString(), any());
+ verify(fetchRestApiClient).callFetch(any(), anyString(), any(), anyLong());
}
@Test
@@ -202,7 +213,7 @@
objectUnderTest.onGitReferenceUpdated(event);
- verify(fetchRestApiClient).callFetch(any(), anyString(), any());
+ verify(fetchRestApiClient).callFetch(any(), anyString(), any(), anyLong());
}
@Test
@@ -248,7 +259,15 @@
objectUnderTest =
new ReplicationQueue(
- wq, rd, dis, sl, fetchClientFactory, refsFilter, revReader, applyObjectMetrics);
+ wq,
+ rd,
+ dis,
+ sl,
+ fetchClientFactory,
+ refsFilter,
+ revReader,
+ applyObjectMetrics,
+ fetchMetrics);
Event event = new TestEvent("refs/multi-site/version");
objectUnderTest.onGitReferenceUpdated(event);
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 e1ad565..9af2d10 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
@@ -35,6 +35,7 @@
import com.googlesource.gerrit.plugins.replication.pull.SourcesCollection;
import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException;
import java.net.URISyntaxException;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -56,6 +57,7 @@
@Mock Source source;
@Mock SourcesCollection sources;
@Mock DynamicItem<EventDispatcher> eventDispatcher;
+ @Mock PullReplicationApiRequestMetrics apiRequestMetrics;
@SuppressWarnings("rawtypes")
@Mock
@@ -76,7 +78,7 @@
when(fetchReplicationStateFactory.create(any())).thenReturn(state);
when(source.getRemoteConfigName()).thenReturn(label);
when(sources.getAll()).thenReturn(Lists.newArrayList(source));
- when(source.schedule(eq(projectName), eq(REF_NAME_TO_FETCH), eq(state), any()))
+ when(source.schedule(eq(projectName), eq(REF_NAME_TO_FETCH), eq(state), any(), any()))
.thenReturn(CompletableFuture.completedFuture(null));
objectUnderTest =
new FetchCommand(fetchReplicationStateFactory, fetchStateLog, sources, eventDispatcher);
@@ -88,16 +90,18 @@
TimeoutException {
objectUnderTest.fetchSync(projectName, label, REF_NAME_TO_FETCH);
- verify(source, times(1)).schedule(projectName, REF_NAME_TO_FETCH, state, SYNC);
+ verify(source, times(1))
+ .schedule(projectName, REF_NAME_TO_FETCH, state, SYNC, Optional.empty());
}
@Test
public void shouldScheduleRefFetchWithDelay()
throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
TimeoutException {
- objectUnderTest.fetchAsync(projectName, label, REF_NAME_TO_FETCH);
+ objectUnderTest.fetchAsync(projectName, label, REF_NAME_TO_FETCH, apiRequestMetrics);
- verify(source, times(1)).schedule(projectName, REF_NAME_TO_FETCH, state, ASYNC);
+ verify(source, times(1))
+ .schedule(projectName, REF_NAME_TO_FETCH, state, ASYNC, Optional.of(apiRequestMetrics));
}
@Test
@@ -106,7 +110,8 @@
TimeoutException {
objectUnderTest.fetchSync(projectName, label, REF_NAME_TO_FETCH);
- verify(source, times(1)).schedule(projectName, REF_NAME_TO_FETCH, state, SYNC);
+ verify(source, times(1))
+ .schedule(projectName, REF_NAME_TO_FETCH, state, SYNC, Optional.empty());
verify(state, times(1)).markAllFetchTasksScheduled();
}
@@ -123,7 +128,8 @@
public void shouldUpdateStateWhenInterruptedException()
throws InterruptedException, ExecutionException, TimeoutException {
when(future.get(anyLong(), eq(TimeUnit.SECONDS))).thenThrow(new InterruptedException());
- when(source.schedule(projectName, REF_NAME_TO_FETCH, state, SYNC)).thenReturn(future);
+ when(source.schedule(projectName, REF_NAME_TO_FETCH, state, SYNC, Optional.empty()))
+ .thenReturn(future);
InterruptedException e =
assertThrows(
@@ -138,7 +144,8 @@
throws InterruptedException, ExecutionException, TimeoutException {
when(future.get(anyLong(), eq(TimeUnit.SECONDS)))
.thenThrow(new ExecutionException(new Exception()));
- when(source.schedule(projectName, REF_NAME_TO_FETCH, state, SYNC)).thenReturn(future);
+ when(source.schedule(projectName, REF_NAME_TO_FETCH, state, SYNC, Optional.empty()))
+ .thenReturn(future);
ExecutionException e =
assertThrows(
@@ -152,7 +159,8 @@
public void shouldUpdateStateWhenTimeoutException()
throws InterruptedException, ExecutionException, TimeoutException {
when(future.get(anyLong(), eq(TimeUnit.SECONDS))).thenThrow(new TimeoutException());
- when(source.schedule(projectName, REF_NAME_TO_FETCH, state, SYNC)).thenReturn(future);
+ when(source.schedule(projectName, REF_NAME_TO_FETCH, state, SYNC, Optional.empty()))
+ .thenReturn(future);
TimeoutException e =
assertThrows(
