mirrors/nomulus
1
0
Fork 0
mirror of https://github.com/google/nomulus synced 2026-09-12 02:56:26 +00:00
Code Issues Releases Wiki Activity

Remove Nordn pull queue code (#1966)

Browse Source 
The SQL-based flow is verified to work on production.
This commit is contained in:
Lai Jiang
2023-03-16 17:37:48 -04:00
committed by GitHub
parent 178a2323d9
commit 95f1bca3fb
11 changed files with 106 additions and 567 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/google/registry/env/common/default/WEB-INF/queue.xml
Vendored
+5 -13
View File
@@ -6,6 +6,7 @@
<mode>pull</mode>
</queue>
<!-- Queue for publishing DNS updates in batches. -->
<queue>
<name>dns-publish</name>
<rate>100/s</rate>
@@ -18,6 +19,7 @@
</retry-parameters>
</queue>
<!-- Queue for uploading RDE deposits to the escrow provider. -->
<queue>
<name>rde-upload</name>
<rate>10/m</rate>
@@ -28,6 +30,7 @@
</retry-parameters>
</queue>
<!-- Queue for uploading RDE reports to ICANN. -->
<queue>
<name>rde-report</name>
<rate>1/s</rate>
@@ -37,6 +40,7 @@
</retry-parameters>
</queue>
<!-- Queue for copying BRDA deposits to GCS. -->
<queue>
<name>brda</name>
<rate>1/m</rate>
@@ -74,7 +78,7 @@
</retry-parameters>
</queue>
<!-- Queue for tasks to produce LORDN CSV reports, either by the query or queue method. -->
<!-- Queue for tasks to produce LORDN CSV reports, populated by a Cloud Scheduler fanout job. -->
<queue>
<name>nordn</name>
<rate>1/s</rate>
@@ -84,18 +88,6 @@
</retry-parameters>
</queue>
<!-- Queue for LORDN Claims CSV rows to be periodically queried and then uploaded in batches. -->
<queue>
<name>lordn-claims</name>
<mode>pull</mode>
</queue>
<!-- Queue for LORDN Sunrise CSV rows to be periodically queried and then uploaded in batches. -->
<queue>
<name>lordn-sunrise</name>
<mode>pull</mode>
</queue>
<!-- Queue for tasks that sync data to Google Spreadsheets. -->
<queue>
<name>sheet</name>
core/src/main/java/google/registry/env/production/default/WEB-INF/cloud-scheduler-tasks.xml
Vendored
-24
View File
@@ -173,30 +173,6 @@
<schedule>0 */12 * * *</schedule>
</task>
<task>
<url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordnPhase=sunrise&pullQueue]]></url>
<name>nordnUploadSunrisePullQueue</name>
<description>
This job uploads LORDN Sunrise CSV files for each TLD to MarksDB using
pull queue. It should be run at most every three hours, or at absolute
minimum every 26 hours.
</description>
<!-- This may be set anywhere between "every 3 hours" and "every 25 hours". -->
<schedule>0 */12 * * *</schedule>
</task>
<task>
<url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordnPhase=claims&pullQueue]]></url>
<name>nordnUploadClaimsPullQueue</name>
<description>
This job uploads LORDN Claims CSV files for each TLD to MarksDB using pull
queue. It should be run at most every three hours, or at absolute minimum
every 26 hours.
</description>
<!-- This may be set anywhere between "every 3 hours" and "every 25 hours". -->
<schedule>0 */12 * * *</schedule>
</task>
<task>
<url><![CDATA[/_dr/cron/fanout?queue=retryable-cron-tasks&endpoint=/_dr/task/deleteProberData&runInEmpty]]></url>
<name>deleteProberData</name>
core/src/main/java/google/registry/flows/domain/DomainCreateFlow.java
+6 -26
View File
@@ -86,8 +86,6 @@ import google.registry.model.billing.BillingEvent.Flag;
import google.registry.model.billing.BillingEvent.Reason;
import google.registry.model.billing.BillingEvent.Recurring;
import google.registry.model.billing.BillingEvent.RenewalPriceBehavior;
import google.registry.model.common.DatabaseMigrationStateSchedule;
import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
import google.registry.model.domain.Domain;
import google.registry.model.domain.DomainCommand;
import google.registry.model.domain.DomainCommand.Create;
@@ -124,7 +122,6 @@ import google.registry.model.tld.label.ReservationType;
import google.registry.model.tmch.ClaimsList;
import google.registry.model.tmch.ClaimsListDao;
import google.registry.persistence.VKey;
import google.registry.tmch.LordnTaskUtils;
import google.registry.tmch.LordnTaskUtils.LordnPhase;
import java.util.Map;
import java.util.Optional;
@@ -406,14 +403,9 @@ public final class DomainCreateFlow implements TransactionalFlow {
.addGracePeriod(
GracePeriod.forBillingEvent(GracePeriodStatus.ADD, repoId, createBillingEvent))
.setLordnPhase(
!(DatabaseMigrationStateSchedule.getValueAtTime(tm().getTransactionTime())
.equals(MigrationState.NORDN_SQL)
|| DatabaseMigrationStateSchedule.getValueAtTime(tm().getTransactionTime())
.equals(MigrationState.DNS_SQL))
? LordnPhase.NONE
: hasSignedMarks
? LordnPhase.SUNRISE
: hasClaimsNotice ? LordnPhase.CLAIMS : LordnPhase.NONE);
hasSignedMarks
? LordnPhase.SUNRISE
: hasClaimsNotice ? LordnPhase.CLAIMS : LordnPhase.NONE);
Domain domain = domainBuilder.build();
if (allocationToken.isPresent()
&& allocationToken.get().getTokenType().equals(TokenType.PACKAGE)) {
@@ -436,8 +428,9 @@ public final class DomainCreateFlow implements TransactionalFlow {
allocationTokenFlowUtils.redeemToken(
allocationToken.get(), domainHistory.getHistoryEntryId()));
}
enqueueTasks(domain, hasSignedMarks, hasClaimsNotice);
if (domain.shouldPublishToDns()) {
dnsUtils.requestDomainDnsRefresh(domain.getDomainName());
}
EntityChanges entityChanges =
flowCustomLogic.beforeSave(
DomainCreateFlowCustomLogic.BeforeSaveParameters.newBuilder()
@@ -710,19 +703,6 @@ public final class DomainCreateFlow implements TransactionalFlow {
.build();
}
private void enqueueTasks(Domain newDomain, boolean hasSignedMarks, boolean hasClaimsNotice) {
if (newDomain.shouldPublishToDns()) {
dnsUtils.requestDomainDnsRefresh(newDomain.getDomainName());
}
if (!(DatabaseMigrationStateSchedule.getValueAtTime(tm().getTransactionTime())
.equals(MigrationState.NORDN_SQL)
|| DatabaseMigrationStateSchedule.getValueAtTime(tm().getTransactionTime())
.equals(MigrationState.DNS_SQL))
&& (hasClaimsNotice || hasSignedMarks)) {
LordnTaskUtils.enqueueDomainTask(newDomain);
}
}
/**
* Determines the {@link RenewalPriceBehavior} and the renewal price that needs be stored in the
* {@link Recurring} billing events.
core/src/main/java/google/registry/tmch/LordnTaskUtils.java
+3 -42
View File
@@ -14,21 +14,15 @@
package google.registry.tmch;
import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
import static com.google.common.base.Preconditions.checkState;
import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
import com.google.appengine.api.taskqueue.TaskOptions;
import com.google.appengine.api.taskqueue.TaskOptions.Method;
import com.google.common.base.Joiner;
import google.registry.model.domain.Domain;
import google.registry.model.registrar.Registrar;
import java.util.Optional;
import org.joda.time.DateTime;
/**
* Helper methods for creating tasks containing CSV line data in the lordn-sunrise and lordn-claims
* queues based on {@link Domain} changes.
* Helper methods for creating tasks containing CSV line data based on {@link Domain#getLordnPhase}.
*
* <p>Note that, per the <a href="https://tools.ietf.org/html/draft-ietf-regext-tmch-func-spec-04">
* TMCH RFC</a>, while the application-datetime data is optional (which we never send because there
@@ -36,65 +30,32 @@ import org.joda.time.DateTime;
*/
public final class LordnTaskUtils {
public static final String QUEUE_SUNRISE = "lordn-sunrise";
public static final String QUEUE_CLAIMS = "lordn-claims";
public static final String COLUMNS_CLAIMS =
"roid,domain-name,notice-id,registrar-id,"
+ "registration-datetime,ack-datetime,application-datetime";
public static final String COLUMNS_SUNRISE =
"roid,domain-name,SMD-id,registrar-id," + "registration-datetime,application-datetime";
/** Enqueues a task in the LORDN queue representing a line of CSV for LORDN export. */
public static void enqueueDomainTask(Domain domain) {
tm().assertInTransaction();
// This method needs to use transactionTime as the Domain's creationTime because CreationTime
// isn't yet populated when this method is called during the resource flow.
String tld = domain.getTld();
if (domain.getLaunchNotice() == null) {
getQueue(QUEUE_SUNRISE)
.add(
TaskOptions.Builder.withTag(tld)
.method(Method.PULL)
.payload(getCsvLineForSunriseDomain(domain, tm().getTransactionTime())));
} else {
getQueue(QUEUE_CLAIMS)
.add(
TaskOptions.Builder.withTag(tld)
.method(Method.PULL)
.payload(getCsvLineForClaimsDomain(domain, tm().getTransactionTime())));
}
}
/** Returns the corresponding CSV LORDN line for a sunrise domain. */
public static String getCsvLineForSunriseDomain(Domain domain) {
return getCsvLineForSunriseDomain(domain, domain.getCreationTime());
}
// TODO: Merge into the function above after pull queue migration.
private static String getCsvLineForSunriseDomain(Domain domain, DateTime transactionTime) {
return Joiner.on(',')
.join(
domain.getRepoId(),
domain.getDomainName(),
domain.getSmdId(),
getIanaIdentifier(domain.getCreationRegistrarId()),
transactionTime); // Used as creation time.
domain.getCreationTime()); // Used as creation time.
}
/** Returns the corresponding CSV LORDN line for a claims domain. */
public static String getCsvLineForClaimsDomain(Domain domain) {
return getCsvLineForClaimsDomain(domain, domain.getCreationTime());
}
// TODO: Merge into the function above after pull queue migration.
private static String getCsvLineForClaimsDomain(Domain domain, DateTime transactionTime) {
return Joiner.on(',')
.join(
domain.getRepoId(),
domain.getDomainName(),
domain.getLaunchNotice().getNoticeId().getTcnId(),
getIanaIdentifier(domain.getCreationRegistrarId()),
transactionTime, // Used as creation time.
domain.getCreationTime(), // Used as creation time.
domain.getLaunchNotice().getAcceptedTime());
}
core/src/main/java/google/registry/tmch/NordnUploadAction.java
+52 -98
View File
@@ -14,7 +14,6 @@
package google.registry.tmch;
import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.net.HttpHeaders.LOCATION;
@@ -43,7 +42,6 @@ import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Ordering;
import com.google.common.flogger.FluentLogger;
import google.registry.batch.CloudTasksUtils;
@@ -66,7 +64,6 @@ import java.net.URL;
import java.security.GeneralSecurityException;
import java.security.SecureRandom;
import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import javax.inject.Inject;
@@ -90,8 +87,6 @@ public final class NordnUploadAction implements Runnable {
static final String PATH = "/_dr/task/nordnUpload";
static final String LORDN_PHASE_PARAM = "lordnPhase";
// TODO: Delete after migrating off of pull queue.
static final String PULL_QUEUE_PARAM = "pullQueue";
private static final int BATCH_SIZE = 1000;
private static final FluentLogger logger = FluentLogger.forEnclosingClass();
@@ -122,10 +117,6 @@ public final class NordnUploadAction implements Runnable {
@Parameter(RequestParameters.PARAM_TLD)
String tld;
@Inject
@Parameter(PULL_QUEUE_PARAM)
Optional<Boolean> usePullQueue;
@Inject CloudTasksUtils cloudTasksUtils;
@Inject
@@ -146,67 +137,59 @@ public final class NordnUploadAction implements Runnable {
@Override
public void run() {
if (usePullQueue.orElse(false)) {
try {
processLordnTasks();
} catch (IOException | GeneralSecurityException e) {
throw new RuntimeException(e);
}
} else {
checkArgument(
phase.equals(PARAM_LORDN_PHASE_SUNRISE) || phase.equals(PARAM_LORDN_PHASE_CLAIMS),
"Invalid phase specified to NordnUploadAction: %s.",
phase);
tm().transact(
() -> {
// Note here that we load all domains pending Nordn in one batch, which should not
// be a problem for the rate of domain registration that we see. If we anticipate
// a peak in claims during TLD launch (sunrise is NOT first-come-first-serve, so
// there should be no expectation of a peak during it), we can consider temporarily
// increasing the frequency of Nordn upload to reduce the size of each batch.
//
// We did not further divide the domains into smaller batches because the
// read-upload-write operation per small batch needs to be inside a single
// transaction to prevent race conditions, and running several uploads in rapid
// sucession will likely overwhelm the MarksDB upload server, which recommands a
// maximum upload frequency of every 3 hours.
//
// See:
// https://datatracker.ietf.org/doc/html/draft-ietf-regext-tmch-func-spec-01#section-5.2.3.3
List<Domain> domains =
tm().createQueryComposer(Domain.class)
.where("lordnPhase", EQ, LordnPhase.valueOf(Ascii.toUpperCase(phase)))
.where("tld", EQ, tld)
.orderBy("creationTime")
.list();
if (domains.isEmpty()) {
return;
}
StringBuilder csv = new StringBuilder();
ImmutableList.Builder<Domain> newDomains = new ImmutableList.Builder<>();
checkArgument(
phase.equals(PARAM_LORDN_PHASE_SUNRISE) || phase.equals(PARAM_LORDN_PHASE_CLAIMS),
"Invalid phase specified to NordnUploadAction: %s.",
phase);
tm().transact(
() -> {
// Note here that we load all domains pending Nordn in one batch, which should not
// be a problem for the rate of domain registration that we see. If we anticipate
// a peak in claims during TLD launch (sunrise is NOT first-come-first-serve, so
// there should be no expectation of a peak during it), we can consider temporarily
// increasing the frequency of Nordn upload to reduce the size of each batch.
//
// We did not further divide the domains into smaller batches because the
// read-upload-write operation per small batch needs to be inside a single
// transaction to prevent race conditions, and running several uploads in rapid
// sucession will likely overwhelm the MarksDB upload server, which recommands a
// maximum upload frequency of every 3 hours.
//
// See:
// https://datatracker.ietf.org/doc/html/draft-ietf-regext-tmch-func-spec-01#section-5.2.3.3
List<Domain> domains =
tm().createQueryComposer(Domain.class)
.where("lordnPhase", EQ, LordnPhase.valueOf(Ascii.toUpperCase(phase)))
.where("tld", EQ, tld)
.orderBy("creationTime")
.list();
if (domains.isEmpty()) {
return;
}
StringBuilder csv = new StringBuilder();
ImmutableList.Builder<Domain> newDomains = new ImmutableList.Builder<>();
domains.forEach(
domain -> {
if (phase.equals(PARAM_LORDN_PHASE_SUNRISE)) {
csv.append(getCsvLineForSunriseDomain(domain)).append('\n');
} else {
csv.append(getCsvLineForClaimsDomain(domain)).append('\n');
}
Domain newDomain = domain.asBuilder().setLordnPhase(LordnPhase.NONE).build();
newDomains.add(newDomain);
});
String columns =
phase.equals(PARAM_LORDN_PHASE_SUNRISE) ? COLUMNS_SUNRISE : COLUMNS_CLAIMS;
String header =
String.format("1,%s,%d\n%s\n", clock.nowUtc(), domains.size(), columns);
try {
uploadCsvToLordn(String.format("/LORDN/%s/%s", tld, phase), header + csv);
} catch (IOException | GeneralSecurityException e) {
throw new RuntimeException(e);
}
tm().updateAll(newDomains.build());
});
}
domains.forEach(
domain -> {
if (phase.equals(PARAM_LORDN_PHASE_SUNRISE)) {
csv.append(getCsvLineForSunriseDomain(domain)).append('\n');
} else {
csv.append(getCsvLineForClaimsDomain(domain)).append('\n');
}
Domain newDomain = domain.asBuilder().setLordnPhase(LordnPhase.NONE).build();
newDomains.add(newDomain);
});
String columns =
phase.equals(PARAM_LORDN_PHASE_SUNRISE) ? COLUMNS_SUNRISE : COLUMNS_CLAIMS;
String header =
String.format("1,%s,%d\n%s\n", clock.nowUtc(), domains.size(), columns);
try {
uploadCsvToLordn(String.format("/LORDN/%s/%s", tld, phase), header + csv);
} catch (IOException | GeneralSecurityException e) {
throw new RuntimeException(e);
}
tm().updateAll(newDomains.build());
});
}
/**
@@ -249,35 +232,6 @@ public final class NordnUploadAction implements Runnable {
}
}
private void processLordnTasks() throws IOException, GeneralSecurityException {
checkArgument(
phase.equals(PARAM_LORDN_PHASE_SUNRISE) || phase.equals(PARAM_LORDN_PHASE_CLAIMS),
"Invalid phase specified to NordnUploadAction: %s.",
phase);
DateTime now = clock.nowUtc();
Queue queue =
getQueue(
phase.equals(PARAM_LORDN_PHASE_SUNRISE)
? LordnTaskUtils.QUEUE_SUNRISE
: LordnTaskUtils.QUEUE_CLAIMS);
String columns = phase.equals(PARAM_LORDN_PHASE_SUNRISE) ? COLUMNS_SUNRISE : COLUMNS_CLAIMS;
List<TaskHandle> tasks = loadAllTasks(queue, tld);
// Note: This upload/task deletion isn't done atomically (it's not clear how one would do so
// anyway). As a result, it is possible that the upload might succeed yet the deletion of
// enqueued tasks might fail. If so, this would result in the same lines being uploaded to NORDN
// across multiple uploads. This is probably OK; all that we really cannot have is a missing
// line.
if (!tasks.isEmpty()) {
String csvData = convertTasksToCsv(tasks, now, columns);
uploadCsvToLordn(String.format("/LORDN/%s/%s", tld, phase), csvData);
Lists.partition(tasks, BATCH_SIZE)
.forEach(
batch ->
retrier.callWithRetry(
() -> queue.deleteTask(batch), TransientFailureException.class));
}
}
/**
* Upload LORDN file to MarksDB.
*
core/src/main/java/google/registry/tmch/TmchModule.java
-8
View File
@@ -16,7 +16,6 @@ package google.registry.tmch;
import static com.google.common.io.Resources.asByteSource;
import static com.google.common.io.Resources.getResource;
import static google.registry.request.RequestParameters.extractOptionalBooleanParameter;
import static google.registry.request.RequestParameters.extractRequiredParameter;
import dagger.Module;
@@ -26,7 +25,6 @@ import google.registry.request.HttpException.BadRequestException;
import google.registry.request.Parameter;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Optional;
import javax.servlet.http.HttpServletRequest;
import org.bouncycastle.openpgp.PGPPublicKey;
@@ -66,10 +64,4 @@ public final class TmchModule {
static String provideNordnLogId(HttpServletRequest req) {
return extractRequiredParameter(req, NordnVerifyAction.NORDN_LOG_ID_PARAM);
}
@Provides
@Parameter(NordnUploadAction.PULL_QUEUE_PARAM)
static Optional<Boolean> provideUsePullQueue(HttpServletRequest req) {
return extractOptionalBooleanParameter(req, NordnUploadAction.PULL_QUEUE_PARAM);
}
}