mapreduce
diff --git a/java/com/google/domain/registry/tools/server/KillAllCommitLogsAction.java b/java/com/google/domain/registry/tools/server/KillAllCommitLogsAction.java
index fb44c8d40..ec2c3f30f 100644
--- a/java/com/google/domain/registry/tools/server/KillAllCommitLogsAction.java
+++ b/java/com/google/domain/registry/tools/server/KillAllCommitLogsAction.java
@@ -15,7 +15,6 @@
package com.google.domain.registry.tools.server;
import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.collect.Iterables.partition;
import static com.google.common.collect.Lists.partition;
import static com.google.domain.registry.model.ofy.ObjectifyService.ofy;
import static com.google.domain.registry.request.Action.Method.POST;
@@ -33,26 +32,13 @@ import com.google.domain.registry.request.Action;
import com.google.domain.registry.request.Response;
import com.googlecode.objectify.Key;
-import com.googlecode.objectify.VoidWork;
-
-import java.util.List;
import javax.inject.Inject;
-/**
- * Deletes all commit logs in datastore.
- *
- * Before running this, use the datastore admin page to delete all {@code CommitLogManifest} and
- * {@code CommitLogMutation} entities. That will take care of most (likely all) commit log entities
- * (except perhaps for very recently created entities that are missed by the eventually consistent
- * query driving that deletion) and it will be much faster than this mapreduce. After that, run this
- * to get a guarantee that everything was deleted.
- */
+/** Deletes all commit logs in datastore. */
@Action(path = "/_dr/task/killAllCommitLogs", method = POST)
public class KillAllCommitLogsAction implements MapreduceAction {
- private static final int BATCH_SIZE = 100;
-
@Inject MapreduceRunner mrRunner;
@Inject Response response;
@Inject KillAllCommitLogsAction() {}
@@ -60,16 +46,19 @@ public class KillAllCommitLogsAction implements MapreduceAction {
@Override
public void run() {
checkArgument( // safety
- RegistryEnvironment.get() == RegistryEnvironment.ALPHA
+ RegistryEnvironment.get() == RegistryEnvironment.CRASH
|| RegistryEnvironment.get() == RegistryEnvironment.UNITTEST,
- "DO NOT RUN ANYWHERE ELSE EXCEPT ALPHA OR TESTS.");
+ "DO NOT RUN ANYWHERE ELSE EXCEPT CRASH OR TESTS.");
// Create a in-memory input, assigning each bucket to its own shard for maximum parallelization.
Input> input =
new InMemoryInput<>(partition(CommitLogBucket.getAllBucketKeys().asList(), 1));
response.sendJavaScriptRedirect(createJobPath(mrRunner
.setJobName("Delete all commit logs")
.setModuleName("tools")
- .runMapOnly(new KillAllCommitLogsMapper(), ImmutableList.of(input))));
+ .runMapreduce(
+ new KillAllCommitLogsMapper(),
+ new KillAllEntitiesReducer(),
+ ImmutableList.of(input))));
}
/**
@@ -82,24 +71,17 @@ public class KillAllCommitLogsAction implements MapreduceAction {
* {@code CommitLogMutation}
*
*/
- static class KillAllCommitLogsMapper extends Mapper, Void, Void> {
+ static class KillAllCommitLogsMapper extends Mapper, Key, Key> {
private static final long serialVersionUID = 1504266335352952033L;
@Override
public void map(Key bucket) {
- // The query on the bucket could time out, but we are not worried about that because of the
- // procedure outlined above.
- for (final List> batch
- : partition(ofy().load().ancestor(bucket).keys(), BATCH_SIZE)) {
- ofy().transact(new VoidWork() {
- @Override
- public void vrun() {
- ofy().deleteWithoutBackup().entities(batch);
- }});
- getContext().incrementCounter("deleted entities", batch.size());
- }
- getContext().incrementCounter("completed buckets");
+ for (Key key : ofy().load().ancestor(bucket).keys()) {
+ emit(bucket, key);
+ getContext().incrementCounter("entities emitted");
+ getContext().incrementCounter(String.format("%s emitted", key.getKind()));
+ }
}
}
}
diff --git a/java/com/google/domain/registry/tools/server/KillAllEntitiesReducer.java b/java/com/google/domain/registry/tools/server/KillAllEntitiesReducer.java
new file mode 100644
index 000000000..3f8e63b74
--- /dev/null
+++ b/java/com/google/domain/registry/tools/server/KillAllEntitiesReducer.java
@@ -0,0 +1,53 @@
+// Copyright 2016 The Domain Registry 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 com.google.domain.registry.tools.server;
+
+import static com.google.common.collect.Iterators.partition;
+import static com.google.domain.registry.model.ofy.ObjectifyService.ofy;
+
+import com.google.appengine.tools.mapreduce.Reducer;
+import com.google.appengine.tools.mapreduce.ReducerInput;
+
+import com.googlecode.objectify.Key;
+import com.googlecode.objectify.VoidWork;
+
+import java.util.Iterator;
+import java.util.List;
+
+/** Reducer that deletes a group of keys, identified by a shared ancestor key. */
+public class KillAllEntitiesReducer extends Reducer, Key, Void> {
+
+ private static final long serialVersionUID = 7939357855356876000L;
+
+ private static final int BATCH_SIZE = 100;
+
+ @Override
+ public void reduce(Key ancestor, final ReducerInput> keysToDelete) {
+ Iterator>> batches = partition(keysToDelete, BATCH_SIZE);
+ while (batches.hasNext()) {
+ final List> batch = batches.next();
+ // Use a transaction to get retrying for free.
+ ofy().transact(new VoidWork() {
+ @Override
+ public void vrun() {
+ ofy().deleteWithoutBackup().keys(batch);
+ }});
+ getContext().incrementCounter("entities deleted", batch.size());
+ for (Key key : batch) {
+ getContext().incrementCounter(String.format("%s deleted", key.getKind()));
+ }
+ }
+ }
+}
diff --git a/java/com/google/domain/registry/tools/server/KillAllEppResourcesAction.java b/java/com/google/domain/registry/tools/server/KillAllEppResourcesAction.java
index 29a604e6b..7dd053e49 100644
--- a/java/com/google/domain/registry/tools/server/KillAllEppResourcesAction.java
+++ b/java/com/google/domain/registry/tools/server/KillAllEppResourcesAction.java
@@ -15,7 +15,6 @@
package com.google.domain.registry.tools.server;
import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.collect.Iterables.partition;
import static com.google.domain.registry.model.ofy.ObjectifyService.ofy;
import static com.google.domain.registry.request.Action.Method.POST;
import static com.google.domain.registry.util.PipelineUtils.createJobPath;
@@ -35,9 +34,7 @@ import com.google.domain.registry.request.Action;
import com.google.domain.registry.request.Response;
import com.googlecode.objectify.Key;
-import com.googlecode.objectify.VoidWork;
-
-import java.util.List;
+import com.googlecode.objectify.Work;
import javax.inject.Inject;
@@ -45,8 +42,6 @@ import javax.inject.Inject;
@Action(path = "/_dr/task/killAllEppResources", method = POST)
public class KillAllEppResourcesAction implements MapreduceAction {
- private static final int BATCH_SIZE = 100;
-
@Inject MapreduceRunner mrRunner;
@Inject Response response;
@Inject KillAllEppResourcesAction() {}
@@ -54,20 +49,21 @@ public class KillAllEppResourcesAction implements MapreduceAction {
@Override
public void run() {
checkArgument( // safety
- RegistryEnvironment.get() == RegistryEnvironment.ALPHA
+ RegistryEnvironment.get() == RegistryEnvironment.CRASH
|| RegistryEnvironment.get() == RegistryEnvironment.UNITTEST,
- "DO NOT RUN ANYWHERE ELSE EXCEPT ALPHA OR TESTS.");
+ "DO NOT RUN ANYWHERE ELSE EXCEPT CRASH OR TESTS.");
response.sendJavaScriptRedirect(createJobPath(mrRunner
.setJobName("Delete all EppResources, children, and indices")
.setModuleName("tools")
- .runMapOnly(
+ .runMapreduce(
new KillAllEppResourcesMapper(),
+ new KillAllEntitiesReducer(),
ImmutableList.of(EppResourceInputs.createIndexInput()))));
}
- static class KillAllEppResourcesMapper extends Mapper {
+ static class KillAllEppResourcesMapper extends Mapper, Key> {
- private static final long serialVersionUID = 103826288518612669L;
+ private static final long serialVersionUID = 8205309000002507407L;
/**
* Delete an {@link EppResourceIndex}, its referent, all descendants of each referent, and the
@@ -86,25 +82,31 @@ public class KillAllEppResourcesAction implements MapreduceAction {
*/
@Override
public void map(final EppResourceIndex eri) {
- EppResource resource = eri.getReference().get();
- for (final List> batch
- : partition(ofy().load().ancestor(resource).keys(), BATCH_SIZE)) {
- ofy().transact(new VoidWork() {
- @Override
- public void vrun() {
- ofy().deleteWithoutBackup().entities(batch);
- }});
- getContext().incrementCounter("deleted descendants", batch.size());
+ Key eriKey = Key.create(eri);
+ emitAndIncrementCounter(eriKey, eriKey);
+ Key resourceKey = eri.getReference().getKey();
+ for (Key key : ofy().load().ancestor(resourceKey).keys()) {
+ emitAndIncrementCounter(resourceKey, key);
}
- final Key foreignKey = resource instanceof DomainApplication
+ // Load in a transaction to make sure we don't get stale data (in case of host renames).
+ // TODO(b/27424173): A transaction is overkill. When we have memcache-skipping, use that.
+ EppResource resource = ofy().transactNewReadOnly(
+ new Work() {
+ @Override
+ public EppResource run() {
+ return eri.getReference().get();
+ }});
+ // TODO(b/28247733): What about FKI's for renamed hosts?
+ Key indexKey = resource instanceof DomainApplication
? DomainApplicationIndex.createKey((DomainApplication) resource)
: ForeignKeyIndex.createKey(resource);
- ofy().transact(new VoidWork() {
- @Override
- public void vrun() {
- ofy().deleteWithoutBackup().keys(Key.create(eri), foreignKey).now();
- }});
- getContext().incrementCounter("deleted eris");
+ emitAndIncrementCounter(indexKey, indexKey);
+ }
+
+ private void emitAndIncrementCounter(Key ancestor, Key child) {
+ emit(ancestor, child);
+ getContext().incrementCounter("entities emitted");
+ getContext().incrementCounter(String.format("%s emitted", child.getKind()));
}
}
}