mirror of
https://github.com/google/nomulus
synced 2026-09-08 17:17:02 +00:00
Create a Dataflow pipeline to resave EPP resources (#1553)
* Create a Dataflow pipeline to resave EPP resources This has two modes. If `fast` is false, then we will just load all EPP resources, project them to the current time, and save them. If `fast` is true, we will attempt to intelligently load and save only resources that we expect to have changes applied when we project them to the current time. This means resources with pending transfers that have expired, domains with expired grace periods, and non-deleted domains that have expired (we expect that they autorenewed).
This commit is contained in:
gbrodman
@@ -20,6 +20,7 @@ import static google.registry.testing.DatabaseHelper.persistActiveContact;
|
||||
import static google.registry.testing.DatabaseHelper.persistContactWithPendingTransfer;
|
||||
import static org.joda.time.DateTimeZone.UTC;
|
||||
|
||||
import google.registry.model.annotations.DeleteAfterMigration;
|
||||
import google.registry.model.contact.ContactResource;
|
||||
import google.registry.model.transfer.TransferStatus;
|
||||
import google.registry.testing.FakeResponse;
|
||||
@@ -29,6 +30,9 @@ import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
/** Unit tests for {@link ResaveAllEppResourcesAction}. */
|
||||
// No longer needed in SQL. Subject to future removal.
|
||||
@Deprecated
|
||||
@DeleteAfterMigration
|
||||
class ResaveAllEppResourcesActionTest extends MapreduceTestCase<ResaveAllEppResourcesAction> {
|
||||
|
||||
@BeforeEach
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
// Copyright 2022 The Nomulus Authors. All Rights Reserved.
|
||||
//
|
||||
// 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 google.registry.batch;
|
||||
|
||||
import static com.google.common.truth.Truth.assertThat;
|
||||
import static google.registry.batch.ResaveAllEppResourcesPipelineAction.PARAM_FAST;
|
||||
import static google.registry.batch.ResaveAllEppResourcesPipelineAction.PIPELINE_NAME;
|
||||
import static google.registry.beam.BeamUtils.createJobName;
|
||||
import static javax.servlet.http.HttpServletResponse.SC_OK;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
import com.google.api.services.dataflow.model.LaunchFlexTemplateParameter;
|
||||
import com.google.api.services.dataflow.model.LaunchFlexTemplateRequest;
|
||||
import com.google.common.collect.ImmutableMap;
|
||||
import google.registry.beam.BeamActionTestBase;
|
||||
import google.registry.config.RegistryEnvironment;
|
||||
import google.registry.testing.AppEngineExtension;
|
||||
import google.registry.testing.FakeClock;
|
||||
import java.util.Optional;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.RegisterExtension;
|
||||
|
||||
/** Unit tests for {@link ResaveAllEppResourcesPipelineAction}. */
|
||||
public class ResaveAllEppResourcesPipelineActionTest extends BeamActionTestBase {
|
||||
|
||||
@RegisterExtension
|
||||
final AppEngineExtension appEngine =
|
||||
AppEngineExtension.builder().withDatastoreAndCloudSql().build();
|
||||
|
||||
private final FakeClock fakeClock = new FakeClock();
|
||||
|
||||
private ResaveAllEppResourcesPipelineAction createAction(boolean isFast) {
|
||||
return new ResaveAllEppResourcesPipelineAction(
|
||||
"test-project",
|
||||
"test-region",
|
||||
"staging-bucket",
|
||||
Optional.of(isFast),
|
||||
fakeClock,
|
||||
response,
|
||||
dataflow);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testLaunch_notFast() throws Exception {
|
||||
createAction(false).run();
|
||||
assertThat(response.getStatus()).isEqualTo(SC_OK);
|
||||
assertThat(response.getPayload()).isEqualTo("Launched resaveAllEppResources pipeline: jobid");
|
||||
verify(templates).launch("test-project", "test-region", createLaunchTemplateRequest(false));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testLaunch_fast() throws Exception {
|
||||
createAction(true).run();
|
||||
assertThat(response.getStatus()).isEqualTo(SC_OK);
|
||||
assertThat(response.getPayload()).isEqualTo("Launched resaveAllEppResources pipeline: jobid");
|
||||
verify(templates).launch("test-project", "test-region", createLaunchTemplateRequest(true));
|
||||
}
|
||||
|
||||
private LaunchFlexTemplateRequest createLaunchTemplateRequest(boolean isFast) {
|
||||
return new LaunchFlexTemplateRequest()
|
||||
.setLaunchParameter(
|
||||
new LaunchFlexTemplateParameter()
|
||||
.setJobName(createJobName("resave-all-epp-resources", fakeClock))
|
||||
.setContainerSpecGcsPath(
|
||||
String.format("%s/%s_metadata.json", "staging-bucket", PIPELINE_NAME))
|
||||
.setParameters(
|
||||
new ImmutableMap.Builder<String, String>()
|
||||
.put(PARAM_FAST, Boolean.toString(isFast))
|
||||
.put("registryEnvironment", RegistryEnvironment.get().name())
|
||||
.build()));
|
||||
}
|
||||
}
|
||||
+204
@@ -0,0 +1,204 @@
|
||||
// Copyright 2022 The Nomulus Authors. All Rights Reserved.
|
||||
//
|
||||
// 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 google.registry.beam.resave;
|
||||
|
||||
import static com.google.common.collect.ImmutableSet.toImmutableSet;
|
||||
import static com.google.common.truth.Truth.assertThat;
|
||||
import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
|
||||
import static google.registry.testing.DatabaseHelper.createTld;
|
||||
import static google.registry.testing.DatabaseHelper.loadAllOf;
|
||||
import static google.registry.testing.DatabaseHelper.loadByEntity;
|
||||
import static google.registry.testing.DatabaseHelper.persistActiveContact;
|
||||
import static google.registry.testing.DatabaseHelper.persistActiveDomain;
|
||||
import static google.registry.testing.DatabaseHelper.persistContactWithPendingTransfer;
|
||||
import static google.registry.testing.DatabaseHelper.persistDomainWithDependentResources;
|
||||
import static google.registry.testing.DatabaseHelper.persistDomainWithPendingTransfer;
|
||||
import static google.registry.testing.DatabaseHelper.persistNewRegistrars;
|
||||
import static org.mockito.Mockito.spy;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
import google.registry.beam.TestPipelineExtension;
|
||||
import google.registry.model.EppResource;
|
||||
import google.registry.model.contact.ContactResource;
|
||||
import google.registry.model.domain.DomainBase;
|
||||
import google.registry.model.domain.GracePeriod;
|
||||
import google.registry.model.eppcommon.StatusValue;
|
||||
import google.registry.persistence.transaction.JpaTestExtensions;
|
||||
import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationTestExtension;
|
||||
import google.registry.persistence.transaction.JpaTransactionManager;
|
||||
import google.registry.persistence.transaction.TransactionManagerFactory;
|
||||
import google.registry.testing.DatastoreEntityExtension;
|
||||
import google.registry.testing.FakeClock;
|
||||
import google.registry.testing.TmOverrideExtension;
|
||||
import org.apache.beam.sdk.options.PipelineOptionsFactory;
|
||||
import org.joda.time.DateTime;
|
||||
import org.joda.time.Duration;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Order;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.RegisterExtension;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
|
||||
/** Tests for {@link ResaveAllEppResourcesPipeline}. */
|
||||
public class ResaveAllEppResourcesPipelineTest {
|
||||
|
||||
private final FakeClock fakeClock = new FakeClock(DateTime.parse("2020-03-10T00:00:00.000Z"));
|
||||
|
||||
@RegisterExtension
|
||||
@Order(Order.DEFAULT - 1)
|
||||
final transient DatastoreEntityExtension datastore =
|
||||
new DatastoreEntityExtension().allThreads(true);
|
||||
|
||||
@RegisterExtension
|
||||
final TestPipelineExtension testPipeline =
|
||||
TestPipelineExtension.create().enableAbandonedNodeEnforcement(true);
|
||||
|
||||
@RegisterExtension
|
||||
final JpaIntegrationTestExtension database =
|
||||
new JpaTestExtensions.Builder().withClock(fakeClock).buildIntegrationTestExtension();
|
||||
|
||||
@RegisterExtension
|
||||
@Order(Order.DEFAULT + 1)
|
||||
TmOverrideExtension tmOverrideExtension = TmOverrideExtension.withJpa();
|
||||
|
||||
private final ResaveAllEppResourcesPipelineOptions options =
|
||||
PipelineOptionsFactory.create().as(ResaveAllEppResourcesPipelineOptions.class);
|
||||
|
||||
@BeforeEach
|
||||
void beforeEach() {
|
||||
options.setFast(true);
|
||||
persistNewRegistrars("TheRegistrar", "NewRegistrar");
|
||||
createTld("tld");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPipeline_unchangedEntity() {
|
||||
ContactResource contact = persistActiveContact("test123");
|
||||
DateTime creationTime = contact.getUpdateTimestamp().getTimestamp();
|
||||
fakeClock.advanceOneMilli();
|
||||
assertThat(loadByEntity(contact).getUpdateTimestamp().getTimestamp()).isEqualTo(creationTime);
|
||||
fakeClock.advanceOneMilli();
|
||||
runPipeline();
|
||||
assertThat(loadByEntity(contact)).isEqualTo(contact);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPipeline_fulfilledContactTransfer() {
|
||||
ContactResource contact = persistActiveContact("test123");
|
||||
DateTime now = fakeClock.nowUtc();
|
||||
contact = persistContactWithPendingTransfer(contact, now, now.plusDays(5), now);
|
||||
fakeClock.advanceBy(Duration.standardDays(10));
|
||||
assertThat(loadByEntity(contact).getStatusValues()).contains(StatusValue.PENDING_TRANSFER);
|
||||
runPipeline();
|
||||
assertThat(loadByEntity(contact).getStatusValues())
|
||||
.doesNotContain(StatusValue.PENDING_TRANSFER);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPipeline_fulfilledDomainTransfer() {
|
||||
options.setFast(true);
|
||||
DateTime now = fakeClock.nowUtc();
|
||||
DomainBase domain =
|
||||
persistDomainWithPendingTransfer(
|
||||
persistDomainWithDependentResources(
|
||||
"domain",
|
||||
"tld",
|
||||
persistActiveContact("jd1234"),
|
||||
now.minusDays(5),
|
||||
now.minusDays(5),
|
||||
now.plusYears(2)),
|
||||
now.minusDays(4),
|
||||
now.minusDays(1),
|
||||
now.plusYears(2));
|
||||
assertThat(domain.getStatusValues()).contains(StatusValue.PENDING_TRANSFER);
|
||||
assertThat(domain.getUpdateTimestamp().getTimestamp()).isEqualTo(now);
|
||||
fakeClock.advanceOneMilli();
|
||||
runPipeline();
|
||||
DomainBase postPipeline = loadByEntity(domain);
|
||||
assertThat(postPipeline.getStatusValues()).doesNotContain(StatusValue.PENDING_TRANSFER);
|
||||
assertThat(postPipeline.getUpdateTimestamp().getTimestamp()).isEqualTo(fakeClock.nowUtc());
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPipeline_autorenewedDomain() {
|
||||
DateTime now = fakeClock.nowUtc();
|
||||
DomainBase domain =
|
||||
persistDomainWithDependentResources(
|
||||
"domain", "tld", persistActiveContact("jd1234"), now, now, now.plusYears(1));
|
||||
assertThat(domain.getRegistrationExpirationTime()).isEqualTo(now.plusYears(1));
|
||||
fakeClock.advanceBy(Duration.standardDays(500));
|
||||
runPipeline();
|
||||
DomainBase postPipeline = loadByEntity(domain);
|
||||
assertThat(postPipeline.getRegistrationExpirationTime()).isEqualTo(now.plusYears(2));
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPipeline_expiredGracePeriod() {
|
||||
DateTime now = fakeClock.nowUtc();
|
||||
persistDomainWithDependentResources(
|
||||
"domain", "tld", persistActiveContact("jd1234"), now, now, now.plusYears(1));
|
||||
assertThat(loadAllOf(GracePeriod.class)).hasSize(1);
|
||||
fakeClock.advanceBy(Duration.standardDays(500));
|
||||
runPipeline();
|
||||
assertThat(loadAllOf(GracePeriod.class)).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPipeline_fastOnlySavesChanged() {
|
||||
DateTime now = fakeClock.nowUtc();
|
||||
ContactResource contact = persistActiveContact("jd1234");
|
||||
persistDomainWithDependentResources("renewed", "tld", contact, now, now, now.plusYears(1));
|
||||
persistActiveDomain("nonrenewed.tld", now, now.plusYears(20));
|
||||
// Spy the transaction manager so we can be sure we're only saving the renewed domain
|
||||
JpaTransactionManager spy = spy(jpaTm());
|
||||
TransactionManagerFactory.setJpaTm(() -> spy);
|
||||
ArgumentCaptor<DomainBase> domainPutCaptor = ArgumentCaptor.forClass(DomainBase.class);
|
||||
runPipeline();
|
||||
// We should only be attempting to put the one changed domain into the DB
|
||||
verify(spy).put(domainPutCaptor.capture());
|
||||
assertThat(domainPutCaptor.getValue().getDomainName()).isEqualTo("renewed.tld");
|
||||
}
|
||||
|
||||
@Test
|
||||
void testPipeline_notFastResavesAll() {
|
||||
options.setFast(false);
|
||||
DateTime now = fakeClock.nowUtc();
|
||||
ContactResource contact = persistActiveContact("jd1234");
|
||||
DomainBase renewed =
|
||||
persistDomainWithDependentResources("renewed", "tld", contact, now, now, now.plusYears(1));
|
||||
DomainBase nonRenewed =
|
||||
persistDomainWithDependentResources(
|
||||
"nonrenewed", "tld", contact, now, now, now.plusYears(20));
|
||||
// Spy the transaction manager so we can be sure we're attempting to save everything
|
||||
JpaTransactionManager spy = spy(jpaTm());
|
||||
TransactionManagerFactory.setJpaTm(() -> spy);
|
||||
ArgumentCaptor<EppResource> eppResourcePutCaptor = ArgumentCaptor.forClass(EppResource.class);
|
||||
runPipeline();
|
||||
// We should be attempting to put both domains (and the contact) in, even the unchanged ones
|
||||
verify(spy, times(3)).put(eppResourcePutCaptor.capture());
|
||||
assertThat(
|
||||
eppResourcePutCaptor.getAllValues().stream()
|
||||
.map(EppResource::getRepoId)
|
||||
.collect(toImmutableSet()))
|
||||
.containsExactly(contact.getRepoId(), renewed.getRepoId(), nonRenewed.getRepoId());
|
||||
}
|
||||
|
||||
private void runPipeline() {
|
||||
ResaveAllEppResourcesPipeline pipeline = new ResaveAllEppResourcesPipeline(options);
|
||||
pipeline.setupPipeline(testPipeline);
|
||||
testPipeline.run().waitUntilFinish();
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,7 @@ PATH CLASS
|
||||
/_dr/task/relockDomain RelockDomainAction POST y INTERNAL,API APP ADMIN
|
||||
/_dr/task/replayCommitLogsToSql ReplayCommitLogsToSqlAction POST y INTERNAL,API APP ADMIN
|
||||
/_dr/task/resaveAllEppResources ResaveAllEppResourcesAction GET n INTERNAL,API APP ADMIN
|
||||
/_dr/task/resaveAllEppResourcesPipeline ResaveAllEppResourcesPipelineAction GET n INTERNAL,API APP ADMIN
|
||||
/_dr/task/resaveEntity ResaveEntityAction POST n INTERNAL,API APP ADMIN
|
||||
/_dr/task/sendExpiringCertificateNotificationEmail SendExpiringCertificateNotificationEmailAction GET n INTERNAL,API APP ADMIN
|
||||
/_dr/task/syncDatastoreToSqlSnapshot SyncDatastoreToSqlSnapshotAction POST n INTERNAL,API APP ADMIN
|
||||
|
||||
Reference in New Issue
Block a user