mirrors/nomulus
1
0
Fork 0
mirror of https://github.com/google/nomulus synced 2026-09-20 06:54:43 +00:00
Code Issues Releases Wiki Activity

Remove the use of GCS APIs provided from GAE SDK (#1228)

Browse Source 
The API provided by the GAE SDK will not be available outside GAE
runtime. This presents a problem when we migrate off of GAE. More
pressingly, the RDE pipeline migration to Beam requires that we write to
GCS on GCE. Previously we were able to sidestep the issue by delegating
the writes to FileIO provided by Beam, which knows how to write to GCS.
However the RDE pipeline cannot use FileIO directly as it needs to write
to multiple files in one go and explicit use of GCS API is needed.

An unfortunate side effect of the API migration is that the new testing
library contains a bug which makes serializing GcsUtils impossible. It
is fixed upstream but not released yet. The fix has been backported for
the time being.

<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/google/nomulus/1228)
<!-- Reviewable:end -->
This commit is contained in:
Lai Jiang
2021-07-13 14:52:37 -04:00
committed by GitHub
parent 62c556cebf
commit 676616a172
186 changed files with 2415 additions and 1600 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/google/registry/backup/ExportCommitLogDiffAction.java
+12 -12
View File
@@ -28,18 +28,17 @@ import static google.registry.model.ofy.CommitLogBucket.getBucketKey;
import static google.registry.model.ofy.ObjectifyService.auditedOfy;
import static google.registry.util.DateTimeUtils.START_OF_TIME;
import static google.registry.util.DateTimeUtils.isAtOrAfter;
import static java.nio.channels.Channels.newOutputStream;
import static java.util.Comparator.comparingLong;
import com.google.appengine.tools.cloudstorage.GcsFileOptions;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.appengine.tools.cloudstorage.GcsService;
import com.google.cloud.storage.BlobId;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Streams;
import com.google.common.flogger.FluentLogger;
import com.googlecode.objectify.Key;
import google.registry.config.RegistryConfig.Config;
import google.registry.gcs.GcsUtils;
import google.registry.model.ImmutableObject;
import google.registry.model.ofy.CommitLogBucket;
import google.registry.model.ofy.CommitLogCheckpoint;
@@ -74,7 +73,8 @@ public final class ExportCommitLogDiffAction implements Runnable {
public static final String DIFF_FILE_PREFIX = "commit_diff_until_";
@Inject GcsService gcsService;
@Inject GcsUtils gcsUtils;
@Inject @Config("commitLogGcsBucket") String gcsBucket;
@Inject @Config("commitLogDiffExportBatchSize") int batchSize;
@Inject @Parameter(LOWER_CHECKPOINT_TIME_PARAM) DateTime lowerCheckpointTime;
@@ -102,13 +102,13 @@ public final class ExportCommitLogDiffAction implements Runnable {
List<Key<CommitLogManifest>> sortedKeys = loadAllDiffKeys(lowerCheckpoint, upperCheckpoint);
logger.atInfo().log("Found %d manifests to export", sortedKeys.size());
// Open an output channel to GCS, wrapped in a stream for convenience.
try (OutputStream gcsStream = newOutputStream(gcsService.createOrReplace(
new GcsFilename(gcsBucket, DIFF_FILE_PREFIX + upperCheckpointTime),
new GcsFileOptions.Builder()
.addUserMetadata(LOWER_BOUND_CHECKPOINT, lowerCheckpointTime.toString())
.addUserMetadata(UPPER_BOUND_CHECKPOINT, upperCheckpointTime.toString())
.addUserMetadata(NUM_TRANSACTIONS, Integer.toString(sortedKeys.size()))
.build()))) {
try (OutputStream gcsStream =
gcsUtils.openOutputStream(
BlobId.of(gcsBucket, DIFF_FILE_PREFIX + upperCheckpointTime),
ImmutableMap.of(
LOWER_BOUND_CHECKPOINT, lowerCheckpointTime.toString(),
UPPER_BOUND_CHECKPOINT, upperCheckpointTime.toString(),
NUM_TRANSACTIONS, Integer.toString(sortedKeys.size())))) {
// Export the upper checkpoint itself.
serializeEntity(upperCheckpoint, gcsStream);
// If there are no manifests to export, stop early, now that we've written out the file with
core/src/main/java/google/registry/backup/GcsDiffFileLister.java
+33 -42
View File
@@ -21,19 +21,16 @@ import static google.registry.util.DateTimeUtils.START_OF_TIME;
import static google.registry.util.DateTimeUtils.isBeforeOrAt;
import static google.registry.util.DateTimeUtils.latestOf;
import com.google.appengine.tools.cloudstorage.GcsFileMetadata;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.appengine.tools.cloudstorage.GcsService;
import com.google.appengine.tools.cloudstorage.ListItem;
import com.google.appengine.tools.cloudstorage.ListOptions;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
import com.google.common.collect.ImmutableList;
import com.google.common.flogger.FluentLogger;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import google.registry.backup.BackupModule.Backups;
import google.registry.gcs.GcsUtils;
import java.io.IOException;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeMap;
import javax.annotation.Nullable;
@@ -45,7 +42,8 @@ class GcsDiffFileLister {
private static final FluentLogger logger = FluentLogger.forEnclosingClass();
@Inject GcsService gcsService;
@Inject GcsUtils gcsUtils;
@Inject @Backups ListeningExecutorService executor;
@Inject GcsDiffFileLister() {}
@@ -56,22 +54,22 @@ class GcsDiffFileLister {
*/
private boolean constructDiffSequence(
String gcsBucket,
Map<DateTime, ListenableFuture<GcsFileMetadata>> upperBoundTimesToMetadata,
Map<DateTime, ListenableFuture<BlobInfo>> upperBoundTimesToBlobInfo,
DateTime fromTime,
DateTime lastTime,
TreeMap<DateTime, GcsFileMetadata> sequence) {
TreeMap<DateTime, BlobInfo> sequence) {
DateTime checkpointTime = lastTime;
while (isBeforeOrAt(fromTime, checkpointTime)) {
GcsFileMetadata metadata;
if (upperBoundTimesToMetadata.containsKey(checkpointTime)) {
metadata = Futures.getUnchecked(upperBoundTimesToMetadata.get(checkpointTime));
BlobInfo blobInfo;
if (upperBoundTimesToBlobInfo.containsKey(checkpointTime)) {
blobInfo = Futures.getUnchecked(upperBoundTimesToBlobInfo.get(checkpointTime));
} else {
String filename = DIFF_FILE_PREFIX + checkpointTime;
logger.atInfo().log("Patching GCS list; discovered file: %s", filename);
metadata = getMetadata(gcsBucket, filename);
blobInfo = getBlobInfo(gcsBucket, filename);
// If we hit a gap, quit.
if (metadata == null) {
if (blobInfo == null) {
logger.atInfo().log(
"Gap discovered in sequence terminating at %s, missing file: %s",
sequence.lastKey(), filename);
@@ -79,14 +77,14 @@ class GcsDiffFileLister {
return false;
}
}
sequence.put(checkpointTime, metadata);
checkpointTime = getLowerBoundTime(metadata);
sequence.put(checkpointTime, blobInfo);
checkpointTime = getLowerBoundTime(blobInfo);
}
logger.atInfo().log("Found sequence from %s to %s", checkpointTime, lastTime);
return true;
}
ImmutableList<GcsFileMetadata> listDiffFiles(
ImmutableList<BlobInfo> listDiffFiles(
String gcsBucket, DateTime fromTime, @Nullable DateTime toTime) {
logger.atInfo().log("Requested restore from time: %s", fromTime);
if (toTime != null) {
@@ -95,28 +93,25 @@ class GcsDiffFileLister {
// List all of the diff files on GCS and build a map from each file's upper checkpoint time
// (extracted from the filename) to its asynchronously-loaded metadata, keeping only files with
// an upper checkpoint time > fromTime.
TreeMap<DateTime, ListenableFuture<GcsFileMetadata>> upperBoundTimesToMetadata
= new TreeMap<>();
Iterator<ListItem> listItems;
TreeMap<DateTime, ListenableFuture<BlobInfo>> upperBoundTimesToBlobInfo = new TreeMap<>();
ImmutableList<String> strippedFilenames;
try {
// TODO(b/23554360): Use a smarter prefixing strategy to speed this up.
listItems = gcsService.list(
gcsBucket,
new ListOptions.Builder().setPrefix(DIFF_FILE_PREFIX).build());
strippedFilenames = gcsUtils.listFolderObjects(gcsBucket, DIFF_FILE_PREFIX);
} catch (IOException e) {
throw new RuntimeException(e);
}
DateTime lastUpperBoundTime = START_OF_TIME;
while (listItems.hasNext()) {
final String filename = listItems.next().getName();
DateTime upperBoundTime = DateTime.parse(filename.substring(DIFF_FILE_PREFIX.length()));
for (String strippedFilename : strippedFilenames) {
final String filename = DIFF_FILE_PREFIX + strippedFilename;
DateTime upperBoundTime = DateTime.parse(strippedFilename);
if (isInRange(upperBoundTime, fromTime, toTime)) {
upperBoundTimesToMetadata.put(
upperBoundTime, executor.submit(() -> getMetadata(gcsBucket, filename)));
upperBoundTimesToBlobInfo.put(
upperBoundTime, executor.submit(() -> getBlobInfo(gcsBucket, filename)));
lastUpperBoundTime = latestOf(upperBoundTime, lastUpperBoundTime);
}
}
if (upperBoundTimesToMetadata.isEmpty()) {
if (upperBoundTimesToBlobInfo.isEmpty()) {
logger.atInfo().log("No files found");
return ImmutableList.of();
}
@@ -129,23 +124,23 @@ class GcsDiffFileLister {
// metadata of a file is sufficient to identify the preceding file, so if we start from the
// last file and work backwards we can verify that we have no holes in our chain (although we
// may be missing files at the end).
TreeMap<DateTime, GcsFileMetadata> sequence = new TreeMap<>();
TreeMap<DateTime, BlobInfo> sequence = new TreeMap<>();
logger.atInfo().log("Restoring until: %s", lastUpperBoundTime);
boolean inconsistentFileSet =
!constructDiffSequence(
gcsBucket, upperBoundTimesToMetadata, fromTime, lastUpperBoundTime, sequence);
gcsBucket, upperBoundTimesToBlobInfo, fromTime, lastUpperBoundTime, sequence);
// Verify that all of the elements in the original set are represented in the sequence. If we
// find anything that's not represented, construct a sequence for it.
boolean checkForMoreExtraDiffs = true; // Always loop at least once.
while (checkForMoreExtraDiffs) {
checkForMoreExtraDiffs = false;
for (DateTime key : upperBoundTimesToMetadata.descendingKeySet()) {
for (DateTime key : upperBoundTimesToBlobInfo.descendingKeySet()) {
if (!isInRange(key, fromTime, toTime)) {
break;
}
if (!sequence.containsKey(key)) {
constructDiffSequence(gcsBucket, upperBoundTimesToMetadata, fromTime, key, sequence);
constructDiffSequence(gcsBucket, upperBoundTimesToBlobInfo, fromTime, key, sequence);
checkForMoreExtraDiffs = true;
inconsistentFileSet = true;
break;
@@ -173,15 +168,11 @@ class GcsDiffFileLister {
return isBeforeOrAt(start, time) && (end == null || isBeforeOrAt(time, end));
}
private DateTime getLowerBoundTime(GcsFileMetadata metadata) {
return DateTime.parse(metadata.getOptions().getUserMetadata().get(LOWER_BOUND_CHECKPOINT));
private DateTime getLowerBoundTime(BlobInfo blobInfo) {
return DateTime.parse(blobInfo.getMetadata().get(LOWER_BOUND_CHECKPOINT));
}
private GcsFileMetadata getMetadata(String gcsBucket, String filename) {
try {
return gcsService.getMetadata(new GcsFilename(gcsBucket, filename));
} catch (IOException e) {
throw new RuntimeException(e);
}
private BlobInfo getBlobInfo(String gcsBucket, String filename) {
return gcsUtils.getBlobInfo(BlobId.of(gcsBucket, filename));
}
}
core/src/main/java/google/registry/backup/ReplayCommitLogsToSqlAction.java
+17 -24
View File
@@ -25,12 +25,12 @@ import static org.joda.time.Duration.standardHours;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.tools.cloudstorage.GcsFileMetadata;
import com.google.appengine.tools.cloudstorage.GcsService;
import com.google.cloud.storage.BlobInfo;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableList;
import com.google.common.flogger.FluentLogger;
import google.registry.config.RegistryConfig.Config;
import google.registry.gcs.GcsUtils;
import google.registry.model.common.DatabaseMigrationStateSchedule;
import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
@@ -51,7 +51,6 @@ import google.registry.util.Clock;
import google.registry.util.RequestStatusChecker;
import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.Channels;
import java.util.Optional;
import javax.inject.Inject;
import javax.servlet.http.HttpServletResponse;
@@ -70,14 +69,13 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
static final String PATH = "/_dr/task/replayCommitLogsToSql";
private static final FluentLogger logger = FluentLogger.forEnclosingClass();
private static final int BLOCK_SIZE =
1024 * 1024; // Buffer 1mb at a time, for no particular reason.
private static final Duration LEASE_LENGTH = standardHours(1);
// Stop / pause where we are if we've been replaying for more than five minutes to avoid GAE
// request timeouts
private static final Duration REPLAY_TIMEOUT_DURATION = Duration.standardMinutes(5);
@Inject GcsService gcsService;
@Inject GcsUtils gcsUtils;
@Inject Response response;
@Inject RequestStatusChecker requestStatusChecker;
@Inject GcsDiffFileLister diffLister;
@@ -122,13 +120,13 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
}
try {
logger.atInfo().log("Beginning replay of commit logs.");
ImmutableList<GcsFileMetadata> commitLogFiles = getFilesToReplay();
ImmutableList<BlobInfo> commitLogFiles = getFilesToReplay();
if (dryRun) {
response.setStatus(HttpServletResponse.SC_OK);
ImmutableList<String> filenames =
commitLogFiles.stream()
.limit(10)
.map(file -> file.getFilename().getObjectName())
.map(file -> file.getName())
.collect(toImmutableList());
String dryRunMessage =
"Running in dry-run mode; would have processed %d files. They are (limit 10):\n"
@@ -152,22 +150,22 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
}
}
private ImmutableList<GcsFileMetadata> getFilesToReplay() {
private ImmutableList<BlobInfo> getFilesToReplay() {
// Start at the first millisecond we haven't seen yet
DateTime fromTime = jpaTm().transact(() -> SqlReplayCheckpoint.get().plusMillis(1));
logger.atInfo().log("Starting replay from: %s.", fromTime);
// If there's an inconsistent file set, this will throw IllegalStateException and the job
// will try later -- this is likely because an export hasn't finished yet.
ImmutableList<GcsFileMetadata> commitLogFiles =
ImmutableList<BlobInfo> commitLogFiles =
diffLister.listDiffFiles(gcsBucket, fromTime, /* current time */ null);
logger.atInfo().log("Found %d new commit log files to process.", commitLogFiles.size());
return commitLogFiles;
}
private void replayFiles(ImmutableList<GcsFileMetadata> commitLogFiles) {
private void replayFiles(ImmutableList<BlobInfo> commitLogFiles) {
DateTime replayTimeoutTime = clock.nowUtc().plus(REPLAY_TIMEOUT_DURATION);
int processedFiles = 0;
for (GcsFileMetadata metadata : commitLogFiles) {
for (BlobInfo metadata : commitLogFiles) {
// One transaction per GCS file
jpaTm().transact(() -> processFile(metadata));
processedFiles++;
@@ -181,31 +179,26 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
logger.atInfo().log("Replayed %d commit log files to SQL successfully.", processedFiles);
}
private void processFile(GcsFileMetadata metadata) {
private void processFile(BlobInfo metadata) {
logger.atInfo().log(
"Processing commit log file %s of size %d B.",
metadata.getFilename(), metadata.getLength());
try (InputStream input =
Channels.newInputStream(
gcsService.openPrefetchingReadChannel(metadata.getFilename(), 0, BLOCK_SIZE))) {
"Processing commit log file %s of size %d B.", metadata.getName(), metadata.getSize());
try (InputStream input = gcsUtils.openInputStream(metadata.getBlobId())) {
// Load and process the Datastore transactions one at a time
ImmutableList<ImmutableList<VersionedEntity>> allTransactions =
CommitLogImports.loadEntitiesByTransaction(input);
logger.atInfo().log(
"Replaying %d transactions from commit log file %s.",
allTransactions.size(), metadata.getFilename());
allTransactions.size(), metadata.getName());
allTransactions.forEach(this::replayTransaction);
// if we succeeded, set the last-seen time
DateTime checkpoint =
DateTime.parse(
metadata.getFilename().getObjectName().substring(DIFF_FILE_PREFIX.length()));
DateTime checkpoint = DateTime.parse(metadata.getName().substring(DIFF_FILE_PREFIX.length()));
SqlReplayCheckpoint.set(checkpoint);
logger.atInfo().log(
"Replayed %d transactions from commit log file %s.",
allTransactions.size(), metadata.getFilename());
allTransactions.size(), metadata.getName());
} catch (IOException e) {
throw new RuntimeException(
"Errored out while replaying commit log file " + metadata.getFilename(), e);
"Errored out while replaying commit log file " + metadata.getName(), e);
}
}
core/src/main/java/google/registry/backup/RestoreCommitLogsAction.java
+8 -11
View File
@@ -23,8 +23,7 @@ import static google.registry.model.ofy.ObjectifyService.auditedOfy;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityTranslator;
import com.google.appengine.tools.cloudstorage.GcsFileMetadata;
import com.google.appengine.tools.cloudstorage.GcsService;
import com.google.cloud.storage.BlobInfo;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
@@ -36,6 +35,7 @@ import com.googlecode.objectify.Result;
import com.googlecode.objectify.util.ResultNow;
import google.registry.config.RegistryConfig.Config;
import google.registry.config.RegistryEnvironment;
import google.registry.gcs.GcsUtils;
import google.registry.model.ImmutableObject;
import google.registry.model.ofy.CommitLogBucket;
import google.registry.model.ofy.CommitLogCheckpoint;
@@ -48,7 +48,6 @@ import google.registry.request.auth.Auth;
import google.registry.util.Retrier;
import java.io.IOException;
import java.io.InputStream;
import java.nio.channels.Channels;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -69,8 +68,6 @@ public class RestoreCommitLogsAction implements Runnable {
private static final FluentLogger logger = FluentLogger.forEnclosingClass();
static final int BLOCK_SIZE = 1024 * 1024; // Buffer 1mb at a time, for no particular reason.
public static final String PATH = "/_dr/task/restoreCommitLogs";
static final String DRY_RUN_PARAM = "dryRun";
static final String FROM_TIME_PARAM = "fromTime";
@@ -80,7 +77,8 @@ public class RestoreCommitLogsAction implements Runnable {
private static final ImmutableSet<RegistryEnvironment> FORBIDDEN_ENVIRONMENTS =
ImmutableSet.of(RegistryEnvironment.PRODUCTION, RegistryEnvironment.SANDBOX);
@Inject GcsService gcsService;
@Inject GcsUtils gcsUtils;
@Inject @Parameter(DRY_RUN_PARAM) boolean dryRun;
@Inject @Parameter(FROM_TIME_PARAM) DateTime fromTime;
@Inject @Parameter(TO_TIME_PARAM) DateTime toTime;
@@ -109,17 +107,16 @@ public class RestoreCommitLogsAction implements Runnable {
}
String gcsBucket = gcsBucketOverride.orElse(defaultGcsBucket);
logger.atInfo().log("Restoring from %s.", gcsBucket);
List<GcsFileMetadata> diffFiles = diffLister.listDiffFiles(gcsBucket, fromTime, toTime);
List<BlobInfo> diffFiles = diffLister.listDiffFiles(gcsBucket, fromTime, toTime);
if (diffFiles.isEmpty()) {
logger.atInfo().log("Nothing to restore");
return;
}
Map<Integer, DateTime> bucketTimestamps = new HashMap<>();
CommitLogCheckpoint lastCheckpoint = null;
for (GcsFileMetadata metadata : diffFiles) {
logger.atInfo().log("Restoring: %s", metadata.getFilename().getObjectName());
try (InputStream input = Channels.newInputStream(
gcsService.openPrefetchingReadChannel(metadata.getFilename(), 0, BLOCK_SIZE))) {
for (BlobInfo metadata : diffFiles) {
logger.atInfo().log("Restoring: %s", metadata.getName());
try (InputStream input = gcsUtils.openInputStream(metadata.getBlobId())) {
PeekingIterator<ImmutableObject> commitLogs =
peekingIterator(createDeserializingIterator(input, true));
lastCheckpoint = (CommitLogCheckpoint) commitLogs.next();
core/src/main/java/google/registry/config/RegistryConfig.java
-11
View File
@@ -416,17 +416,6 @@ public final class RegistryConfig {
return Optional.ofNullable(config.cloudDns.servicePath);
}
/**
* Returns size of Google Cloud Storage client connection buffer in bytes.
*
* @see google.registry.gcs.GcsUtils
*/
@Provides
@Config("gcsBufferSize")
public static int provideGcsBufferSize() {
return 1024 * 1024;
}
/**
* Returns the email address of the admin account on the G Suite app used to perform
* administrative actions.
core/src/main/java/google/registry/export/ExportDomainListsAction.java
+11 -15
View File
@@ -14,7 +14,6 @@
package google.registry.export;
import static com.google.appengine.tools.cloudstorage.GcsServiceFactory.createGcsService;
import static com.google.common.base.Verify.verifyNotNull;
import static google.registry.mapreduce.inputs.EppResourceInputs.createEntityInput;
import static google.registry.model.EppResourceUtils.isActive;
@@ -24,11 +23,10 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
import static google.registry.request.Action.Method.POST;
import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.appengine.tools.cloudstorage.RetryParams;
import com.google.appengine.tools.mapreduce.Mapper;
import com.google.appengine.tools.mapreduce.Reducer;
import com.google.appengine.tools.mapreduce.ReducerInput;
import com.google.cloud.storage.BlobId;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Suppliers;
@@ -79,9 +77,9 @@ public class ExportDomainListsAction implements Runnable {
@Inject Response response;
@Inject Clock clock;
@Inject DriveConnection driveConnection;
@Inject GcsUtils gcsUtils;
@Inject @Config("domainListsGcsBucket") String gcsBucket;
@Inject @Config("gcsBufferSize") int gcsBufferSize;
@Inject ExportDomainListsAction() {}
@Override
@@ -95,7 +93,7 @@ public class ExportDomainListsAction implements Runnable {
.setDefaultReduceShards(Math.min(realTlds.size(), MAX_NUM_REDUCE_SHARDS))
.runMapreduce(
new ExportDomainListsMapper(clock.nowUtc(), realTlds),
new ExportDomainListsReducer(gcsBucket, gcsBufferSize),
new ExportDomainListsReducer(gcsBucket, gcsUtils),
ImmutableList.of(createEntityInput(DomainBase.class)))
.sendLinkToMapreduceConsole(response);
} else {
@@ -134,7 +132,7 @@ public class ExportDomainListsAction implements Runnable {
String domainsList = Joiner.on("\n").join(domains);
logger.atInfo().log(
"Exporting %d domains for TLD %s to GCS and Drive.", domains.size(), tld);
exportToGcs(tld, domainsList, gcsBucket, gcsBufferSize);
exportToGcs(tld, domainsList, gcsBucket, gcsUtils);
exportToDrive(tld, domainsList, driveConnection);
});
}
@@ -168,11 +166,9 @@ public class ExportDomainListsAction implements Runnable {
}
protected static boolean exportToGcs(
String tld, String domains, String gcsBucket, int gcsBufferSize) {
GcsFilename filename = new GcsFilename(gcsBucket, tld + ".txt");
GcsUtils cloudStorage =
new GcsUtils(createGcsService(RetryParams.getDefaultInstance()), gcsBufferSize);
try (OutputStream gcsOutput = cloudStorage.openOutputStream(filename);
String tld, String domains, String gcsBucket, GcsUtils gcsUtils) {
BlobId blobId = BlobId.of(gcsBucket, tld + ".txt");
try (OutputStream gcsOutput = gcsUtils.openOutputStream(blobId);
Writer osWriter = new OutputStreamWriter(gcsOutput, UTF_8)) {
osWriter.write(domains);
} catch (Throwable e) {
@@ -214,7 +210,7 @@ public class ExportDomainListsAction implements Runnable {
Suppliers.memoize(() -> DaggerDriveModule_DriveComponent.create().driveConnection());
private final String gcsBucket;
private final int gcsBufferSize;
private final GcsUtils gcsUtils;
/**
* Non-serializable {@link DriveConnection} that will be created when an instance of {@link
@@ -224,9 +220,9 @@ public class ExportDomainListsAction implements Runnable {
*/
private transient DriveConnection driveConnection;
public ExportDomainListsReducer(String gcsBucket, int gcsBufferSize) {
public ExportDomainListsReducer(String gcsBucket, GcsUtils gcsUtils) {
this.gcsBucket = gcsBucket;
this.gcsBufferSize = gcsBufferSize;
this.gcsUtils = gcsUtils;
}
@SuppressWarnings("unused")
@@ -240,7 +236,7 @@ public class ExportDomainListsAction implements Runnable {
ImmutableList<String> domains = ImmutableList.sortedCopyOf(() -> fqdns);
String domainsList = Joiner.on('\n').join(domains);
logger.atInfo().log("Exporting %d domains for TLD %s to GCS and Drive.", domains.size(), tld);
if (exportToGcs(tld, domainsList, gcsBucket, gcsBufferSize)) {
if (exportToGcs(tld, domainsList, gcsBucket, gcsUtils)) {
getContext().incrementCounter("domain lists successful written out to GCS");
} else {
getContext().incrementCounter("domain lists failed to write out to GCS");
core/src/main/java/google/registry/gcs/GcsServiceModule.java
-34
View File
@@ -1,34 +0,0 @@
// Copyright 2017 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.gcs;
import com.google.appengine.tools.cloudstorage.GcsService;
import com.google.appengine.tools.cloudstorage.GcsServiceFactory;
import com.google.appengine.tools.cloudstorage.RetryParams;
import dagger.Module;
import dagger.Provides;
/** Dagger module for {@link GcsService}. */
@Module
public final class GcsServiceModule {
private static final GcsService gcsService =
GcsServiceFactory.createGcsService(RetryParams.getDefaultInstance());
@Provides
static GcsService provideGcsService() {
return gcsService;
}
}
core/src/main/java/google/registry/gcs/GcsUtils.java
+105 -43
View File
@@ -14,30 +14,39 @@
package google.registry.gcs;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.Iterables.getLast;
import com.google.appengine.tools.cloudstorage.GcsFileMetadata;
import com.google.appengine.tools.cloudstorage.GcsFileOptions;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.appengine.tools.cloudstorage.GcsService;
import com.google.appengine.tools.cloudstorage.ListOptions;
import com.google.appengine.tools.cloudstorage.ListResult;
import com.google.cloud.storage.Blob;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.Storage.BlobListOption;
import com.google.cloud.storage.StorageException;
import com.google.cloud.storage.StorageOptions;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Streams;
import com.google.common.flogger.FluentLogger;
import com.google.common.net.MediaType;
import google.registry.config.RegistryConfig.Config;
import google.registry.config.CredentialModule.DefaultCredential;
import google.registry.util.GoogleCredentialsBundle;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.io.Serializable;
import java.nio.channels.Channels;
import javax.annotation.CheckReturnValue;
import javax.inject.Inject;
/** Utilities for working with Google Cloud Storage. */
public class GcsUtils {
/**
* Utilities for working with Google Cloud Storage.
*
* <p>It is {@link Serializable} so that it can be used in MapReduce or Beam.
*/
public class GcsUtils implements Serializable {
private static final FluentLogger logger = FluentLogger.forEnclosingClass();
@@ -48,30 +57,65 @@ public class GcsUtils {
.put("json", MediaType.JSON_UTF_8)
.build();
private final GcsService gcsService;
private final int bufferSize;
private final StorageOptions storageOptions;
@Inject
public GcsUtils(GcsService gcsService, @Config("gcsBufferSize") int bufferSize) {
this.gcsService = gcsService;
this.bufferSize = bufferSize;
private Storage storage() {
return storageOptions.getService();
}
/** Opens a GCS file for reading as an {@link InputStream} with prefetching. */
@Inject
public GcsUtils(@DefaultCredential GoogleCredentialsBundle credentialsBundle) {
this(
StorageOptions.newBuilder()
.setCredentials(credentialsBundle.getGoogleCredentials())
.build());
}
@VisibleForTesting
public GcsUtils(StorageOptions storageOptions) {
this.storageOptions = storageOptions;
}
/** Opens a GCS file for reading as an {@link InputStream}. */
@CheckReturnValue
public InputStream openInputStream(GcsFilename filename) {
return Channels.newInputStream(gcsService.openPrefetchingReadChannel(filename, 0, bufferSize));
public InputStream openInputStream(BlobId blobId) {
return Channels.newInputStream(storage().reader(blobId));
}
/** Opens a GCS file for writing as an {@link OutputStream}, overwriting existing files. */
@CheckReturnValue
public OutputStream openOutputStream(GcsFilename filename) throws IOException {
return Channels.newOutputStream(gcsService.createOrReplace(filename, getOptions(filename)));
public OutputStream openOutputStream(BlobId blobId) {
return Channels.newOutputStream(storage().writer(createBlobInfo(blobId)));
}
/**
* Opens a GCS file for writing as an {@link OutputStream}, overwriting existing files and setting
* the given metadata.
*/
@CheckReturnValue
public OutputStream openOutputStream(BlobId blobId, ImmutableMap<String, String> metadata) {
return Channels.newOutputStream(
storage().writer(BlobInfo.newBuilder(blobId).setMetadata(metadata).build()));
}
/** Creates a GCS file with the given byte contents, overwriting existing files. */
public void createFromBytes(GcsFilename filename, byte[] bytes) throws IOException {
gcsService.createOrReplace(filename, getOptions(filename), ByteBuffer.wrap(bytes));
public void createFromBytes(BlobId blobId, byte[] bytes) throws StorageException {
createFromBytes(createBlobInfo(blobId), bytes);
}
/** Creates a GCS file with the given byte contents and metadata, overwriting existing files. */
public void createFromBytes(BlobInfo blobInfo, byte[] bytes) throws StorageException {
storage().create(blobInfo, bytes);
}
/** Read the content of the given GCS file and return it in a byte array. */
public byte[] readBytesFrom(BlobId blobId) throws StorageException {
return storage().readAllBytes(blobId);
}
/** Delete the given GCS file. */
public void delete(BlobId blobId) throws StorageException {
storage().delete(blobId);
}
/**
@@ -86,37 +130,55 @@ public class GcsUtils {
*/
public ImmutableList<String> listFolderObjects(String bucketName, String prefix)
throws IOException {
ListResult result =
gcsService.list(bucketName, new ListOptions.Builder().setPrefix(prefix).build());
final ImmutableList.Builder<String> builder = new ImmutableList.Builder<>();
result.forEachRemaining(
listItem -> {
if (!listItem.isDirectory()) {
builder.add(listItem.getName().replaceFirst(prefix, ""));
}
});
return builder.build();
return Streams.stream(storage().list(bucketName, BlobListOption.prefix(prefix)).iterateAll())
.map(blob -> blob.getName().substring(prefix.length()))
.collect(toImmutableList());
}
/** Returns {@code true} if a file exists and is non-empty on Google Cloud Storage. */
public boolean existsAndNotEmpty(GcsFilename file) {
GcsFileMetadata metadata;
public boolean existsAndNotEmpty(BlobId blobId) {
try {
metadata = gcsService.getMetadata(file);
} catch (IOException e) {
Blob blob = storage().get(blobId);
return blob != null && blob.getSize() > 0;
} catch (StorageException e) {
logger.atWarning().withCause(e).log("Failed to check if GCS file exists");
return false;
}
return metadata != null && metadata.getLength() > 0;
}
/** Determines most appropriate {@link GcsFileOptions} based on filename extension. */
private static GcsFileOptions getOptions(GcsFilename filename) {
GcsFileOptions.Builder builder = new GcsFileOptions.Builder().cacheControl("no-cache");
MediaType mediaType = EXTENSIONS.get(getLast(Splitter.on('.').split(filename.getObjectName())));
/** Returns the user defined metadata of a GCS file if the file exists, or an empty map. */
public ImmutableMap<String, String> getMetadata(BlobId blobId) throws StorageException {
Blob blob = storage().get(blobId);
return blob == null ? ImmutableMap.of() : ImmutableMap.copyOf(blob.getMetadata());
}
/**
* Returns the {@link BlobInfo} of the given GCS file.
*
* <p>Note that a {@link Blob} is returned, but on the {@link BlobInfo} part of it is usable.
*/
public BlobInfo getBlobInfo(BlobId blobId) throws StorageException {
return storage().get(blobId);
}
/** Determines most appropriate {@link BlobInfo} based on filename extension. */
private static BlobInfo createBlobInfo(BlobId blobId) {
BlobInfo.Builder builder = BlobInfo.newBuilder(blobId).setCacheControl("no-cache");
MediaType mediaType = EXTENSIONS.get(getLast(Splitter.on('.').split(blobId.getName())));
if (mediaType != null) {
builder = builder.mimeType(mediaType.type());
builder = builder.setContentType(mediaType.toString());
}
return builder.build();
}
// These two methods are needed to check whether serialization is done correctly in tests.
@Override
public boolean equals(Object obj) {
return obj instanceof GcsUtils && ((GcsUtils) obj).storageOptions.equals(storageOptions);
}
@Override
public int hashCode() {
return storageOptions.hashCode();
}
}
core/src/main/java/google/registry/module/backend/BackendComponent.java
-2
View File
@@ -26,7 +26,6 @@ import google.registry.export.datastore.DatastoreAdminModule;
import google.registry.export.sheet.SheetsServiceModule;
import google.registry.flows.ServerTridProviderModule;
import google.registry.flows.custom.CustomLogicFactoryModule;
import google.registry.gcs.GcsServiceModule;
import google.registry.groups.DirectoryModule;
import google.registry.groups.GroupsModule;
import google.registry.groups.GroupssettingsModule;
@@ -64,7 +63,6 @@ import javax.inject.Singleton;
DirectoryModule.class,
DummyKeyringModule.class,
DriveModule.class,
GcsServiceModule.class,
GroupsModule.class,
GroupssettingsModule.class,
JSchModule.class,
core/src/main/java/google/registry/module/tools/ToolsComponent.java
-2
View File
@@ -22,7 +22,6 @@ import google.registry.config.RegistryConfig.ConfigModule;
import google.registry.export.DriveModule;
import google.registry.flows.ServerTridProviderModule;
import google.registry.flows.custom.CustomLogicFactoryModule;
import google.registry.gcs.GcsServiceModule;
import google.registry.groups.DirectoryModule;
import google.registry.groups.GroupsModule;
import google.registry.groups.GroupssettingsModule;
@@ -54,7 +53,6 @@ import javax.inject.Singleton;
DirectoryModule.class,
DummyKeyringModule.class,
DriveModule.class,
GcsServiceModule.class,
GroupsModule.class,
GroupssettingsModule.class,
Jackson2Module.class,
core/src/main/java/google/registry/rde/BrdaCopyAction.java
+6 -6
View File
@@ -17,7 +17,7 @@ package google.registry.rde;
import static google.registry.model.rde.RdeMode.THIN;
import static google.registry.request.Action.Method.POST;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.flogger.FluentLogger;
import com.google.common.io.ByteStreams;
import google.registry.config.RegistryConfig.Config;
@@ -85,10 +85,10 @@ public final class BrdaCopyAction implements Runnable {
private void copyAsRyde() throws IOException {
String prefix = RdeNamingUtils.makeRydeFilename(tld, watermark, THIN, 1, 0);
GcsFilename xmlFilename = new GcsFilename(stagingBucket, prefix + ".xml.ghostryde");
GcsFilename xmlLengthFilename = new GcsFilename(stagingBucket, prefix + ".xml.length");
GcsFilename rydeFile = new GcsFilename(brdaBucket, prefix + ".ryde");
GcsFilename sigFile = new GcsFilename(brdaBucket, prefix + ".sig");
BlobId xmlFilename = BlobId.of(stagingBucket, prefix + ".xml.ghostryde");
BlobId xmlLengthFilename = BlobId.of(stagingBucket, prefix + ".xml.length");
BlobId rydeFile = BlobId.of(brdaBucket, prefix + ".ryde");
BlobId sigFile = BlobId.of(brdaBucket, prefix + ".sig");
long xmlLength = readXmlLength(xmlLengthFilename);
@@ -107,7 +107,7 @@ public final class BrdaCopyAction implements Runnable {
}
/** Reads the contents of a file from Cloud Storage that contains nothing but an integer. */
private long readXmlLength(GcsFilename xmlLengthFilename) throws IOException {
private long readXmlLength(BlobId xmlLengthFilename) throws IOException {
try (InputStream input = gcsUtils.openInputStream(xmlLengthFilename)) {
return Ghostryde.readLength(input);
}
core/src/main/java/google/registry/rde/RdeReportAction.java
+3 -3
View File
@@ -23,7 +23,7 @@ import static google.registry.persistence.transaction.TransactionManagerUtil.tra
import static google.registry.request.Action.Method.POST;
import static google.registry.util.DateTimeUtils.isBeforeOrAt;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.flogger.FluentLogger;
import com.google.common.io.ByteStreams;
import google.registry.config.RegistryConfig.Config;
@@ -97,7 +97,7 @@ public final class RdeReportAction implements Runnable, EscrowTask {
.orElseThrow(
() -> new IllegalStateException("RdeRevision was not set on generated deposit"));
String prefix = RdeNamingUtils.makeRydeFilename(tld, watermark, FULL, 1, revision);
GcsFilename reportFilename = new GcsFilename(bucket, prefix + "-report.xml.ghostryde");
BlobId reportFilename = BlobId.of(bucket, prefix + "-report.xml.ghostryde");
verify(gcsUtils.existsAndNotEmpty(reportFilename), "Missing file: %s", reportFilename);
reporter.send(readReportFromGcs(reportFilename));
response.setContentType(PLAIN_TEXT_UTF_8);
@@ -106,7 +106,7 @@ public final class RdeReportAction implements Runnable, EscrowTask {
}
/** Reads and decrypts the XML file from cloud storage. */
private byte[] readReportFromGcs(GcsFilename reportFilename) throws IOException {
private byte[] readReportFromGcs(BlobId reportFilename) throws IOException {
try (InputStream gcsInput = gcsUtils.openInputStream(reportFilename);
InputStream ghostrydeDecoder = Ghostryde.decoder(gcsInput, stagingDecryptionKey)) {
return ByteStreams.toByteArray(ghostrydeDecoder);
core/src/main/java/google/registry/rde/RdeStagingAction.java
+3 -1
View File
@@ -27,6 +27,7 @@ import com.google.common.collect.ImmutableSetMultimap;
import com.google.common.collect.Multimaps;
import com.google.common.flogger.FluentLogger;
import google.registry.config.RegistryConfig.Config;
import google.registry.gcs.GcsUtils;
import google.registry.mapreduce.MapreduceRunner;
import google.registry.mapreduce.inputs.EppResourceInputs;
import google.registry.mapreduce.inputs.NullInput;
@@ -206,6 +207,7 @@ public final class RdeStagingAction implements Runnable {
@Inject PendingDepositChecker pendingDepositChecker;
@Inject RdeStagingReducer.Factory reducerFactory;
@Inject Response response;
@Inject GcsUtils gcsUtils;
@Inject MapreduceRunner mrRunner;
@Inject @Config("transactionCooldown") Duration transactionCooldown;
@Inject @Parameter(RdeModule.PARAM_MANUAL) boolean manual;
@@ -234,7 +236,7 @@ public final class RdeStagingAction implements Runnable {
}
ValidationMode validationMode = lenient ? LENIENT : STRICT;
RdeStagingMapper mapper = new RdeStagingMapper(validationMode, pendings);
RdeStagingReducer reducer = reducerFactory.create(validationMode);
RdeStagingReducer reducer = reducerFactory.create(validationMode, gcsUtils);
mrRunner
.setJobName("Stage escrow deposits for all TLDs")
core/src/main/java/google/registry/rde/RdeStagingReducer.java
+14 -19
View File
@@ -16,7 +16,6 @@ package google.registry.rde;
import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
import static com.google.appengine.api.taskqueue.TaskOptions.Builder.withUrl;
import static com.google.appengine.tools.cloudstorage.GcsServiceFactory.createGcsService;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.base.Verify.verify;
import static google.registry.model.common.Cursor.getCursorTimeOrStartOfTime;
@@ -24,10 +23,9 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.appengine.tools.cloudstorage.RetryParams;
import com.google.appengine.tools.mapreduce.Reducer;
import com.google.appengine.tools.mapreduce.ReducerInput;
import com.google.cloud.storage.BlobId;
import com.google.common.flogger.FluentLogger;
import google.registry.config.RegistryConfig.Config;
import google.registry.gcs.GcsUtils;
@@ -69,27 +67,27 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
private final TaskQueueUtils taskQueueUtils;
private final LockHandler lockHandler;
private final int gcsBufferSize;
private final String bucket;
private final Duration lockTimeout;
private final byte[] stagingKeyBytes;
private final RdeMarshaller marshaller;
private final GcsUtils gcsUtils;
RdeStagingReducer(
TaskQueueUtils taskQueueUtils,
LockHandler lockHandler,
int gcsBufferSize,
String bucket,
Duration lockTimeout,
byte[] stagingKeyBytes,
ValidationMode validationMode) {
ValidationMode validationMode,
GcsUtils gcsUtils) {
this.taskQueueUtils = taskQueueUtils;
this.lockHandler = lockHandler;
this.gcsBufferSize = gcsBufferSize;
this.bucket = bucket;
this.lockTimeout = lockTimeout;
this.stagingKeyBytes = stagingKeyBytes;
this.marshaller = new RdeMarshaller(validationMode);
this.gcsUtils = gcsUtils;
}
@Override
@@ -113,8 +111,6 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
// Construct things that Dagger would inject if this wasn't serialized.
PGPPublicKey stagingKey = PgpHelper.loadPublicKeyBytes(stagingKeyBytes);
GcsUtils cloudStorage =
new GcsUtils(createGcsService(RetryParams.getDefaultInstance()), gcsBufferSize);
RdeCounter counter = new RdeCounter();
// Determine some basic things about the deposit.
@@ -130,14 +126,14 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
checkState(key.directoryWithTrailingSlash() != null, "Manual subdirectory not specified");
prefix = "manual/" + key.directoryWithTrailingSlash() + prefix;
}
GcsFilename xmlFilename = new GcsFilename(bucket, prefix + ".xml.ghostryde");
BlobId xmlFilename = BlobId.of(bucket, prefix + ".xml.ghostryde");
// This file will contain the byte length (ASCII) of the raw unencrypted XML.
//
// This is necessary because RdeUploadAction creates a tar file which requires that the length
// be outputted. We don't want to have to decrypt the entire ghostryde file to determine the
// length, so we just save it separately.
GcsFilename xmlLengthFilename = new GcsFilename(bucket, prefix + ".xml.length");
GcsFilename reportFilename = new GcsFilename(bucket, prefix + "-report.xml.ghostryde");
BlobId xmlLengthFilename = BlobId.of(bucket, prefix + ".xml.length");
BlobId reportFilename = BlobId.of(bucket, prefix + "-report.xml.ghostryde");
// These variables will be populated as we write the deposit XML and used for other files.
boolean failed = false;
@@ -146,8 +142,8 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
// Write a gigantic XML file to GCS. We'll start by opening encrypted out/err file handles.
logger.atInfo().log("Writing %s and %s", xmlFilename, xmlLengthFilename);
try (OutputStream gcsOutput = cloudStorage.openOutputStream(xmlFilename);
OutputStream lengthOutput = cloudStorage.openOutputStream(xmlLengthFilename);
try (OutputStream gcsOutput = gcsUtils.openOutputStream(xmlFilename);
OutputStream lengthOutput = gcsUtils.openOutputStream(xmlLengthFilename);
OutputStream ghostrydeEncoder = Ghostryde.encoder(gcsOutput, stagingKey, lengthOutput);
Writer output = new OutputStreamWriter(ghostrydeEncoder, UTF_8)) {
@@ -194,7 +190,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
// This will be sent to ICANN once we're done uploading the big XML to the escrow provider.
if (mode == RdeMode.FULL) {
logger.atInfo().log("Writing %s", reportFilename);
try (OutputStream gcsOutput = cloudStorage.openOutputStream(reportFilename);
try (OutputStream gcsOutput = gcsUtils.openOutputStream(reportFilename);
OutputStream ghostrydeEncoder = Ghostryde.encoder(gcsOutput, stagingKey)) {
counter.makeReport(id, watermark, header, revision).marshal(ghostrydeEncoder, UTF_8);
} catch (IOException | XmlException e) {
@@ -249,22 +245,21 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
static class Factory {
@Inject TaskQueueUtils taskQueueUtils;
@Inject LockHandler lockHandler;
@Inject @Config("gcsBufferSize") int gcsBufferSize;
@Inject @Config("rdeBucket") String bucket;
@Inject @Config("rdeStagingLockTimeout") Duration lockTimeout;
@Inject @KeyModule.Key("rdeStagingEncryptionKey") byte[] stagingKeyBytes;
@Inject Factory() {}
RdeStagingReducer create(ValidationMode validationMode) {
RdeStagingReducer create(ValidationMode validationMode, GcsUtils gcsUtils) {
return new RdeStagingReducer(
taskQueueUtils,
lockHandler,
gcsBufferSize,
bucket,
lockTimeout,
stagingKeyBytes,
validationMode);
validationMode,
gcsUtils);
}
}
}
core/src/main/java/google/registry/rde/RdeUploadAction.java
+25 -24
View File
@@ -30,7 +30,7 @@ import static google.registry.util.DateTimeUtils.isBeforeOrAt;
import static java.util.Arrays.asList;
import com.google.appengine.api.taskqueue.Queue;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.flogger.FluentLogger;
import com.google.common.io.ByteStreams;
@@ -165,9 +165,9 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
.orElseThrow(
() -> new IllegalStateException("RdeRevision was not set on generated deposit"));
final String name = RdeNamingUtils.makeRydeFilename(tld, watermark, FULL, 1, revision);
final GcsFilename xmlFilename = new GcsFilename(bucket, name + ".xml.ghostryde");
final GcsFilename xmlLengthFilename = new GcsFilename(bucket, name + ".xml.length");
GcsFilename reportFilename = new GcsFilename(bucket, name + "-report.xml.ghostryde");
final BlobId xmlFilename = BlobId.of(bucket, name + ".xml.ghostryde");
final BlobId xmlLengthFilename = BlobId.of(bucket, name + ".xml.length");
BlobId reportFilename = BlobId.of(bucket, name + "-report.xml.ghostryde");
verifyFileExists(xmlFilename);
verifyFileExists(xmlLengthFilename);
verifyFileExists(reportFilename);
@@ -187,29 +187,30 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
}
/**
* Performs a blocking upload of a cloud storage XML file to escrow provider, converting
* it to the RyDE format along the way by applying tar+compress+encrypt+sign, and saving the
* created RyDE file on GCS for future reference.
* Performs a blocking upload of a cloud storage XML file to escrow provider, converting it to the
* RyDE format along the way by applying tar+compress+encrypt+sign, and saving the created RyDE
* file on GCS for future reference.
*
* <p>This is done by layering a bunch of {@link java.io.FilterOutputStream FilterOutputStreams}
* on top of each other in reverse order that turn XML bytes into a RyDE file while
* simultaneously uploading it to the SFTP endpoint, and then using {@link ByteStreams#copy} to
* blocking-copy bytes from the cloud storage {@code InputStream} to the RyDE/SFTP pipeline.
* on top of each other in reverse order that turn XML bytes into a RyDE file while simultaneously
* uploading it to the SFTP endpoint, and then using {@link ByteStreams#copy} to blocking-copy
* bytes from the cloud storage {@code InputStream} to the RyDE/SFTP pipeline.
*
* <p>In pseudo-shell, the whole process looks like the following:
*
* <pre> {@code
* gcs read $xmlFile \ # Get GhostRyDE from cloud storage.
* | decrypt | decompress \ # Convert it to XML.
* | tar | file | compress | encrypt | sign /tmp/sig \ # Convert it to a RyDE file.
* | tee gs://bucket/$rydeFilename.ryde \ # Save a copy of the RyDE file to GCS.
* | sftp put $dstUrl/$rydeFilename.ryde \ # Upload to SFTP server.
* && sftp put $dstUrl/$rydeFilename.sig </tmp/sig \ # Upload detached signature.
* && cat /tmp/sig > gs://bucket/$rydeFilename.sig # Save a copy of signature to GCS.
* }</pre>
* <pre>{@code
* gcs read $xmlFile \ # Get GhostRyDE from cloud storage.
* | decrypt | decompress \ # Convert it to XML.
* | tar | file | compress | encrypt | sign /tmp/sig \ # Convert it to a RyDE file.
* | tee gs://bucket/$rydeFilename.ryde \ # Save a copy of the RyDE file to GCS.
* | sftp put $dstUrl/$rydeFilename.ryde \ # Upload to SFTP server.
* && sftp put $dstUrl/$rydeFilename.sig </tmp/sig \ # Upload detached signature.
* && cat /tmp/sig > gs://bucket/$rydeFilename.sig # Save a copy of signature to GCS.
*
* }</pre>
*/
@VisibleForTesting
protected void upload(GcsFilename xmlFile, long xmlLength, DateTime watermark, String name)
protected void upload(BlobId xmlFile, long xmlLength, DateTime watermark, String name)
throws Exception {
logger.atInfo().log("Uploading XML file '%s' to remote path '%s'.", xmlFile, uploadUrl);
try (InputStream gcsInput = gcsUtils.openInputStream(xmlFile);
@@ -218,7 +219,7 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
JSchSftpChannel ftpChan = session.openSftpChannel()) {
ByteArrayOutputStream sigOut = new ByteArrayOutputStream();
String rydeFilename = name + ".ryde";
GcsFilename rydeGcsFilename = new GcsFilename(bucket, rydeFilename);
BlobId rydeGcsFilename = BlobId.of(bucket, rydeFilename);
try (OutputStream ftpOutput =
ftpChan.get().put(rydeFilename, sftpProgressMonitor, OVERWRITE);
OutputStream gcsOutput = gcsUtils.openOutputStream(rydeGcsFilename);
@@ -234,7 +235,7 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
}
String sigFilename = name + ".sig";
byte[] signature = sigOut.toByteArray();
gcsUtils.createFromBytes(new GcsFilename(bucket, sigFilename), signature);
gcsUtils.createFromBytes(BlobId.of(bucket, sigFilename), signature);
ftpChan.get().put(new ByteArrayInputStream(signature), sigFilename);
logger.atInfo().log("uploaded %,d bytes: %s", signature.length, sigFilename);
}
@@ -242,13 +243,13 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
}
/** Reads the contents of a file from Cloud Storage that contains nothing but an integer. */
private long readXmlLength(GcsFilename xmlLengthFilename) throws IOException {
private long readXmlLength(BlobId xmlLengthFilename) throws IOException {
try (InputStream input = gcsUtils.openInputStream(xmlLengthFilename)) {
return Ghostryde.readLength(input);
}
}
private void verifyFileExists(GcsFilename filename) {
private void verifyFileExists(BlobId filename) {
verify(gcsUtils.existsAndNotEmpty(filename), "Missing file: %s", filename);
}
}
core/src/main/java/google/registry/reporting/billing/BillingEmailUtils.java
+2 -3
View File
@@ -17,7 +17,7 @@ package google.registry.reporting.billing;
import static com.google.common.base.Throwables.getRootCause;
import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.collect.ImmutableList;
import com.google.common.io.CharStreams;
import com.google.common.net.MediaType;
@@ -72,8 +72,7 @@ public class BillingEmailUtils {
void emailOverallInvoice() {
try {
String invoiceFile = String.format("%s-%s.csv", invoiceFilePrefix, yearMonth);
GcsFilename invoiceFilename =
new GcsFilename(billingBucket, invoiceDirectoryPrefix + invoiceFile);
BlobId invoiceFilename = BlobId.of(billingBucket, invoiceDirectoryPrefix + invoiceFile);
try (InputStream in = gcsUtils.openInputStream(invoiceFilename)) {
emailService.sendEmail(
EmailMessage.newBuilder()
core/src/main/java/google/registry/reporting/billing/CopyDetailReportsAction.java
+2 -2
View File
@@ -19,7 +19,7 @@ import static google.registry.request.Action.Method.POST;
import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
import static javax.servlet.http.HttpServletResponse.SC_OK;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
@@ -120,7 +120,7 @@ public final class CopyDetailReportsAction implements Runnable {
() -> {
try (InputStream input =
gcsUtils.openInputStream(
new GcsFilename(billingBucket, invoiceDirectoryPrefix + detailReportName))) {
BlobId.of(billingBucket, invoiceDirectoryPrefix + detailReportName))) {
driveConnection.createOrUpdateFile(
detailReportName,
MediaType.CSV_UTF_8,
core/src/main/java/google/registry/reporting/icann/IcannReportingStager.java
+3 -3
View File
@@ -21,7 +21,7 @@ import static google.registry.reporting.icann.IcannReportingModule.MANIFEST_FILE
import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.base.Ascii;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableCollection;
@@ -260,7 +260,7 @@ public class IcannReportingStager {
Ascii.toLowerCase(reportType.toString()),
DateTimeFormat.forPattern("yyyyMM").print(yearMonth));
String reportBucketname = String.format("%s/%s", reportingBucket, subdir);
final GcsFilename gcsFilename = new GcsFilename(reportBucketname, reportFilename);
final BlobId gcsFilename = BlobId.of(reportBucketname, reportFilename);
gcsUtils.createFromBytes(gcsFilename, reportBytes);
logger.atInfo().log("Wrote %d bytes to file location %s", reportBytes.length, gcsFilename);
return reportFilename;
@@ -269,7 +269,7 @@ public class IcannReportingStager {
/** Creates and stores a manifest file on GCS, indicating which reports were generated. */
void createAndUploadManifest(String subdir, ImmutableList<String> filenames) throws IOException {
String reportBucketname = String.format("%s/%s", reportingBucket, subdir);
final GcsFilename gcsFilename = new GcsFilename(reportBucketname, MANIFEST_FILE_NAME);
final BlobId gcsFilename = BlobId.of(reportBucketname, MANIFEST_FILE_NAME);
StringBuilder manifestString = new StringBuilder();
filenames.forEach((filename) -> manifestString.append(filename).append("\n"));
gcsUtils.createFromBytes(gcsFilename, manifestString.toString().getBytes(UTF_8));
core/src/main/java/google/registry/reporting/icann/IcannReportingUploadAction.java
+6 -6
View File
@@ -20,7 +20,7 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
import static google.registry.request.Action.Method.POST;
import static javax.servlet.http.HttpServletResponse.SC_OK;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
@@ -147,7 +147,7 @@ public final class IcannReportingUploadAction implements Runnable {
cursorTimeMinusMonth.getYear(), cursorTimeMinusMonth.getMonthOfYear());
String reportBucketname = String.format("%s/%s", reportingBucket, reportSubdir);
String filename = getFileName(cursorType, cursorTime, tldStr);
final GcsFilename gcsFilename = new GcsFilename(reportBucketname, filename);
final BlobId gcsFilename = BlobId.of(reportBucketname, filename);
logger.atInfo().log("Reading ICANN report %s from bucket %s", filename, reportBucketname);
// Check that the report exists
try {
@@ -298,18 +298,18 @@ public final class IcannReportingUploadAction implements Runnable {
emailService.sendEmail(EmailMessage.create(subject, body, recipient, sender));
}
private byte[] readBytesFromGcs(GcsFilename reportFilename) throws IOException {
private byte[] readBytesFromGcs(BlobId reportFilename) throws IOException {
try (InputStream gcsInput = gcsUtils.openInputStream(reportFilename)) {
return ByteStreams.toByteArray(gcsInput);
}
}
private void verifyFileExists(GcsFilename gcsFilename) {
private void verifyFileExists(BlobId gcsFilename) {
checkArgument(
gcsUtils.existsAndNotEmpty(gcsFilename),
"Object %s in bucket %s not found",
gcsFilename.getObjectName(),
gcsFilename.getBucketName());
gcsFilename.getName(),
gcsFilename.getBucket());
}
}
core/src/main/java/google/registry/reporting/spec11/Spec11RegistrarThreatMatchesParser.java
+5 -5
View File
@@ -17,7 +17,7 @@ package google.registry.reporting.spec11;
import static com.google.common.collect.ImmutableSet.toImmutableSet;
import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.cloud.storage.BlobId;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
@@ -62,7 +62,7 @@ public class Spec11RegistrarThreatMatchesParser {
}
/** Returns registrar:set-of-threat-match pairings from the file, or empty if it doesn't exist. */
public ImmutableSet<RegistrarThreatMatches> getFromFile(GcsFilename spec11ReportFilename)
public ImmutableSet<RegistrarThreatMatches> getFromFile(BlobId spec11ReportFilename)
throws IOException {
if (!gcsUtils.existsAndNotEmpty(spec11ReportFilename)) {
return ImmutableSet.of();
@@ -81,7 +81,7 @@ public class Spec11RegistrarThreatMatchesParser {
public Optional<LocalDate> getPreviousDateWithMatches(LocalDate date) {
LocalDate yesterday = date.minusDays(1);
GcsFilename gcsFilename = getGcsFilename(yesterday);
BlobId gcsFilename = getGcsFilename(yesterday);
if (gcsUtils.existsAndNotEmpty(gcsFilename)) {
return Optional.of(yesterday);
}
@@ -98,8 +98,8 @@ public class Spec11RegistrarThreatMatchesParser {
return Optional.empty();
}
private GcsFilename getGcsFilename(LocalDate localDate) {
return new GcsFilename(reportingBucket, Spec11Pipeline.getSpec11ReportFilePath(localDate));
private BlobId getGcsFilename(LocalDate localDate) {
return BlobId.of(reportingBucket, Spec11Pipeline.getSpec11ReportFilePath(localDate));
}
private RegistrarThreatMatches parseRegistrarThreatMatch(String line) throws JSONException {
core/src/main/java/google/registry/tools/server/GenerateZoneFilesAction.java
+9 -13
View File
@@ -14,7 +14,6 @@
package google.registry.tools.server;
import static com.google.appengine.tools.cloudstorage.GcsServiceFactory.createGcsService;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.Iterators.filter;
import static com.google.common.io.BaseEncoding.base16;
@@ -25,11 +24,10 @@ import static google.registry.request.Action.Method.POST;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.joda.time.DateTimeZone.UTC;
import com.google.appengine.tools.cloudstorage.GcsFilename;
import com.google.appengine.tools.cloudstorage.RetryParams;
import com.google.appengine.tools.mapreduce.Mapper;
import com.google.appengine.tools.mapreduce.Reducer;
import com.google.appengine.tools.mapreduce.ReducerInput;
import com.google.cloud.storage.BlobId;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
@@ -98,13 +96,14 @@ public class GenerateZoneFilesAction implements Runnable, JsonActionRunner.JsonA
@Inject MapreduceRunner mrRunner;
@Inject JsonActionRunner jsonActionRunner;
@Inject @Config("zoneFilesBucket") String bucket;
@Inject @Config("gcsBufferSize") int gcsBufferSize;
@Inject @Config("commitLogDatastoreRetention") Duration datastoreRetention;
@Inject @Config("dnsDefaultATtl") Duration dnsDefaultATtl;
@SuppressWarnings("DurationVariableWithUnits") // false-positive Error Prone check
@Inject @Config("dnsDefaultNsTtl") Duration dnsDefaultNsTtl;
@Inject @Config("dnsDefaultDsTtl") Duration dnsDefaultDsTtl;
@Inject Clock clock;
@Inject GcsUtils gcsUtils;
@Inject GenerateZoneFilesAction() {}
@Override
@@ -140,7 +139,7 @@ public class GenerateZoneFilesAction implements Runnable, JsonActionRunner.JsonA
.runMapreduce(
new GenerateBindFileMapper(
tlds, exportTime, dnsDefaultATtl, dnsDefaultNsTtl, dnsDefaultDsTtl),
new GenerateBindFileReducer(bucket, exportTime, gcsBufferSize),
new GenerateBindFileReducer(bucket, exportTime, gcsUtils),
ImmutableList.of(new NullInput<>(), createEntityInput(DomainBase.class)))
.getLinkToMapreduceConsole();
ImmutableList<String> filenames =
@@ -236,22 +235,19 @@ public class GenerateZoneFilesAction implements Runnable, JsonActionRunner.JsonA
private final String bucket;
private final DateTime exportTime;
private final int gcsBufferSize;
private final GcsUtils gcsUtils;
GenerateBindFileReducer(String bucket, DateTime exportTime, int gcsBufferSize) {
GenerateBindFileReducer(String bucket, DateTime exportTime, GcsUtils gcsUtils) {
this.bucket = bucket;
this.exportTime = exportTime;
this.gcsBufferSize = gcsBufferSize;
this.gcsUtils = gcsUtils;
}
@Override
public void reduce(String tld, ReducerInput<String> stanzas) {
String stanzaCounter = tld + " stanzas";
GcsFilename filename =
new GcsFilename(bucket, String.format(FILENAME_FORMAT, tld, exportTime));
GcsUtils cloudStorage =
new GcsUtils(createGcsService(RetryParams.getDefaultInstance()), gcsBufferSize);
try (OutputStream gcsOutput = cloudStorage.openOutputStream(filename);
BlobId filename = BlobId.of(bucket, String.format(FILENAME_FORMAT, tld, exportTime));
try (OutputStream gcsOutput = gcsUtils.openOutputStream(filename);
Writer osWriter = new OutputStreamWriter(gcsOutput, UTF_8);
PrintWriter writer = new PrintWriter(osWriter)) {
writer.printf(HEADER_FORMAT, tld);