mirror of
https://github.com/google/nomulus
synced 2026-07-31 20:36:09 +00:00
Add locking and a response in ReplicateToDatastoreAction (#1328)
* Add locking and a response in ReplicateToDatastoreAction The response is necessary to get nicer logs in GAE and nicer cron job behavior. In addition: - fix issues where locks would be backed up and replayed to Datastore (they shouldn't be replayed) - do ignore-read-only writes when replaying the transactions
This commit is contained in:
gbrodman
@@ -18,6 +18,10 @@ import static google.registry.model.ofy.ObjectifyService.auditedOfy;
|
||||
import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
|
||||
import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
|
||||
import static google.registry.request.Action.Method.GET;
|
||||
import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
|
||||
import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
|
||||
import static javax.servlet.http.HttpServletResponse.SC_OK;
|
||||
import static org.joda.time.Duration.standardHours;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import com.google.common.collect.ImmutableList;
|
||||
@@ -26,15 +30,20 @@ import google.registry.model.UpdateAutoTimestamp;
|
||||
import google.registry.model.common.DatabaseMigrationStateSchedule;
|
||||
import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
|
||||
import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
|
||||
import google.registry.model.server.Lock;
|
||||
import google.registry.persistence.transaction.Transaction;
|
||||
import google.registry.persistence.transaction.TransactionEntity;
|
||||
import google.registry.request.Action;
|
||||
import google.registry.request.Response;
|
||||
import google.registry.request.auth.Auth;
|
||||
import google.registry.util.Clock;
|
||||
import google.registry.util.RequestStatusChecker;
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import javax.inject.Inject;
|
||||
import javax.persistence.NoResultException;
|
||||
import org.joda.time.Duration;
|
||||
|
||||
/** Cron task to replicate from Cloud SQL to datastore. */
|
||||
@Action(
|
||||
@@ -55,11 +64,18 @@ public class ReplicateToDatastoreAction implements Runnable {
|
||||
*/
|
||||
public static final int BATCH_SIZE = 200;
|
||||
|
||||
private static final Duration LEASE_LENGTH = standardHours(1);
|
||||
|
||||
private final Clock clock;
|
||||
private final RequestStatusChecker requestStatusChecker;
|
||||
private final Response response;
|
||||
|
||||
@Inject
|
||||
public ReplicateToDatastoreAction(Clock clock) {
|
||||
public ReplicateToDatastoreAction(
|
||||
Clock clock, RequestStatusChecker requestStatusChecker, Response response) {
|
||||
this.clock = clock;
|
||||
this.requestStatusChecker = requestStatusChecker;
|
||||
this.response = response;
|
||||
}
|
||||
|
||||
@VisibleForTesting
|
||||
@@ -143,24 +159,55 @@ public class ReplicateToDatastoreAction implements Runnable {
|
||||
public void run() {
|
||||
MigrationState state = DatabaseMigrationStateSchedule.getValueAtTime(clock.nowUtc());
|
||||
if (!state.getReplayDirection().equals(ReplayDirection.SQL_TO_DATASTORE)) {
|
||||
logger.atInfo().log(
|
||||
"Skipping ReplicateToDatastoreAction because we are in migration phase %s.", state);
|
||||
String message =
|
||||
String.format(
|
||||
"Skipping ReplicateToDatastoreAction because we are in migration phase %s.", state);
|
||||
logger.atInfo().log(message);
|
||||
// App Engine will retry on any non-2xx status code, which we don't want in this case.
|
||||
response.setStatus(SC_NO_CONTENT);
|
||||
response.setPayload(message);
|
||||
return;
|
||||
}
|
||||
// TODO(b/181758163): Deal with objects that don't exist in Cloud SQL, e.g. ForeignKeyIndex,
|
||||
// EppResourceIndex.
|
||||
logger.atInfo().log("Processing transaction replay batch Cloud SQL -> Cloud Datastore");
|
||||
int numTransactionsReplayed = 0;
|
||||
for (TransactionEntity txnEntity : getTransactionBatch()) {
|
||||
try {
|
||||
applyTransaction(txnEntity);
|
||||
} catch (Throwable t) {
|
||||
logger.atSevere().withCause(t).log("Errored out replaying files");
|
||||
return;
|
||||
}
|
||||
numTransactionsReplayed++;
|
||||
Optional<Lock> lock =
|
||||
Lock.acquire(
|
||||
this.getClass().getSimpleName(), null, LEASE_LENGTH, requestStatusChecker, false);
|
||||
if (!lock.isPresent()) {
|
||||
String message = "Can't acquire ReplicateToDatastoreAction lock, aborting.";
|
||||
logger.atSevere().log(message);
|
||||
// App Engine will retry on any non-2xx status code, which we don't want in this case.
|
||||
response.setStatus(SC_NO_CONTENT);
|
||||
response.setPayload(message);
|
||||
return;
|
||||
}
|
||||
logger.atInfo().log(
|
||||
"Replayed %d transactions from Cloud SQL -> Datastore", numTransactionsReplayed);
|
||||
try {
|
||||
logger.atInfo().log("Processing transaction replay batch Cloud SQL -> Cloud Datastore");
|
||||
int numTransactionsReplayed = replayAllTransactions();
|
||||
String resultMessage =
|
||||
String.format(
|
||||
"Replayed %d transaction(s) from Cloud SQL -> Datastore", numTransactionsReplayed);
|
||||
logger.atInfo().log(resultMessage);
|
||||
response.setPayload(resultMessage);
|
||||
response.setStatus(SC_OK);
|
||||
} catch (Throwable t) {
|
||||
String message = "Errored out replaying files";
|
||||
logger.atSevere().withCause(t).log(message);
|
||||
response.setStatus(SC_INTERNAL_SERVER_ERROR);
|
||||
response.setPayload(message);
|
||||
} finally {
|
||||
lock.ifPresent(Lock::release);
|
||||
}
|
||||
}
|
||||
|
||||
private int replayAllTransactions() {
|
||||
int numTransactionsReplayed = 0;
|
||||
List<TransactionEntity> transactionBatch;
|
||||
do {
|
||||
transactionBatch = getTransactionBatch();
|
||||
for (TransactionEntity transaction : transactionBatch) {
|
||||
applyTransaction(transaction);
|
||||
numTransactionsReplayed++;
|
||||
}
|
||||
} while (!transactionBatch.isEmpty());
|
||||
return numTransactionsReplayed;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@
|
||||
package google.registry.model.server;
|
||||
|
||||
import static com.google.common.base.Preconditions.checkArgument;
|
||||
import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
|
||||
import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
|
||||
import static google.registry.util.DateTimeUtils.isAtOrAfter;
|
||||
import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
|
||||
@@ -35,6 +36,7 @@ import google.registry.util.RequestStatusChecker;
|
||||
import google.registry.util.RequestStatusCheckerImpl;
|
||||
import java.io.Serializable;
|
||||
import java.util.Optional;
|
||||
import java.util.function.Supplier;
|
||||
import javax.annotation.Nullable;
|
||||
import javax.persistence.Column;
|
||||
import javax.persistence.IdClass;
|
||||
@@ -215,45 +217,45 @@ public class Lock extends ImmutableObject implements DatastoreAndSqlEntity, Seri
|
||||
// It's important to use transactNew rather than transact, because a Lock can be used to control
|
||||
// access to resources like GCS that can't be transactionally rolled back. Therefore, the lock
|
||||
// must be definitively acquired before it is used, even when called inside another transaction.
|
||||
Supplier<AcquireResult> lockAcquirer =
|
||||
() -> {
|
||||
DateTime now = tm().getTransactionTime();
|
||||
|
||||
// Checking if an unexpired lock still exists - if so, the lock can't be acquired.
|
||||
Lock lock =
|
||||
tm().loadByKeyIfPresent(
|
||||
VKey.create(
|
||||
Lock.class,
|
||||
new LockId(resourceName, scope),
|
||||
Key.create(Lock.class, lockId)))
|
||||
.orElse(null);
|
||||
if (lock != null) {
|
||||
logger.atInfo().log(
|
||||
"Loaded existing lock: %s for request: %s", lock.lockId, lock.requestLogId);
|
||||
}
|
||||
LockState lockState;
|
||||
if (lock == null) {
|
||||
lockState = LockState.FREE;
|
||||
} else if (isAtOrAfter(now, lock.expirationTime)) {
|
||||
lockState = LockState.TIMED_OUT;
|
||||
} else if (checkThreadRunning && !requestStatusChecker.isRunning(lock.requestLogId)) {
|
||||
lockState = LockState.OWNER_DIED;
|
||||
} else {
|
||||
lockState = LockState.IN_USE;
|
||||
return AcquireResult.create(now, lock, null, lockState);
|
||||
}
|
||||
|
||||
Lock newLock =
|
||||
create(resourceName, scope, requestStatusChecker.getLogId(), now, leaseLength);
|
||||
// Locks are not parented under an EntityGroupRoot (so as to avoid write
|
||||
// contention) and don't need to be backed up.
|
||||
tm().putIgnoringReadOnly(newLock);
|
||||
|
||||
return AcquireResult.create(now, lock, newLock, lockState);
|
||||
};
|
||||
// In ofy, backup is determined per-action, but in SQL it's determined per-transaction
|
||||
AcquireResult acquireResult =
|
||||
tm().transactNew(
|
||||
() -> {
|
||||
DateTime now = tm().getTransactionTime();
|
||||
|
||||
// Checking if an unexpired lock still exists - if so, the lock can't be acquired.
|
||||
Lock lock =
|
||||
tm().loadByKeyIfPresent(
|
||||
VKey.create(
|
||||
Lock.class,
|
||||
new LockId(resourceName, scope),
|
||||
Key.create(Lock.class, lockId)))
|
||||
.orElse(null);
|
||||
if (lock != null) {
|
||||
logger.atInfo().log(
|
||||
"Loaded existing lock: %s for request: %s", lock.lockId, lock.requestLogId);
|
||||
}
|
||||
LockState lockState;
|
||||
if (lock == null) {
|
||||
lockState = LockState.FREE;
|
||||
} else if (isAtOrAfter(now, lock.expirationTime)) {
|
||||
lockState = LockState.TIMED_OUT;
|
||||
} else if (checkThreadRunning
|
||||
&& !requestStatusChecker.isRunning(lock.requestLogId)) {
|
||||
lockState = LockState.OWNER_DIED;
|
||||
} else {
|
||||
lockState = LockState.IN_USE;
|
||||
return AcquireResult.create(now, lock, null, lockState);
|
||||
}
|
||||
|
||||
Lock newLock =
|
||||
create(
|
||||
resourceName, scope, requestStatusChecker.getLogId(), now, leaseLength);
|
||||
// Locks are not parented under an EntityGroupRoot (so as to avoid write
|
||||
// contention) and don't need to be backed up.
|
||||
tm().putIgnoringReadOnly(newLock);
|
||||
|
||||
return AcquireResult.create(now, lock, newLock, lockState);
|
||||
});
|
||||
tm().isOfy() ? tm().transactNew(lockAcquirer) : jpaTm().transactWithoutBackup(lockAcquirer);
|
||||
|
||||
logAcquireResult(acquireResult);
|
||||
lockMetrics.recordAcquire(resourceName, scope, acquireResult.lockState());
|
||||
@@ -263,34 +265,41 @@ public class Lock extends ImmutableObject implements DatastoreAndSqlEntity, Seri
|
||||
/** Release the lock. */
|
||||
public void release() {
|
||||
// Just use the default clock because we aren't actually doing anything that will use the clock.
|
||||
tm().transact(
|
||||
() -> {
|
||||
// To release a lock, check that no one else has already obtained it and if not
|
||||
// delete it. If the lock in Datastore was different then this lock is gone already;
|
||||
// this can happen if release() is called around the expiration time and the lock
|
||||
// expires underneath us.
|
||||
VKey<Lock> key =
|
||||
VKey.create(
|
||||
Lock.class, new LockId(resourceName, tld), Key.create(Lock.class, lockId));
|
||||
Lock loadedLock = tm().loadByKeyIfPresent(key).orElse(null);
|
||||
if (Lock.this.equals(loadedLock)) {
|
||||
// Use deleteWithoutBackup() so that we don't create a commit log entry for deleting
|
||||
// the lock.
|
||||
logger.atInfo().log("Deleting lock: %s", lockId);
|
||||
tm().deleteIgnoringReadOnly(key);
|
||||
Supplier<Void> lockReleaser =
|
||||
() -> {
|
||||
// To release a lock, check that no one else has already obtained it and if not
|
||||
// delete it. If the lock in Datastore was different then this lock is gone already;
|
||||
// this can happen if release() is called around the expiration time and the lock
|
||||
// expires underneath us.
|
||||
VKey<Lock> key =
|
||||
VKey.create(
|
||||
Lock.class, new LockId(resourceName, tld), Key.create(Lock.class, lockId));
|
||||
Lock loadedLock = tm().loadByKeyIfPresent(key).orElse(null);
|
||||
if (Lock.this.equals(loadedLock)) {
|
||||
// Use deleteIgnoringReadOnly() so that we don't create a commit log entry for deleting
|
||||
// the lock.
|
||||
logger.atInfo().log("Deleting lock: %s", lockId);
|
||||
tm().deleteIgnoringReadOnly(key);
|
||||
|
||||
lockMetrics.recordRelease(
|
||||
resourceName, tld, new Duration(acquiredTime, tm().getTransactionTime()));
|
||||
} else {
|
||||
logger.atSevere().log(
|
||||
"The lock we acquired was transferred to someone else before we"
|
||||
+ " released it! Did action take longer than lease length?"
|
||||
+ " Our lock: %s, current lock: %s",
|
||||
Lock.this, loadedLock);
|
||||
logger.atInfo().log(
|
||||
"Not deleting lock: %s - someone else has it: %s", lockId, loadedLock);
|
||||
}
|
||||
});
|
||||
lockMetrics.recordRelease(
|
||||
resourceName, tld, new Duration(acquiredTime, tm().getTransactionTime()));
|
||||
} else {
|
||||
logger.atSevere().log(
|
||||
"The lock we acquired was transferred to someone else before we"
|
||||
+ " released it! Did action take longer than lease length?"
|
||||
+ " Our lock: %s, current lock: %s",
|
||||
Lock.this, loadedLock);
|
||||
logger.atInfo().log(
|
||||
"Not deleting lock: %s - someone else has it: %s", lockId, loadedLock);
|
||||
}
|
||||
return null;
|
||||
};
|
||||
// In ofy, backup is determined per-action, but in SQL it's determined per-transaction
|
||||
if (tm().isOfy()) {
|
||||
tm().transact(lockReleaser);
|
||||
} else {
|
||||
jpaTm().transactWithoutBackup(lockReleaser);
|
||||
}
|
||||
}
|
||||
|
||||
static class LockId extends ImmutableObject implements Serializable {
|
||||
|
||||
@@ -242,7 +242,7 @@ public class Transaction extends ImmutableObject implements Buildable {
|
||||
if (entity instanceof DatastoreEntity) {
|
||||
((DatastoreEntity) entity).beforeDatastoreSaveOnReplay();
|
||||
}
|
||||
ofyTm().put(entity);
|
||||
ofyTm().putIgnoringReadOnly(entity);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -280,7 +280,7 @@ public class Transaction extends ImmutableObject implements Buildable {
|
||||
|
||||
@Override
|
||||
public void writeToDatastore() {
|
||||
ofyTm().delete(key);
|
||||
ofyTm().deleteIgnoringReadOnly(key);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
Reference in New Issue
Block a user