Use pull replication REST Api for replication Pull replication Rest Api client implementation allows replication of ref-update based on object id. Feature: Issue 11605 Change-Id: I5eb976cc81a891e2e31ba851dab406c041cbabb3
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 4c6edb0..43cd4ef 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
@@ -36,10 +36,12 @@ import java.io.IOException; import java.util.Collection; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; import org.eclipse.jgit.errors.NoRemoteRepositoryException; import org.eclipse.jgit.errors.NotSupportedException; import org.eclipse.jgit.errors.RemoteRepositoryException; @@ -50,6 +52,7 @@ import org.eclipse.jgit.lib.Repository; import org.eclipse.jgit.transport.CredentialsProvider; import org.eclipse.jgit.transport.FetchResult; +import org.eclipse.jgit.transport.RefSpec; import org.eclipse.jgit.transport.RemoteConfig; import org.eclipse.jgit.transport.TrackingRefUpdate; import org.eclipse.jgit.transport.Transport; @@ -373,9 +376,17 @@ tn.applyConfig(config); tn.setCredentialsProvider(credentialsProvider); - repLog.info("Fetch references {} from {}", config.getFetchRefSpecs(), uri); + List<RefSpec> fetchRefSpecs = getFetchRefSpecs(); - return tn.fetch(NullProgressMonitor.INSTANCE, config.getFetchRefSpecs()); + repLog.info("Fetch references {} from {}", fetchRefSpecs, uri); + return tn.fetch(NullProgressMonitor.INSTANCE, fetchRefSpecs); + } + + private List<RefSpec> getFetchRefSpecs() { + if (delta.isEmpty()) { + return config.getFetchRefSpecs(); + } + return delta.stream().map(ref -> new RefSpec(ref + ":" + ref)).collect(Collectors.toList()); } private void updateStates(Collection<TrackingRefUpdate> refUpdates) throws IOException {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/NoopObservableQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/NoopObservableQueue.java deleted file mode 100644 index 0e752c7..0000000 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/NoopObservableQueue.java +++ /dev/null
@@ -1,30 +0,0 @@ -// Copyright (C) 2019 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.googlesource.gerrit.plugins.replication.ObservableQueue; - -public class NoopObservableQueue implements ObservableQueue { - - @Override - public boolean isRunning() { - return true; - } - - @Override - public boolean isReplaying() { - return false; - } -}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java index 6eb1d30..1b079dd 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
@@ -19,6 +19,7 @@ import com.google.common.eventbus.EventBus; import com.google.gerrit.extensions.annotations.Exports; import com.google.gerrit.extensions.config.CapabilityDefinition; +import com.google.gerrit.extensions.events.GitReferenceUpdatedListener; import com.google.gerrit.extensions.events.LifecycleListener; import com.google.gerrit.extensions.registration.DynamicSet; import com.google.gerrit.server.config.SitePaths; @@ -38,9 +39,12 @@ import com.googlesource.gerrit.plugins.replication.ReplicationFileBasedConfig; import com.googlesource.gerrit.plugins.replication.StartReplicationCapability; import com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationApiModule; +import com.googlesource.gerrit.plugins.replication.pull.client.FetchRestApiClient; +import com.googlesource.gerrit.plugins.replication.pull.client.HttpClientProvider; import java.io.File; import java.io.IOException; import java.nio.file.Path; +import org.apache.http.impl.client.CloseableHttpClient; import org.eclipse.jgit.errors.ConfigInvalidException; import org.eclipse.jgit.storage.file.FileBasedConfig; import org.eclipse.jgit.util.FS; @@ -57,7 +61,11 @@ protected void configure() { install(new PullReplicationApiModule()); + + bind(CloseableHttpClient.class).toProvider(HttpClientProvider.class).in(Scopes.SINGLETON); install(new FactoryModuleBuilder().build(Source.Factory.class)); + install(new FactoryModuleBuilder().build(FetchRestApiClient.Factory.class)); + bind(FetchReplicationMetrics.class).in(Scopes.SINGLETON); bind(OnStartStop.class).in(Scopes.SINGLETON); @@ -77,7 +85,14 @@ bind(EventBus.class).in(Scopes.SINGLETON); bind(ReplicationSources.class).to(SourcesCollection.class); - bind(ObservableQueue.class).to(NoopObservableQueue.class).in(Scopes.SINGLETON); + + bind(ReplicationQueue.class).in(Scopes.SINGLETON); + bind(ObservableQueue.class).to(ReplicationQueue.class); + bind(LifecycleListener.class) + .annotatedWith(UniqueAnnotations.create()) + .to(ReplicationQueue.class); + + DynamicSet.bind(binder(), GitReferenceUpdatedListener.class).to(ReplicationQueue.class); bind(ReplicationConfigValidator.class).to(SourcesCollection.class);
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 new file mode 100644 index 0000000..2158a59 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java
@@ -0,0 +1,180 @@ +// Copyright (C) 2020 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.auto.value.AutoValue; +import com.google.common.collect.Queues; +import com.google.gerrit.entities.Project; +import com.google.gerrit.extensions.events.GitReferenceUpdatedListener; +import com.google.gerrit.extensions.events.LifecycleListener; +import com.google.gerrit.extensions.registration.DynamicItem; +import com.google.gerrit.server.events.EventDispatcher; +import com.google.gerrit.server.git.WorkQueue; +import com.google.inject.Inject; +import com.google.inject.Provider; +import com.googlesource.gerrit.plugins.replication.ObservableQueue; +import com.googlesource.gerrit.plugins.replication.pull.FetchResultProcessing.GitUpdateProcessing; +import com.googlesource.gerrit.plugins.replication.pull.client.FetchRestApiClient; +import com.googlesource.gerrit.plugins.replication.pull.client.HttpResult; +import java.net.URISyntaxException; +import java.util.HashSet; +import java.util.Queue; +import java.util.Set; +import org.eclipse.jgit.lib.ObjectId; +import org.eclipse.jgit.transport.URIish; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ReplicationQueue + implements ObservableQueue, LifecycleListener, GitReferenceUpdatedListener { + + static final String PULL_REPLICATION_LOG_NAME = "pull_replication_log"; + static final Logger repLog = LoggerFactory.getLogger(PULL_REPLICATION_LOG_NAME); + + private final ReplicationStateListener stateLog; + + private final WorkQueue workQueue; + private final DynamicItem<EventDispatcher> dispatcher; + private final Provider<SourcesCollection> sources; // For Guice circular dependency + private volatile boolean running; + private volatile boolean replaying; + private final Queue<ReferenceUpdatedEvent> beforeStartupEventsQueue; + private FetchRestApiClient.Factory fetchClientFactory; + + @Inject + ReplicationQueue( + WorkQueue wq, + Provider<SourcesCollection> rd, + DynamicItem<EventDispatcher> dis, + ReplicationStateListeners sl, + FetchRestApiClient.Factory fetchClientFactory) { + workQueue = wq; + dispatcher = dis; + sources = rd; + stateLog = sl; + beforeStartupEventsQueue = Queues.newConcurrentLinkedQueue(); + this.fetchClientFactory = fetchClientFactory; + } + + @Override + public void start() { + if (!running) { + sources.get().startup(workQueue); + running = true; + fireBeforeStartupEvents(); + } + } + + @Override + public void stop() { + running = false; + int discarded = sources.get().shutdown(); + if (discarded > 0) { + repLog.warn("Canceled {} replication events during shutdown", discarded); + } + } + + @Override + public boolean isRunning() { + return running; + } + + @Override + public boolean isReplaying() { + return replaying; + } + + @Override + public void onGitReferenceUpdated(GitReferenceUpdatedListener.Event event) { + fire(event.getProjectName(), ObjectId.fromString(event.getNewObjectId()), event.getRefName()); + } + + private void fire(String projectName, ObjectId objectId, String refName) { + ReplicationState state = new ReplicationState(new GitUpdateProcessing(dispatcher.get())); + fire(Project.nameKey(projectName), objectId, refName, state); + state.markAllFetchTasksScheduled(); + } + + private void fire( + Project.NameKey project, ObjectId objectId, String refName, ReplicationState state) { + if (!running) { + stateLog.warn( + "Replication plugin did not finish startup before event, event replication is postponed", + state); + beforeStartupEventsQueue.add(ReferenceUpdatedEvent.create(project.get(), refName, objectId)); + return; + } + + for (Source cfg : sources.get().getAll()) { + if (cfg.wouldFetchProject(project) && cfg.wouldFetchRef(refName)) { + for (String apiUrl : cfg.getApis()) { + try { + URIish uri = new URIish(apiUrl); + FetchRestApiClient fetchClient = fetchClientFactory.create(cfg); + + HttpResult result = fetchClient.callFetch(project, refName, uri); + + if (!result.isSuccessful()) { + stateLog.warn( + String.format( + "Pull replication rest api fetch call failed. Endpoint url: %s, reason:%s", + apiUrl, result.getMessage().orElse("unknown")), + state); + } + } catch (URISyntaxException e) { + stateLog.warn(String.format("Cannot parse pull replication api url:%s", apiUrl), state); + } catch (Exception e) { + stateLog.error( + String.format( + "Exception during the pull replication fetch rest api call. Endpoint url:%s, message:%s", + apiUrl, e.getMessage()), + e, + state); + } + } + } + } + } + + public boolean retry(int attempt, int maxRetries) { + return maxRetries == 0 || attempt < maxRetries; + } + + private void fireBeforeStartupEvents() { + Set<String> eventsReplayed = new HashSet<>(); + for (ReferenceUpdatedEvent event : beforeStartupEventsQueue) { + String eventKey = String.format("%s:%s", event.projectName(), event.refName()); + if (!eventsReplayed.contains(eventKey)) { + repLog.info("Firing pending task {}", event); + fire(event.projectName(), event.objectId(), event.refName()); + eventsReplayed.add(eventKey); + } + } + } + + @AutoValue + abstract static class ReferenceUpdatedEvent { + + static ReferenceUpdatedEvent create(String projectName, String refName, ObjectId objectId) { + return new AutoValue_ReplicationQueue_ReferenceUpdatedEvent(projectName, refName, objectId); + } + + public abstract String projectName(); + + public abstract String refName(); + + public abstract ObjectId objectId(); + } +}
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 a2c70df..e8eae78 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
@@ -77,6 +77,7 @@ import org.eclipse.jgit.lib.Ref; import org.eclipse.jgit.lib.RefUpdate; import org.eclipse.jgit.lib.Repository; +import org.eclipse.jgit.transport.RefSpec; import org.eclipse.jgit.transport.RemoteConfig; import org.eclipse.jgit.transport.URIish; import org.slf4j.Logger; @@ -509,7 +510,22 @@ } } - boolean wouldFetchProject(Project.NameKey project) { + public boolean wouldFetchRef(String ref) { + if (!config.replicatePermissions() && RefNames.REFS_CONFIG.equals(ref)) { + return false; + } + if (FetchOne.ALL_REFS.equals(ref)) { + return true; + } + for (RefSpec s : config.getRemoteConfig().getFetchRefSpecs()) { + if (s.matchSource(ref)) { + return true; + } + } + return false; + } + + public boolean wouldFetchProject(Project.NameKey project) { if (!shouldReplicate(project)) { return false; } @@ -626,6 +642,10 @@ return config.getRemoteConfig().getTimeout(); } + public ImmutableList<String> getApis() { + return config.getApis(); + } + public int getMaxRetries() { return config.getMaxRetries(); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java index 8ef68b0..009d952 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java
@@ -40,6 +40,7 @@ private final ImmutableList<String> projects; private final ImmutableList<String> authGroupNames; private final RemoteConfig remoteConfig; + private final ImmutableList<String> apis; private final int maxRetries; private int slowLatencyThreshold; @@ -47,6 +48,7 @@ this.remoteConfig = remoteConfig; String name = remoteConfig.getName(); urls = ImmutableList.copyOf(cfg.getStringList("remote", name, "url")); + apis = ImmutableList.copyOf(cfg.getStringList("remote", name, "apiUrl")); delay = Math.max(0, getInt(remoteConfig, cfg, "replicationdelay", DEFAULT_REPLICATION_DELAY)); rescheduleDelay = Math.max(3, getInt(remoteConfig, cfg, "rescheduledelay", DEFAULT_RESCHEDULE_DELAY)); @@ -103,6 +105,10 @@ return urls; } + public ImmutableList<String> getApis() { + return apis; + } + @Override public ImmutableList<String> getAdminUrls() { return adminUrls;
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 new file mode 100644 index 0000000..4296f67 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
@@ -0,0 +1,117 @@ +// Copyright (C) 2020 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.client; + +import com.google.common.flogger.FluentLogger; +import com.google.gerrit.entities.Project; +import com.google.gerrit.extensions.restapi.Url; +import com.google.inject.Inject; +import com.google.inject.assistedinject.Assisted; +import com.googlesource.gerrit.plugins.replication.CredentialsFactory; +import com.googlesource.gerrit.plugins.replication.ReplicationFileBasedConfig; +import com.googlesource.gerrit.plugins.replication.pull.Source; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Optional; +import org.apache.http.HttpResponse; +import org.apache.http.ParseException; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.ClientProtocolException; +import org.apache.http.client.CredentialsProvider; +import org.apache.http.client.ResponseHandler; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.protocol.HttpClientContext; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.BasicCredentialsProvider; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.message.BasicHeader; +import org.apache.http.util.EntityUtils; +import org.eclipse.jgit.transport.CredentialItem; +import org.eclipse.jgit.transport.URIish; + +public class FetchRestApiClient implements ResponseHandler<HttpResult> { + private static final FluentLogger logger = FluentLogger.forEnclosingClass(); + static String GERRIT_ADMIN_PROTOCOL_PREFIX = "gerrit+"; + + public interface Factory { + FetchRestApiClient create(Source source); + } + + private final CredentialsFactory credentials; + private final CloseableHttpClient httpClient; + private final Source source; + private final String instanceLabel; + + @Inject + FetchRestApiClient( + CredentialsFactory credentials, + CloseableHttpClient httpClient, + ReplicationFileBasedConfig replicationConfig, + @Assisted Source source) { + this.credentials = credentials; + this.httpClient = httpClient; + this.source = source; + this.instanceLabel = + replicationConfig.getConfig().getString("replication", null, "instanceLabel"); + } + + public HttpResult callFetch(Project.NameKey project, String refName, URIish targetUri) + throws ClientProtocolException, IOException { + String url = + String.format( + "%s/a/projects/%s/pull-replication~fetch", + targetUri.toString(), Url.encode(project.get())); + + HttpPost post = new HttpPost(url); + post.setEntity( + new StringEntity( + String.format("{\"label\":\"%s\", \"ref_name\": \"%s\"}", instanceLabel, refName), + StandardCharsets.UTF_8)); + post.addHeader(new BasicHeader("Content-Type", "application/json")); + return httpClient.execute(post, this, getContext(targetUri)); + } + + @Override + public HttpResult handleResponse(HttpResponse response) { + Optional<String> responseBody = Optional.empty(); + + try { + responseBody = Optional.ofNullable(EntityUtils.toString(response.getEntity())); + } catch (ParseException | IOException e) { + logger.atSevere().withCause(e).log("Unable get response body from %s", response.toString()); + } + return new HttpResult(response.getStatusLine().getStatusCode(), responseBody); + } + + private HttpClientContext getContext(URIish targetUri) { + HttpClientContext ctx = HttpClientContext.create(); + ctx.setCredentialsProvider(adapt(credentials.create(source.getRemoteConfigName()), targetUri)); + return ctx; + } + + private CredentialsProvider adapt(org.eclipse.jgit.transport.CredentialsProvider cp, URIish uri) { + CredentialItem.Username user = new CredentialItem.Username(); + CredentialItem.Password pass = new CredentialItem.Password(); + if (cp.supports(user, pass) && cp.get(uri, user, pass)) { + CredentialsProvider adapted = new BasicCredentialsProvider(); + adapted.setCredentials( + AuthScope.ANY, + new UsernamePasswordCredentials(user.getValue(), new String(pass.getValue()))); + return adapted; + } + return null; + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpClientProvider.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpClientProvider.java new file mode 100644 index 0000000..6d6e803 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpClientProvider.java
@@ -0,0 +1,62 @@ +// Copyright (C) 2020 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.client; + +import com.google.inject.Provider; +import com.google.inject.ProvisionException; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.conn.HttpClientConnectionManager; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; + +/** Provides an HTTP client with SSL capabilities. */ +public class HttpClientProvider implements Provider<CloseableHttpClient> { + private static final int CONNECTIONS_PER_ROUTE = 100; + + // Up to 2 target instances with the max number of connections per host: + private static final int MAX_CONNECTIONS = 2 * CONNECTIONS_PER_ROUTE; + + private static final int MAX_CONNECTION_INACTIVITY_MS = 10000; + private static final int DEFAULT_TIMEOUT_MS = 5000; + + @Override + public CloseableHttpClient get() { + try { + return HttpClients.custom() + .setConnectionManager(customConnectionManager()) + .setDefaultRequestConfig(customRequestConfig()) + .build(); + } catch (Exception e) { + throw new ProvisionException("Couldn't create CloseableHttpClient", e); + } + } + + private RequestConfig customRequestConfig() { + return RequestConfig.custom() + .setConnectTimeout(DEFAULT_TIMEOUT_MS) + .setSocketTimeout(DEFAULT_TIMEOUT_MS) + .setConnectionRequestTimeout(DEFAULT_TIMEOUT_MS) + .build(); + } + + private HttpClientConnectionManager customConnectionManager() throws Exception { + PoolingHttpClientConnectionManager connManager = new PoolingHttpClientConnectionManager(); + connManager.setDefaultMaxPerRoute(CONNECTIONS_PER_ROUTE); + connManager.setMaxTotal(MAX_CONNECTIONS); + connManager.setValidateAfterInactivity(MAX_CONNECTION_INACTIVITY_MS); + return connManager; + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpResult.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpResult.java new file mode 100644 index 0000000..8a2b66d --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpResult.java
@@ -0,0 +1,39 @@ +// Copyright (C) 2020 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.client; + +import static javax.servlet.http.HttpServletResponse.SC_CREATED; +import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT; +import static javax.servlet.http.HttpServletResponse.SC_OK; + +import java.util.Optional; + +public class HttpResult { + private final Optional<String> message; + private final int responseCode; + + HttpResult(int responseCode, Optional<String> message) { + this.message = message; + this.responseCode = responseCode; + } + + public Optional<String> getMessage() { + return message; + } + + public boolean isSuccessful() { + return responseCode == SC_CREATED || responseCode == SC_NO_CONTENT || responseCode == SC_OK; + } +}
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md index 365d4f6..c8c53df 100644 --- a/src/main/resources/Documentation/config.md +++ b/src/main/resources/Documentation/config.md
@@ -18,8 +18,9 @@ file, for example to replicate in parallel from four different hosts:</a> ``` - [remote "host-one"] - url = gerrit2@host-one.example.com:/some/path/${name}.git + [remote "host-two"] + url = gerrit2@host-two.example.com:/some/path/${name}.git + apiUrl = http://host-two.example.com:8080 [remote "pubmirror"] url = mirror1.us.some.org:/pub/git/${name}.git @@ -30,6 +31,8 @@ threads = 3 authGroup = Public Mirror Group authGroup = Second Public Mirror Group + [replication] + instanceLabel = host-one ``` Then reload the replication plugin to pick up the new configuration: @@ -111,6 +114,13 @@ By default, fetches are retried indefinitely. +replication.instanceLabel +: Remote configuration name of the current server. + This label is passed as a part of the payload to notify other + servers to fetch specified ref-update object id from the url + provided in the remote configuration section which name is equal + to instanceLabel. + remote.NAME.url : Address of the remote server to fetch from. Single URL can be specified within a single remote block. A remote node can request @@ -132,6 +142,12 @@ [1]: http://www.git-scm.com/docs/git-fetch#URLS [3]: #remote.NAME.projects +remote.NAME.apiUrl +: Address of the rest api endpoint of the remote server to fetch from. + Multiple URLs may be specified within a single remote block, listing + different destinations which share the same settings. Gerrit calls + all URLs in sequence. + remote.NAME.uploadpack : Path of the `git-upload-pack` executable on the remote system, if using the SSH transport.
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientTest.java new file mode 100644 index 0000000..22dd0e9 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientTest.java
@@ -0,0 +1,146 @@ +// Copyright (C) 2020 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.client; + +import static com.google.common.truth.Truth.assertThat; +import static javax.servlet.http.HttpServletResponse.SC_CREATED; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyString; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.gerrit.entities.Project; +import com.google.gerrit.entities.RefNames; +import com.googlesource.gerrit.plugins.replication.CredentialsFactory; +import com.googlesource.gerrit.plugins.replication.ReplicationFileBasedConfig; +import com.googlesource.gerrit.plugins.replication.pull.Source; +import java.io.IOException; +import java.net.URISyntaxException; +import java.nio.ByteBuffer; +import java.util.Optional; +import org.apache.http.Header; +import org.apache.http.client.ClientProtocolException; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.message.BasicHeader; +import org.eclipse.jgit.storage.file.FileBasedConfig; +import org.eclipse.jgit.transport.CredentialItem; +import org.eclipse.jgit.transport.CredentialsProvider; +import org.eclipse.jgit.transport.URIish; +import org.eclipse.jgit.util.IO; +import org.eclipse.jgit.util.RawParseUtils; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.junit.MockitoJUnitRunner; +import org.mockito.stubbing.Answer; + +@RunWith(MockitoJUnitRunner.class) +public class FetchRestApiClientTest { + @Mock CredentialsProvider credentialProvider; + @Mock CredentialsFactory credentials; + @Mock CloseableHttpClient httpClient; + @Mock FileBasedConfig config; + @Mock ReplicationFileBasedConfig replicationConfig; + @Mock Source source; + @Captor ArgumentCaptor<HttpPost> httpPostCaptor; + + String api = "http://gerrit-host"; + String label = "Replication"; + String refName = RefNames.REFS_HEADS + "master"; + + String expectedPayload = "{\"label\":\"Replication\", \"ref_name\": \"" + refName + "\"}"; + Header expectedHeader = new BasicHeader("Content-Type", "application/json"); + + FetchRestApiClient objectUnderTest; + + @Before + public void setup() throws ClientProtocolException, IOException { + when(credentialProvider.supports(any())) + .thenAnswer( + new Answer<Boolean>() { + + @Override + public Boolean answer(InvocationOnMock invocation) throws Throwable { + CredentialItem.Username user = (CredentialItem.Username) invocation.getArgument(0); + CredentialItem.Password password = + (CredentialItem.Password) invocation.getArgument(1); + user.setValue("admin"); + password.setValue("secret".toCharArray()); + return true; + } + }); + + when(credentialProvider.get(any(), any(CredentialItem.class))).thenReturn(true); + when(credentials.create(anyString())).thenReturn(credentialProvider); + when(replicationConfig.getConfig()).thenReturn(config); + when(source.getRemoteConfigName()).thenReturn("Replication"); + when(config.getString("replication", null, "instanceLabel")).thenReturn(label); + + HttpResult httpResult = new HttpResult(SC_CREATED, Optional.of("result message")); + when(httpClient.execute(any(HttpPost.class), any(), any())).thenReturn(httpResult); + + objectUnderTest = new FetchRestApiClient(credentials, httpClient, replicationConfig, source); + } + + @Test + public void shouldCallFetchEndpoint() + throws ClientProtocolException, IOException, URISyntaxException { + + objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api)); + + verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any()); + + HttpPost httpPost = httpPostCaptor.getValue(); + assertThat(httpPost.getURI().getHost()).isEqualTo("gerrit-host"); + assertThat(httpPost.getURI().getPath()) + .isEqualTo("/a/projects/test_repo/pull-replication~fetch"); + } + + @Test + public void shouldCallFetchEndpointWithPayload() + throws ClientProtocolException, IOException, URISyntaxException { + + objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api)); + + verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any()); + + HttpPost httpPost = httpPostCaptor.getValue(); + assertThat(readPayload(httpPost)).isEqualTo(expectedPayload); + } + + @Test + public void shouldSetContentTypeHeader() + throws ClientProtocolException, IOException, URISyntaxException { + + objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api)); + + verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any()); + + HttpPost httpPost = httpPostCaptor.getValue(); + assertThat(httpPost.getLastHeader("Content-Type").getValue()) + .isEqualTo(expectedHeader.getValue()); + } + + public String readPayload(HttpPost entity) throws UnsupportedOperationException, IOException { + ByteBuffer buf = IO.readWholeStream(entity.getEntity().getContent(), 1024); + return RawParseUtils.decode(buf.array(), buf.arrayOffset(), buf.limit()).trim(); + } +}