Export entity integrity alerts to BigQuery

This is part 2 of a longer series.  Part 3 will add lots more tests, will add a cron entry, and will include an analysis script to run on BigQuery to detect the presence of two consecutive errors.
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=120040267

java/com/google/domain/registry/bigquery/BigqueryFactory.java

@@ -41,6 +41,8 @@ import com.google.domain.registry.util.NonFinalForTesting;
 import java.io.IOException;
 import java.util.Set;
 
+import javax.inject.Inject;
+
 /** Factory for creating {@link Bigquery} connections. */
 public class BigqueryFactory {
 
@@ -57,6 +59,8 @@ public class BigqueryFactory {
   @VisibleForTesting
   Subfactory subfactory = new Subfactory();
 
+  @Inject public BigqueryFactory() {}
+
   /** This class is broken out solely so that it can be mocked inside of tests. */
   static class Subfactory {
 

java/com/google/domain/registry/bigquery/BigquerySchemas.java

@@ -25,6 +25,13 @@ import java.util.Map;
 /** Schemas for BigQuery tables. */
 public final class BigquerySchemas {
 
+  public static final String EPPMETRICS_TABLE_ID = "eppMetrics";
+  public static final String ENTITY_INTEGRITY_ALERTS_TABLE_ID = "alerts";
+  public static final String ENTITY_INTEGRITY_ALERTS_FIELD_SCANTIME = "scanTime";
+  public static final String ENTITY_INTEGRITY_ALERTS_FIELD_SOURCE = "source";
+  public static final String ENTITY_INTEGRITY_ALERTS_FIELD_TARGET = "target";
+  public static final String ENTITY_INTEGRITY_ALERTS_FIELD_MESSAGE = "message";
+
   static final ImmutableList<TableFieldSchema> EPPMETRICS_SCHEMA_FIELDS =
       ImmutableList.<TableFieldSchema>of(
           new TableFieldSchema().setName("requestId").setType(FieldType.STRING.name()),
@@ -37,11 +44,27 @@ public final class BigquerySchemas {
           new TableFieldSchema().setName("eppStatus").setType(FieldType.INTEGER.name()),
           new TableFieldSchema().setName("attempts").setType(FieldType.INTEGER.name()));
 
-  public static final String EPPMETRICS_TABLE_ID = "eppMetrics";
+  static final ImmutableList<TableFieldSchema> ENTITY_INTEGRITY_ALERTS_SCHEMA_FIELDS =
+      ImmutableList.<TableFieldSchema>of(
+          new TableFieldSchema()
+              .setName(ENTITY_INTEGRITY_ALERTS_FIELD_SCANTIME)
+              .setType(FieldType.TIMESTAMP.name()),
+          new TableFieldSchema()
+              .setName(ENTITY_INTEGRITY_ALERTS_FIELD_SOURCE)
+              .setType(FieldType.STRING.name()),
+          new TableFieldSchema()
+              .setName(ENTITY_INTEGRITY_ALERTS_FIELD_TARGET)
+              .setType(FieldType.STRING.name()),
+          new TableFieldSchema()
+              .setName(ENTITY_INTEGRITY_ALERTS_FIELD_MESSAGE)
+              .setType(FieldType.STRING.name()));
 
   @NonFinalForTesting
   static Map<String, ImmutableList<TableFieldSchema>> knownTableSchemas =
-      ImmutableMap.of(EPPMETRICS_TABLE_ID, EPPMETRICS_SCHEMA_FIELDS);
+      new ImmutableMap.Builder<String, ImmutableList<TableFieldSchema>>()
+          .put(EPPMETRICS_TABLE_ID, EPPMETRICS_SCHEMA_FIELDS)
+          .put(ENTITY_INTEGRITY_ALERTS_TABLE_ID, ENTITY_INTEGRITY_ALERTS_SCHEMA_FIELDS)
+          .build();
 
   private BigquerySchemas() {}
 }

java/com/google/domain/registry/model/index/EppResourceIndex.java

@@ -42,6 +42,14 @@ public class EppResourceIndex extends BackupGroupRoot {
   @Index
   String kind;
 
+  public String getId() {
+    return id;
+  }
+
+  public String getKind() {
+    return kind;
+  }
+
   public Ref<? extends EppResource> getReference() {
     return reference;
   }

java/com/google/domain/registry/monitoring/whitebox/BUILD

@@ -8,9 +8,7 @@ java_library(
     srcs = glob(["*.java"]),
     deps = [
         "//apiserving/discoverydata/bigquery:bigqueryv2",
-        "//java/com/google/api/client/extensions/appengine/http",
+        "//java/com/google/api/client/util",
-        "//java/com/google/api/client/googleapis/extensions/appengine/auth/oauth2",
-        "//java/com/google/api/client/json/jackson2",
         "//java/com/google/common/annotations",
         "//java/com/google/common/base",
         "//java/com/google/common/cache",
@@ -22,9 +20,12 @@ java_library(
         "//java/com/google/domain/registry/mapreduce/inputs",
         "//java/com/google/domain/registry/model",
         "//java/com/google/domain/registry/request",
+        "//java/com/google/domain/registry/request:modules",
         "//java/com/google/domain/registry/util",
         "//third_party/java/appengine:appengine-api",
         "//third_party/java/appengine_mapreduce2:appengine_mapreduce",
+        "//third_party/java/auto:auto_factory",
+        "//third_party/java/dagger",
         "//third_party/java/joda_time",
         "//third_party/java/jsr305_annotations",
         "//third_party/java/jsr330_inject",

java/com/google/domain/registry/monitoring/whitebox/VerifyEntityIntegrityAction.java

@@ -22,7 +22,6 @@ import static com.google.domain.registry.util.DateTimeUtils.END_OF_TIME;
 import static com.google.domain.registry.util.DateTimeUtils.START_OF_TIME;
 import static com.google.domain.registry.util.DateTimeUtils.earliestOf;
 import static com.google.domain.registry.util.DateTimeUtils.isBeforeOrAt;
-import static com.google.domain.registry.util.DateTimeUtils.latestOf;
 import static com.google.domain.registry.util.FormattingLogger.getLoggerForCallerClass;
 import static com.google.domain.registry.util.PipelineUtils.createJobPath;
 import static com.googlecode.objectify.Key.getKind;
@@ -59,6 +58,7 @@ import com.google.domain.registry.model.transfer.TransferData.TransferServerAppr
 import com.google.domain.registry.request.Action;
 import com.google.domain.registry.request.Response;
 import com.google.domain.registry.util.FormattingLogger;
+import com.google.domain.registry.util.NonFinalForTesting;
 
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Ref;
@@ -67,9 +67,9 @@ import org.joda.time.DateTime;
 
 import java.io.Serializable;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import javax.annotation.Nullable;
@@ -96,9 +96,11 @@ import javax.inject.Inject;
 @Action(path = "/_dr/task/verifyEntityIntegrity")
 public class VerifyEntityIntegrityAction implements Runnable {
 
-  @VisibleForTesting
+  private static final FormattingLogger logger = getLoggerForCallerClass();
-  static final FormattingLogger logger = getLoggerForCallerClass();
   private static final int NUM_SHARDS = 200;
+  @NonFinalForTesting
+  @VisibleForTesting
+  static WhiteboxComponent component = DaggerWhiteboxComponent.create();
   private static final ImmutableSet<Class<?>> RESOURCE_CLASSES =
       ImmutableSet.<Class<?>>of(
           ForeignKeyDomainIndex.class,
@@ -124,13 +126,14 @@ public class VerifyEntityIntegrityAction implements Runnable {
 
   @Override
   public void run() {
+    DateTime scanTime = DateTime.now(UTC);
     response.sendJavaScriptRedirect(createJobPath(mrRunner
         .setJobName("Verify entity integrity")
         .setModuleName("backend")
         .setDefaultReduceShards(NUM_SHARDS)
         .runMapreduce(
-            new VerifyEntityIntegrityMapper(),
+            new VerifyEntityIntegrityMapper(scanTime),
-            new VerifyEntityIntegrityReducer(),
+            new VerifyEntityIntegrityReducer(scanTime),
             getInputs())));
   }
 
@@ -144,24 +147,38 @@ public class VerifyEntityIntegrityAction implements Runnable {
     return builder.build();
   }
 
+  /**
+   * The mapreduce key that the mapper outputs.  Each {@link EppResource} has two different
+   * mapreduce keys that are output for it: one for its specific type (domain, application, host, or
+   * contact), which is used to check {@link ForeignKeyIndex} constraints, and one that is common
+   * for all EppResources, to check {@link EppResourceIndex} constraints.
+   */
   private static enum EntityKind {
     DOMAIN,
     APPLICATION,
     CONTACT,
-    HOST
+    HOST,
+    /**
+     * Used to verify 1-to-1 constraints between all types of EPP resources and their indexes.
+     */
+    EPP_RESOURCE
   }
 
-  private static class FkAndKind implements Serializable {
+  private static class MapperKey implements Serializable {
 
-    private static final long serialVersionUID = -8466899721968889534L;
+    private static final long serialVersionUID = 3222302549441420932L;
 
-    public String foreignKey;
+    /**
+     * The relevant id for this mapper key, which is either the foreign key of the EppResource (for
+     * verifying foreign key indexes) or its repoId (for verifying EppResourceIndexes).
+     */
+    public String id;
     public EntityKind kind;
 
-    public static FkAndKind create(EntityKind kind, String foreignKey) {
+    public static MapperKey create(EntityKind kind, String id) {
-      FkAndKind instance = new FkAndKind();
+      MapperKey instance = new MapperKey();
       instance.kind = kind;
-      instance.foreignKey = foreignKey;
+      instance.id = id;
       return instance;
     }
   }
@@ -171,11 +188,26 @@ public class VerifyEntityIntegrityAction implements Runnable {
    * check integrity of foreign key entities.
    */
   public static class VerifyEntityIntegrityMapper
-      extends Mapper<Object, FkAndKind, Key<? extends ImmutableObject>> {
+      extends Mapper<Object, MapperKey, Key<? extends ImmutableObject>> {
 
-    private static final long serialVersionUID = -8881987421971102016L;
+    private static final long serialVersionUID = -5413882340475018051L;
+    private final DateTime scanTime;
 
-    public VerifyEntityIntegrityMapper() {}
+    private transient VerifyEntityIntegrityStreamer integrityStreamer;
+
+    // The integrityStreamer field must be marked as transient so that instances of the Mapper class
+    // can be serialized by the MapReduce framework.  Thus, every time is used, lazily construct it
+    // if it doesn't exist yet.
+    private VerifyEntityIntegrityStreamer integrity() {
+      if (integrityStreamer == null) {
+        integrityStreamer = component.verifyEntityIntegrityStreamerFactory().create(scanTime);
+      }
+      return integrityStreamer;
+    }
+
+    public VerifyEntityIntegrityMapper(DateTime scanTime) {
+      this.scanTime = scanTime;
+    }
 
     @Override
     public final void map(Object keyOrEntity) {
@@ -187,9 +219,9 @@ public class VerifyEntityIntegrityAction implements Runnable {
           keyOrEntity = ofy().load().key(key).now();
         }
         mapEntity(keyOrEntity);
-      } catch (Exception e) {
+      } catch (Throwable e) {
         // Log and swallow so that the mapreduce doesn't abort on first error.
-        logger.severefmt(e, "Integrity error found while parsing entity: %s", keyOrEntity);
+        logger.severefmt(e, "Exception while checking integrity of entity: %s", keyOrEntity);
       }
     }
 
@@ -203,11 +235,13 @@ public class VerifyEntityIntegrityAction implements Runnable {
       } else if (entity instanceof EppResourceIndex) {
         mapEppResourceIndex((EppResourceIndex) entity);
       } else {
-        throw new IllegalStateException(String.format("Unknown entity in mapper: %s", entity));
+        throw new IllegalStateException(
+            String.format("Unknown entity in integrity mapper: %s", entity));
       }
     }
 
     private void mapEppResource(EppResource resource) {
+      emit(MapperKey.create(EntityKind.EPP_RESOURCE, resource.getRepoId()), Key.create(resource));
       if (resource instanceof DomainBase) {
         DomainBase domainBase = (DomainBase) resource;
         Key<?> key = Key.create(domainBase);
@@ -232,7 +266,7 @@ public class VerifyEntityIntegrityAction implements Runnable {
           getContext().incrementCounter("domain applications");
           DomainApplication application = (DomainApplication) domainBase;
           emit(
-              FkAndKind.create(EntityKind.APPLICATION, application.getFullyQualifiedDomainName()),
+              MapperKey.create(EntityKind.APPLICATION, application.getFullyQualifiedDomainName()),
               Key.create(application));
         } else if (domainBase instanceof DomainResource) {
           getContext().incrementCounter("domain resources");
@@ -245,98 +279,97 @@ public class VerifyEntityIntegrityAction implements Runnable {
             verifyExistence(key, gracePeriod.getRecurringBillingEvent());
           }
           emit(
-              FkAndKind.create(EntityKind.DOMAIN, domain.getFullyQualifiedDomainName()),
+              MapperKey.create(EntityKind.DOMAIN, domain.getFullyQualifiedDomainName()),
               Key.create(domain));
         }
       } else if (resource instanceof ContactResource) {
         getContext().incrementCounter("contact resources");
         ContactResource contact = (ContactResource) resource;
         emit(
-            FkAndKind.create(EntityKind.CONTACT, contact.getContactId()),
+            MapperKey.create(EntityKind.CONTACT, contact.getContactId()),
             Key.create(contact));
       } else if (resource instanceof HostResource) {
         getContext().incrementCounter("host resources");
         HostResource host = (HostResource) resource;
         verifyExistence(Key.create(host), host.getSuperordinateDomain());
         emit(
-            FkAndKind.create(EntityKind.HOST, host.getFullyQualifiedHostName()),
+            MapperKey.create(EntityKind.HOST, host.getFullyQualifiedHostName()),
             Key.create(host));
       } else {
         throw new IllegalStateException(
-            String.format("EppResource with unknown type: %s", resource));
+            String.format("EppResource with unknown type in integrity mapper: %s", resource));
       }
     }
 
     private void mapForeignKeyIndex(ForeignKeyIndex<?> fki) {
+      Key<ForeignKeyIndex<?>> fkiKey = Key.<ForeignKeyIndex<?>>create(fki);
       @SuppressWarnings("cast")
-      EppResource resource = verifyExistence(Key.create(fki), fki.getReference());
+      EppResource resource = verifyExistence(fkiKey, fki.getReference());
-      checkState(
+      if (resource != null) {
+        integrity().check(
             fki.getForeignKey().equals(resource.getForeignKey()),
-          "Foreign key index %s points to EppResource with different foreign key: %s",
+            fkiKey,
-          fki,
+            Key.create(resource),
-          resource);
+            "Foreign key index points to EppResource with different foreign key");
-      if (resource instanceof DomainResource) {
+      }
+      if (fki instanceof ForeignKeyDomainIndex) {
         getContext().incrementCounter("domain foreign key indexes");
-        emit(FkAndKind.create(EntityKind.DOMAIN, resource.getForeignKey()), Key.create(fki));
+        emit(MapperKey.create(EntityKind.DOMAIN, fki.getForeignKey()), fkiKey);
-      } else if (resource instanceof ContactResource) {
+      } else if (fki instanceof ForeignKeyContactIndex) {
         getContext().incrementCounter("contact foreign key indexes");
-        emit(FkAndKind.create(EntityKind.CONTACT, resource.getForeignKey()), Key.create(fki));
+        emit(MapperKey.create(EntityKind.CONTACT, fki.getForeignKey()), fkiKey);
-      } else if (resource instanceof HostResource) {
+      } else if (fki instanceof ForeignKeyHostIndex) {
         getContext().incrementCounter("host foreign key indexes");
-        emit(FkAndKind.create(EntityKind.HOST, resource.getForeignKey()), Key.create(fki));
+        emit(MapperKey.create(EntityKind.HOST, fki.getForeignKey()), fkiKey);
       } else {
         throw new IllegalStateException(
-            String.format(
+            String.format("Foreign key index is of unknown type: %s", fki));
-                "Foreign key index %s points to EppResource of unknown type: %s", fki, resource));
       }
     }
 
     private void mapDomainApplicationIndex(DomainApplicationIndex dai) {
       getContext().incrementCounter("domain application indexes");
+      Key<DomainApplicationIndex> daiKey = Key.create(dai);
       for (Ref<DomainApplication> ref : dai.getReferences()) {
-        DomainApplication application = verifyExistence(Key.create(dai), ref);
+        DomainApplication application = verifyExistence(daiKey, ref);
-        checkState(
+        if (application != null) {
+          integrity().check(
               dai.getFullyQualifiedDomainName().equals(application.getFullyQualifiedDomainName()),
-            "Domain application index %s points to application with different domain name: %s",
+              daiKey,
-            dai,
+              Key.create(application),
-            application);
+              "Domain application index points to application with different domain name");
+        }
         emit(
-            FkAndKind.create(EntityKind.APPLICATION, application.getFullyQualifiedDomainName()),
+            MapperKey.create(EntityKind.APPLICATION, dai.getFullyQualifiedDomainName()),
-            Key.create(application));
+            daiKey);
       }
     }
 
     private void mapEppResourceIndex(EppResourceIndex eri) {
-      @SuppressWarnings("cast")
+      Key<EppResourceIndex> eriKey = Key.create(eri);
-      EppResource resource = verifyExistence(Key.create(eri), eri.getReference());
+      String eriRepoId = Key.create(eri.getId()).getName();
-      if (resource instanceof DomainResource) {
+      integrity().check(
-        getContext().incrementCounter("domain EPP resource indexes");
+          eriRepoId.equals(eri.getReference().getKey().getName()),
-        emit(FkAndKind.create(EntityKind.DOMAIN, resource.getForeignKey()), Key.create(eri));
+          eriKey,
-      } else if (resource instanceof ContactResource) {
+          eri.getReference().getKey(),
-        getContext().incrementCounter("contact EPP resource indexes");
+          "EPP resource index id does not match repoId of reference");
-        emit(
+      verifyExistence(eriKey, eri.getReference());
-            FkAndKind.create(EntityKind.CONTACT, resource.getForeignKey()), Key.create(eri));
+      emit(MapperKey.create(EntityKind.EPP_RESOURCE, eriRepoId), eriKey);
-      } else if (resource instanceof HostResource) {
+      getContext().incrementCounter("EPP resource indexes to " + eri.getKind());
-        getContext().incrementCounter("host EPP resource indexes");
-        emit(FkAndKind.create(EntityKind.HOST, resource.getForeignKey()), Key.create(eri));
-      } else {
-        throw new IllegalStateException(
-            String.format(
-                "EPP resource index %s points to resource of unknown type: %s", eri, resource));
-      }
     }
 
-    private static <E> void verifyExistence(Key<?> source, Set<Key<E>> keys) {
+    private <E> void verifyExistence(Key<?> source, Set<Key<E>> targets) {
-      Set<Key<E>> missingEntityKeys = Sets.difference(keys, ofy().load().<E>keys(keys).keySet());
+      Set<Key<E>> missingEntityKeys =
-      checkState(
+          Sets.difference(targets, ofy().load().<E>keys(targets).keySet());
+      integrity().checkOneToMany(
           missingEntityKeys.isEmpty(),
-          "Existing entity %s referenced entities that do not exist: %s",
           source,
-          missingEntityKeys);
+          targets,
+          "Target entity does not exist");
     }
 
     @Nullable
-    private static <E> E verifyExistence(Key<?> source, @Nullable Ref<E> target) {
+    private <E> E verifyExistence(Key<?> source, @Nullable Ref<E> target) {
       if (target == null) {
         return null;
       }
@@ -344,15 +377,12 @@ public class VerifyEntityIntegrityAction implements Runnable {
     }
 
     @Nullable
-    private static <E> E verifyExistence(Key<?> source, @Nullable Key<E> target) {
+    private <E> E verifyExistence(Key<?> source, @Nullable Key<E> target) {
       if (target == null) {
         return null;
       }
       E entity = ofy().load().key(target).now();
-      checkState(entity != null,
+      integrity().check(entity != null, source, target, "Target entity does not exist");
-          "Existing entity %s referenced entity that does not exist: %s",
-          source,
-          target);
       return entity;
     }
 
@@ -371,50 +401,108 @@ public class VerifyEntityIntegrityAction implements Runnable {
 
   /** Reducer that checks integrity of foreign key entities. */
   public static class VerifyEntityIntegrityReducer
-      extends Reducer<FkAndKind, Key<? extends ImmutableObject>, Void> {
+      extends Reducer<MapperKey, Key<? extends ImmutableObject>, Void> {
 
-    private static final long serialVersionUID = -8531280188397051521L;
+    private static final long serialVersionUID = -151271247606894783L;
+
+    private final DateTime scanTime;
+
+    private transient VerifyEntityIntegrityStreamer integrityStreamer;
+
+    // The integrityStreamer field must be marked as transient so that instances of the Reducer
+    // class can be serialized by the MapReduce framework.  Thus, every time is used, lazily
+    // construct it if it doesn't exist yet.
+    private VerifyEntityIntegrityStreamer integrity() {
+      if (integrityStreamer == null) {
+        integrityStreamer = component.verifyEntityIntegrityStreamerFactory().create(scanTime);
+      }
+      return integrityStreamer;
+    }
+
+    public VerifyEntityIntegrityReducer(DateTime scanTime) {
+      this.scanTime = scanTime;
+    }
 
     @SuppressWarnings("unchecked")
     @Override
-    public void reduce(FkAndKind fkAndKind, ReducerInput<Key<? extends ImmutableObject>> keys) {
+    public void reduce(MapperKey mapperKey, ReducerInput<Key<? extends ImmutableObject>> keys) {
       try {
-        reduceKeys(fkAndKind, keys);
+        reduceKeys(mapperKey, keys);
-      } catch (Exception e) {
+      } catch (Throwable e) {
         // Log and swallow so that the mapreduce doesn't abort on first error.
         logger.severefmt(
-            e, "Integrity error found while checking foreign key contraints for: %s", fkAndKind);
+            e, "Exception while checking foreign key integrity constraints for: %s", mapperKey);
       }
     }
 
     private void reduceKeys(
-        FkAndKind fkAndKind, ReducerInput<Key<? extends ImmutableObject>> keys) {
+        MapperKey mapperKey, ReducerInput<Key<? extends ImmutableObject>> keys) {
-      switch (fkAndKind.kind) {
+      getContext().incrementCounter("reduced resources " + mapperKey.kind);
+      switch (mapperKey.kind) {
+        case EPP_RESOURCE:
+          checkEppResourceIndexes(keys, mapperKey.id);
+          break;
         case APPLICATION:
-          getContext().incrementCounter("domain applications");
           checkIndexes(
               keys,
-              fkAndKind.foreignKey,
+              mapperKey.id,
               KIND_DOMAIN_BASE_RESOURCE,
               KIND_DOMAIN_APPLICATION_INDEX,
               false);
           break;
         case CONTACT:
-          getContext().incrementCounter("contact resources");
+          checkIndexes(keys, mapperKey.id, KIND_CONTACT_RESOURCE, KIND_CONTACT_INDEX, true);
-          checkIndexes(keys, fkAndKind.foreignKey, KIND_CONTACT_RESOURCE, KIND_CONTACT_INDEX, true);
           break;
         case DOMAIN:
-          getContext().incrementCounter("domain resources");
           checkIndexes(
-              keys, fkAndKind.foreignKey, KIND_DOMAIN_BASE_RESOURCE, KIND_DOMAIN_INDEX, true);
+              keys, mapperKey.id, KIND_DOMAIN_BASE_RESOURCE, KIND_DOMAIN_INDEX, true);
           break;
         case HOST:
-          getContext().incrementCounter("host resources");
+          checkIndexes(keys, mapperKey.id, KIND_HOST_RESOURCE, KIND_HOST_INDEX, true);
-          checkIndexes(keys, fkAndKind.foreignKey, KIND_HOST_RESOURCE, KIND_HOST_INDEX, true);
           break;
         default:
           throw new IllegalStateException(
-              String.format("Unknown type of foreign key %s", fkAndKind.kind));
+              String.format("Unknown type of foreign key %s", mapperKey.kind));
+      }
+    }
+
+    @SuppressWarnings("unchecked")
+    private void checkEppResourceIndexes(
+        Iterator<Key<? extends ImmutableObject>> keys, String repoId) {
+      List<Key<EppResource>> resources = new ArrayList<>();
+      List<Key<EppResourceIndex>> eppResourceIndexes = new ArrayList<>();
+      while (keys.hasNext()) {
+        Key<?> key = keys.next();
+        String kind = key.getKind();
+        if (kind.equals(KIND_EPPRESOURCE_INDEX)) {
+          eppResourceIndexes.add((Key<EppResourceIndex>) key);
+        } else if (kind.equals(KIND_DOMAIN_BASE_RESOURCE)
+            || kind.equals(KIND_CONTACT_RESOURCE)
+            || kind.equals(KIND_HOST_RESOURCE)) {
+          resources.add((Key<EppResource>) key);
+        } else {
+          throw new IllegalStateException(
+              String.format(
+                  "While verifying EppResourceIndexes for repoId %s, found key of unknown type: %s",
+                  repoId,
+                  key));
+        }
+      }
+      // This is a checkState and not an integrity check because the Datastore schema ensures that
+      // there can't be multiple EppResources with the same repoId.
+      checkState(
+          resources.size() == 1,
+          String.format("Found more than one EppResource for repoId %s: %s", repoId, resources));
+      if (integrity().check(
+          !eppResourceIndexes.isEmpty(),
+          null,
+          getOnlyElement(resources),
+          "Missing EPP resource index for EPP resource")) {
+        integrity().checkManyToOne(
+            eppResourceIndexes.size() == 1,
+            eppResourceIndexes,
+            getOnlyElement(resources),
+            "Duplicate EPP resource indexes pointing to same resource");
       }
     }
 
@@ -427,15 +515,12 @@ public class VerifyEntityIntegrityAction implements Runnable {
         boolean thereCanBeOnlyOne) {
       List<Key<R>> resources = new ArrayList<>();
       List<Key<I>> foreignKeyIndexes = new ArrayList<>();
-      List<Key<EppResourceIndex>> eppResourceIndexes = new ArrayList<>();
       while (keys.hasNext()) {
         Key<?> key = keys.next();
         if (key.getKind().equals(resourceKind)) {
           resources.add((Key<R>) key);
         } else if (key.getKind().equals(foreignKeyIndexKind)) {
           foreignKeyIndexes.add((Key<I>) key);
-        } else if (key.getKind().equals(KIND_EPPRESOURCE_INDEX)) {
-          eppResourceIndexes.add((Key<EppResourceIndex>) key);
         } else {
           throw new IllegalStateException(
               String.format(
@@ -445,67 +530,65 @@ public class VerifyEntityIntegrityAction implements Runnable {
                   key));
         }
       }
+      // This is a checkState and not an integrity check because it should truly be impossible to
+      // have multiple foreign key indexes for the same foreign key because of the Datastore schema.
       checkState(
-          foreignKeyIndexes.size() == 1,
+          foreignKeyIndexes.size() <= 1,
           String.format(
-              "Should have found exactly 1 foreign key index for %s, instead found %d: %s",
+              "Found more than one foreign key index for %s: %s", foreignKey, foreignKeyIndexes));
+      integrity().check(
+          !foreignKeyIndexes.isEmpty(),
           foreignKey,
-              foreignKeyIndexes.size(),
+          resourceKind,
-              foreignKeyIndexes));
+          "Missing foreign key index for EppResource");
-      checkState(
-          !resources.isEmpty(),
-          "Foreign key index %s exists, but no matching EPP resources found",
-          getOnlyElement(foreignKeyIndexes));
-      checkState(eppResourceIndexes.size() == 1,
-          "Should have found exactly 1 EPP resource index for %s, instead found: %s",
-          foreignKey,
-          eppResourceIndexes);
       if (thereCanBeOnlyOne) {
-        verifyOnlyOneActiveResource(foreignKey, resources, foreignKeyIndexes);
+        verifyOnlyOneActiveResource(resources, getOnlyElement(foreignKeyIndexes));
       }
     }
 
     private <R extends EppResource, I> void verifyOnlyOneActiveResource(
-        String foreignKey, List<Key<R>> resources, List<Key<I>> foreignKeyIndexes) {
+        List<Key<R>> resources, Key<I> fkiKey) {
-      DateTime now = DateTime.now(UTC);
       DateTime oldestActive = END_OF_TIME;
       DateTime mostRecentInactive = START_OF_TIME;
-      List<R> activeResources = new ArrayList<R>();
+      Key<R> mostRecentInactiveKey = null;
-      Collection<R> allResources = ofy().load().keys(resources).values();
+      List<Key<R>> activeResources = new ArrayList<>();
-      ForeignKeyIndex<?> fki =
+      Map<Key<R>, R> allResources = ofy().load().keys(resources);
-          (ForeignKeyIndex<?>) ofy().load().key(getOnlyElement(foreignKeyIndexes)).now();
+      ForeignKeyIndex<?> fki = (ForeignKeyIndex<?>) ofy().load().key(fkiKey).now();
-      for (R resource : allResources) {
+      for (Map.Entry<Key<R>, R> entry : allResources.entrySet()) {
-        if (isActive(resource, now)) {
+        R resource = entry.getValue();
-          activeResources.add(resource);
+        if (isActive(resource, scanTime)) {
+          activeResources.add(entry.getKey());
           oldestActive = earliestOf(oldestActive, resource.getCreationTime());
         } else {
-          mostRecentInactive = latestOf(mostRecentInactive, resource.getDeletionTime());
+          if (resource.getDeletionTime().isAfter(mostRecentInactive)) {
+            mostRecentInactive = resource.getDeletionTime();
+            mostRecentInactiveKey = entry.getKey();
+          }
         }
       }
       if (activeResources.isEmpty()) {
-        checkState(
+        integrity().check(
             fki.getDeletionTime().isEqual(mostRecentInactive),
-            "Deletion time on foreign key index %s doesn't match"
+            fkiKey,
-                + " most recently deleted resource from: %s",
+            mostRecentInactiveKey,
-            fki,
+            "Foreign key index deletion time not equal to that of most recently deleted resource");
-            allResources);
       } else {
-        checkState(
+        integrity().checkOneToMany(
-            activeResources.size() <= 1,
+            activeResources.size() == 1,
-            "Found multiple active resources with foreign key %s: %s",
+            fkiKey,
-            foreignKey,
+            activeResources,
-            activeResources);
+            "Multiple active EppResources with same foreign key");
-        checkState(
+        integrity().check(
             fki.getDeletionTime().isEqual(END_OF_TIME),
-            "Deletion time on foreign key index %s doesn't match active resource: %s",
+            fkiKey,
-            fki,
+            null,
-            getOnlyElement(activeResources));
+            "Foreign key index has deletion time but active resource exists");
-        checkState(
+        integrity().check(
             isBeforeOrAt(mostRecentInactive, oldestActive),
-            "Found inactive resource that is more recent than active resource in: %s",
+            fkiKey,
-            allResources);
+            mostRecentInactiveKey,
-      }
+            "Found inactive resource deleted more recently than when active resource was created");
-
+      }
     }
   }
 }

java/com/google/domain/registry/monitoring/whitebox/VerifyEntityIntegrityStreamer.java

@@ -0,0 +1,223 @@
+// 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.monitoring.whitebox;
+
+import static com.google.api.client.util.Data.NULL_STRING;
+import static com.google.domain.registry.bigquery.BigquerySchemas.ENTITY_INTEGRITY_ALERTS_FIELD_MESSAGE;
+import static com.google.domain.registry.bigquery.BigquerySchemas.ENTITY_INTEGRITY_ALERTS_FIELD_SCANTIME;
+import static com.google.domain.registry.bigquery.BigquerySchemas.ENTITY_INTEGRITY_ALERTS_FIELD_SOURCE;
+import static com.google.domain.registry.bigquery.BigquerySchemas.ENTITY_INTEGRITY_ALERTS_FIELD_TARGET;
+import static com.google.domain.registry.bigquery.BigquerySchemas.ENTITY_INTEGRITY_ALERTS_TABLE_ID;
+
+import com.google.api.services.bigquery.Bigquery;
+import com.google.api.services.bigquery.Bigquery.Tabledata.InsertAll;
+import com.google.api.services.bigquery.model.TableDataInsertAllRequest;
+import com.google.api.services.bigquery.model.TableDataInsertAllRequest.Rows;
+import com.google.api.services.bigquery.model.TableDataInsertAllResponse;
+import com.google.api.services.bigquery.model.TableDataInsertAllResponse.InsertErrors;
+import com.google.auto.factory.AutoFactory;
+import com.google.auto.factory.Provided;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Supplier;
+import com.google.common.collect.FluentIterable;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.domain.registry.bigquery.BigqueryFactory;
+import com.google.domain.registry.config.RegistryEnvironment;
+import com.google.domain.registry.util.Retrier;
+
+import org.joda.time.DateTime;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import javax.annotation.Nullable;
+
+/**
+ * An injected utility class used to check entity integrity and stream violations to BigQuery.
+ */
+@AutoFactory(allowSubclasses = true)
+public class VerifyEntityIntegrityStreamer {
+
+  private static final String DATASET = "entity_integrity";
+
+  private final DateTime scanTime;
+  private Bigquery bigquery;
+
+  BigqueryFactory bigqueryFactory;
+  RegistryEnvironment environment;
+  Retrier retrier;
+  Supplier<String> idGenerator;
+
+  public VerifyEntityIntegrityStreamer(
+      @Provided BigqueryFactory bigqueryFactory,
+      @Provided RegistryEnvironment environment,
+      @Provided Retrier retrier,
+      @Provided Supplier<String> idGenerator,
+      DateTime scanTime) {
+    this.bigqueryFactory = bigqueryFactory;
+    this.environment = environment;
+    this.retrier = retrier;
+    this.idGenerator = idGenerator;
+    this.scanTime = scanTime;
+  }
+
+  // This is lazily loaded so we only construct the connection when needed, as in a healthy
+  // Datastore almost every single check should short-circuit return and won't output anything to
+  // BigQuery.
+  private Bigquery getBigquery() throws IOException {
+    if (bigquery == null) {
+      bigquery =
+          bigqueryFactory.create(
+              environment.config().getProjectId(), DATASET, ENTITY_INTEGRITY_ALERTS_TABLE_ID);
+    }
+    return bigquery;
+  }
+
+  /**
+   * Check that the given conditional holds, and if not, stream the supplied source, target, and
+   * message information to BigQuery.
+   *
+   * @return Whether the check succeeded.
+   */
+  boolean check(
+      boolean conditional,
+      @Nullable Object source,
+      @Nullable Object target,
+      @Nullable String message) {
+    return checkOneToMany(
+        conditional, source, ImmutableList.of((target == null) ? NULL_STRING : target), message);
+  }
+
+  /**
+   * Check that the given conditional holds, and if not, stream a separate row to BigQuery for each
+   * supplied target (the source and message will be the same for each).
+   *
+   * @return Whether the check succeeded.
+   */
+  <T> boolean checkOneToMany(
+      boolean conditional,
+      @Nullable Object source,
+      Iterable<T> targets,
+      @Nullable String message) {
+    return checkManyToMany(
+        conditional, ImmutableList.of((source == null) ? NULL_STRING : source), targets, message);
+  }
+
+  /**
+   * Check that the given conditional holds, and if not, stream a separate row to BigQuery for each
+   * supplied target (the source and message will be the same for each).
+   *
+   * @return Whether the check succeeded.
+   */
+  <S> boolean checkManyToOne(
+      boolean conditional,
+      Iterable<S> sources,
+      @Nullable Object target,
+      @Nullable String message) {
+    return checkManyToMany(
+        conditional, sources, ImmutableList.of((target == null) ? NULL_STRING : target), message);
+  }
+
+
+  /**
+   * Check that the given conditional holds, and if not, stream a separate row to BigQuery for the
+   * cross product of every supplied target and source (the message will be the same for each).
+   * This is used in preference to records (repeated fields) in BigQuery because they are
+   * significantly harder to work with.
+   *
+   * @return Whether the check succeeded.
+   */
+  private <S, T> boolean checkManyToMany(
+      boolean conditional,
+      Iterable<S> sources,
+      Iterable<T> targets,
+      @Nullable String message) {
+    if (conditional) {
+      return true;
+    }
+    ImmutableList.Builder<Rows> rows = new ImmutableList.Builder<>();
+    for (S source : sources) {
+      for (T target : targets) {
+        Map<String, Object> rowData =
+            new ImmutableMap.Builder<String, Object>()
+                .put(
+                    ENTITY_INTEGRITY_ALERTS_FIELD_SCANTIME,
+                    new com.google.api.client.util.DateTime(scanTime.toDate()))
+                .put(
+                    ENTITY_INTEGRITY_ALERTS_FIELD_SOURCE,
+                    source.toString())
+                .put(
+                    ENTITY_INTEGRITY_ALERTS_FIELD_TARGET,
+                    target.toString())
+                .put(
+                    ENTITY_INTEGRITY_ALERTS_FIELD_MESSAGE,
+                    (message == null) ? NULL_STRING : message)
+                .build();
+        rows.add(
+            new TableDataInsertAllRequest.Rows().setJson(rowData).setInsertId(idGenerator.get()));
+      }
+    }
+    streamToBigqueryWithRetry(rows.build());
+    return false;
+  }
+
+  private void streamToBigqueryWithRetry(List<Rows> rows) {
+    try {
+      final InsertAll request =
+          getBigquery()
+              .tabledata()
+              .insertAll(
+                  environment.config().getProjectId(),
+                  DATASET,
+                  ENTITY_INTEGRITY_ALERTS_TABLE_ID,
+                  new TableDataInsertAllRequest().setRows(rows));
+
+      Callable<Void> callable =
+          new Callable<Void>() {
+            @Override
+            public Void call() throws Exception {
+              TableDataInsertAllResponse response = request.execute();
+              // Turn errors on the response object into RuntimeExceptions that the retrier will
+              // retry.
+              if (response.getInsertErrors() != null && !response.getInsertErrors().isEmpty()) {
+                throw new RuntimeException(
+                    FluentIterable.from(response.getInsertErrors())
+                        .transform(
+                            new Function<InsertErrors, String>() {
+                              @Override
+                              public String apply(InsertErrors error) {
+                                try {
+                                  return error.toPrettyString();
+                                } catch (IOException e) {
+                                  return error.toString();
+                                }
+                              }
+                            })
+                        .join(Joiner.on('\n')));
+              }
+              return null;
+            }
+          };
+      retrier.callWithRetry(callable, RuntimeException.class);
+    } catch (IOException e) {
+      throw new RuntimeException("Error sending integrity error to BigQuery", e);
+    }
+  }
+}
+

java/com/google/domain/registry/monitoring/whitebox/WhiteboxComponent.java

@@ -0,0 +1,36 @@
+// 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.monitoring.whitebox;
+
+import com.google.domain.registry.bigquery.BigqueryModule;
+import com.google.domain.registry.config.ConfigModule;
+import com.google.domain.registry.request.Modules.DatastoreServiceModule;
+
+import dagger.Component;
+
+import javax.inject.Singleton;
+
+/** Dagger component with instance lifetime for Whitebox package. */
+@Singleton
+@Component(
+    modules = {
+        BigqueryModule.class,
+        ConfigModule.class,
+        DatastoreServiceModule.class,
+        WhiteboxModule.class
+    })
+interface WhiteboxComponent {
+  VerifyEntityIntegrityStreamerFactory verifyEntityIntegrityStreamerFactory();
+}

java/com/google/domain/registry/monitoring/whitebox/WhiteboxModule.java

@@ -0,0 +1,46 @@
+// 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.monitoring.whitebox;
+
+import com.google.common.base.Supplier;
+import com.google.domain.registry.util.Sleeper;
+import com.google.domain.registry.util.SystemSleeper;
+
+import dagger.Module;
+import dagger.Provides;
+
+import java.util.UUID;
+
+/**
+ * Dagger module for injecting common settings for Whitebox tasks.
+ */
+@Module
+public class WhiteboxModule {
+
+  @Provides
+  static Supplier<String> provideIdGenerator() {
+    return new Supplier<String>() {
+      @Override
+      public String get() {
+        return UUID.randomUUID().toString();
+      }
+    };
+  }
+
+  @Provides
+  static Sleeper provideSleeper(SystemSleeper systemSleeper) {
+    return systemSleeper;
+  }
+}

javatests/com/google/domain/registry/monitoring/whitebox/BUILD

@@ -12,6 +12,7 @@ java_library(
         "//apiserving/discoverydata/bigquery:bigqueryv2",
         "//java/com/google/api/client/http",
         "//java/com/google/api/client/json",
+        "//java/com/google/api/client/util",
         "//java/com/google/common/base",
         "//java/com/google/common/collect",
         "//java/com/google/common/net",
@@ -21,6 +22,7 @@ java_library(
         "//java/com/google/domain/registry/mapreduce",
         "//java/com/google/domain/registry/model",
         "//java/com/google/domain/registry/monitoring/whitebox",
+        "//java/com/google/domain/registry/util",
         "//javatests/com/google/domain/registry/testing",
         "//javatests/com/google/domain/registry/testing/mapreduce",
         "//third_party/java/appengine:appengine-api-testonly",

javatests/com/google/domain/registry/monitoring/whitebox/VerifyEntityIntegrityActionTest.java

@@ -14,49 +14,117 @@
 
 package com.google.domain.registry.monitoring.whitebox;
 
+import static com.google.common.truth.Truth.assertThat;
 import static com.google.domain.registry.testing.DatastoreHelper.createTld;
 import static com.google.domain.registry.testing.DatastoreHelper.deleteResource;
+import static com.google.domain.registry.testing.DatastoreHelper.newDomainResource;
+import static com.google.domain.registry.testing.DatastoreHelper.persistActiveContact;
 import static com.google.domain.registry.testing.DatastoreHelper.persistActiveDomain;
-import static com.google.domain.registry.testing.LogsSubject.assertAboutLogs;
+import static com.google.domain.registry.testing.DatastoreHelper.persistActiveHost;
-import static java.util.logging.Level.SEVERE;
+import static com.google.domain.registry.testing.DatastoreHelper.persistDeletedContact;
+import static com.google.domain.registry.testing.DatastoreHelper.persistDomainAsDeleted;
+import static com.google.domain.registry.testing.DatastoreHelper.persistResource;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
 
+import com.google.api.client.util.Data;
+import com.google.api.services.bigquery.Bigquery;
+import com.google.api.services.bigquery.model.TableDataInsertAllRequest;
+import com.google.api.services.bigquery.model.TableDataInsertAllRequest.Rows;
+import com.google.api.services.bigquery.model.TableDataInsertAllResponse;
 import com.google.common.base.Optional;
-import com.google.common.testing.TestLogHandler;
+import com.google.common.base.Suppliers;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.domain.registry.bigquery.BigqueryFactory;
+import com.google.domain.registry.config.RegistryEnvironment;
 import com.google.domain.registry.mapreduce.MapreduceRunner;
+import com.google.domain.registry.model.contact.ContactResource;
 import com.google.domain.registry.model.domain.DomainResource;
+import com.google.domain.registry.model.domain.ReferenceUnion;
+import com.google.domain.registry.model.host.HostResource;
+import com.google.domain.registry.model.index.EppResourceIndex;
 import com.google.domain.registry.model.index.ForeignKeyIndex;
+import com.google.domain.registry.model.index.ForeignKeyIndex.ForeignKeyContactIndex;
+import com.google.domain.registry.model.index.ForeignKeyIndex.ForeignKeyDomainIndex;
+import com.google.domain.registry.testing.FakeClock;
 import com.google.domain.registry.testing.FakeResponse;
+import com.google.domain.registry.testing.FakeSleeper;
+import com.google.domain.registry.testing.InjectRule;
 import com.google.domain.registry.testing.mapreduce.MapreduceTestCase;
+import com.google.domain.registry.util.Retrier;
+
+import com.googlecode.objectify.Key;
+import com.googlecode.objectify.Ref;
 
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
-import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import java.util.Map;
 
 /** Unit tests for {@link VerifyEntityIntegrityAction}. */
-@RunWith(JUnit4.class)
+@RunWith(MockitoJUnitRunner.class)
 public class VerifyEntityIntegrityActionTest
     extends MapreduceTestCase<VerifyEntityIntegrityAction> {
 
-  private TestLogHandler handler;
+  @Rule
+  public final InjectRule inject = new InjectRule();
+
+  private VerifyEntityIntegrityStreamer integrity;
+  private ArgumentCaptor<TableDataInsertAllRequest> rowsCaptor;
+  private final DateTime now = DateTime.parse("2012-01-02T03:04:05Z");
+
+  @Mock
+  private Bigquery bigquery;
+
+  @Mock
+  private Bigquery.Tabledata bigqueryTableData;
+
+  @Mock
+  private Bigquery.Tabledata.InsertAll bigqueryInsertAll;
+
+  @Mock
+  private BigqueryFactory bigqueryFactory;
+
+  @Mock
+  private VerifyEntityIntegrityStreamerFactory streamerFactory;
 
   @Before
-  public void before() {
+  public void before() throws Exception {
     createTld("tld");
 
     action = new VerifyEntityIntegrityAction();
-    handler = new TestLogHandler();
-    VerifyEntityIntegrityAction.logger.addHandler(handler);
     action.mrRunner = new MapreduceRunner(Optional.of(2), Optional.of(2));
     action.response = new FakeResponse();
-  }
+    WhiteboxComponent component = mock(WhiteboxComponent.class);
+    inject.setStaticField(VerifyEntityIntegrityAction.class, "component", component);
+    integrity =
+        new VerifyEntityIntegrityStreamer(
+            bigqueryFactory,
+            RegistryEnvironment.UNITTEST,
+            new Retrier(new FakeSleeper(new FakeClock()), 1),
+            Suppliers.ofInstance("rowid"),
+            now);
+    when(bigqueryFactory.create(anyString(), anyString(), anyString())).thenReturn(bigquery);
+    when(component.verifyEntityIntegrityStreamerFactory()).thenReturn(streamerFactory);
+    when(streamerFactory.create(any(DateTime.class))).thenReturn(integrity);
+    when(bigquery.tabledata()).thenReturn(bigqueryTableData);
+    rowsCaptor = ArgumentCaptor.forClass(TableDataInsertAllRequest.class);
+    when(bigqueryTableData.insertAll(anyString(), anyString(), anyString(), rowsCaptor.capture()))
+            .thenReturn(bigqueryInsertAll);
+    when(bigqueryInsertAll.execute()).thenReturn(new TableDataInsertAllResponse());
 
-  @After
-  public void after() {
-    VerifyEntityIntegrityAction.logger.removeHandler(handler);
   }
 
   private void runMapreduce() throws Exception {
@@ -68,20 +136,145 @@ public class VerifyEntityIntegrityActionTest
   public void test_singleDomain_noBadInvariants() throws Exception {
     persistActiveDomain("ninetails.tld");
     runMapreduce();
-    assertAboutLogs().that(handler).hasNoLogsAtLevel(SEVERE);
+    verifyZeroInteractions(bigquery);
   }
 
   @Test
-  public void test_singleDomain_missingFki() throws Exception {
+  public void test_lotsOfData_noBadInvariants() throws Exception {
+    createTld("march");
+    ContactResource contact = persistActiveContact("longbottom");
+    persistResource(newDomainResource("ninetails.tld", contact));
+    persistResource(newDomainResource("tentails.tld", contact));
+    persistDomainAsDeleted(newDomainResource("long.march", contact), now.minusMonths(4));
+    persistResource(
+        newDomainResource("long.march", contact)
+            .asBuilder()
+            .setCreationTimeForTest(now.minusMonths(3))
+            .build());
+    persistDeletedContact("ricketycricket", now.minusDays(3));
+    persistDeletedContact("ricketycricket", now.minusDays(2));
+    persistDeletedContact("ricketycricket", now.minusDays(1));
+    persistActiveContact("ricketycricket");
+    persistActiveHost("ns9001.example.net");
+    runMapreduce();
+    verifyZeroInteractions(bigquery);
+  }
+
+  @Test
+  public void test_missingFki() throws Exception {
     persistActiveDomain("ninetails.tld");
     ForeignKeyIndex<DomainResource> fki =
         ForeignKeyIndex.load(DomainResource.class, "ninetails.tld", DateTime.now(DateTimeZone.UTC));
     deleteResource(fki);
     runMapreduce();
-    // TODO(mcilwain): Check for exception message here.
+    assertIntegrityErrors(IntegrityError.create(
-    assertAboutLogs()
+        "ninetails.tld", "DomainBase", "Missing foreign key index for EppResource"));
-        .that(handler)
+  }
-        .hasLogAtLevelWithMessage(
+
-            SEVERE, "Integrity error found while checking foreign key contraints");
+  @Test
+  public void test_missingEppResourceIndex() throws Exception {
+    Key<ContactResource> cooperKey = Key.create(persistActiveContact("cooper"));
+    deleteResource(EppResourceIndex.create(cooperKey));
+    runMapreduce();
+    assertIntegrityErrors(IntegrityError.create(
+        Data.NULL_STRING, cooperKey.toString(), "Missing EPP resource index for EPP resource"));
+  }
+
+  @Test
+  public void test_referencesToHostsThatDontExist() throws Exception {
+    Key<HostResource> missingHost1 = Key.create(HostResource.class, "DEADBEEF-ROID");
+    Key<HostResource> missingHost2 = Key.create(HostResource.class, "123ABC-ROID");
+    Key<HostResource> missingHost3 = Key.create(HostResource.class, "FADDACA-ROID");
+    DomainResource domain =
+        persistResource(
+            newDomainResource("blah.tld")
+                .asBuilder()
+                .setNameservers(
+                    ImmutableSet.of(
+                        ReferenceUnion.create(Ref.create(missingHost1)),
+                        ReferenceUnion.create(Ref.create(missingHost2)),
+                        ReferenceUnion.create(Ref.create(missingHost3))))
+                .build());
+    String source = Key.create(domain).toString();
+    runMapreduce();
+    assertIntegrityErrors(
+        IntegrityError.create(source, missingHost1.toString(), "Target entity does not exist"),
+        IntegrityError.create(source, missingHost2.toString(), "Target entity does not exist"),
+        IntegrityError.create(source, missingHost3.toString(), "Target entity does not exist"));
+  }
+
+  @Test
+  public void test_overlappingActivePeriods() throws Exception {
+    ContactResource contact123 = persistActiveContact("contact123");
+    // These two have overlapping active periods because they will have both been created at
+    // START_OF_TIME.
+    DomainResource domain1 =
+        persistDomainAsDeleted(newDomainResource("penny.tld", contact123), now.minusYears(2));
+    DomainResource domain2 = persistActiveDomain("penny.tld");
+    runMapreduce();
+    assertIntegrityErrors(
+        IntegrityError.create(
+            ForeignKeyDomainIndex.createKey(domain2).toString(),
+            Key.create(domain1).toString(),
+            "Found inactive resource deleted more recently than when active resource was created"));
+  }
+
+  @Test
+  public void test_multipleActiveContactsWithSameContactId() throws Exception {
+    ContactResource contact1 = persistActiveContact("dupeid");
+    ContactResource contact2 = persistActiveContact("dupeid");
+    runMapreduce();
+    assertIntegrityErrors(
+        IntegrityError.create(
+            Key.create(ForeignKeyContactIndex.class, "dupeid").toString(),
+            Key.create(contact1).toString(),
+            "Multiple active EppResources with same foreign key"),
+        IntegrityError.create(
+            Key.create(ForeignKeyContactIndex.class, "dupeid").toString(),
+            Key.create(contact2).toString(),
+            "Multiple active EppResources with same foreign key"));
+  }
+
+  /** Encapsulates the data representing a single integrity error. */
+  private static class IntegrityError {
+    String source;
+    String target;
+    String message;
+
+    static IntegrityError create(String source, String target, String message) {
+      IntegrityError instance = new IntegrityError();
+      instance.source = source;
+      instance.target = target;
+      instance.message = message;
+      return instance;
+    }
+
+    /**
+     * Returns a Map representing the JSON blob corresponding to the BigQuery output for this
+     * integrity violation at the given scan time.
+     */
+    Map<String, Object> toMap(DateTime scanTime) {
+      return new ImmutableMap.Builder<String, Object>()
+          .put("scanTime", new com.google.api.client.util.DateTime(scanTime.toDate()))
+          .put("source", source)
+          .put("target", target)
+          .put("message", message)
+          .build();
+    }
+
+    private IntegrityError() {}
+  }
+
+  /** Asserts that the given integrity errors, and no others, were logged to BigQuery. */
+  private void assertIntegrityErrors(IntegrityError... errors) {
+    ImmutableList.Builder<Rows> expected = new ImmutableList.Builder<>();
+    for (IntegrityError error : errors) {
+      expected.add(new Rows().setInsertId("rowid").setJson(error.toMap(now)));
+    }
+    ImmutableList.Builder<Rows> allRows = new ImmutableList.Builder<>();
+    for (TableDataInsertAllRequest req : rowsCaptor.getAllValues()) {
+      allRows.addAll(req.getRows());
+    }
+    assertThat(allRows.build()).containsExactlyElementsIn(expected.build());
   }
 }

javatests/com/google/domain/registry/testing/DatastoreHelper.java

@@ -46,6 +46,7 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.collect.Iterables;
 import com.google.domain.registry.config.RegistryEnvironment;
+import com.google.domain.registry.model.Buildable;
 import com.google.domain.registry.model.EppResource;
 import com.google.domain.registry.model.EppResource.ForeignKeyedEppResource;
 import com.google.domain.registry.model.ImmutableObject;
@@ -710,6 +711,9 @@ public class DatastoreHelper {
   }
 
   private static <R> R persistResource(final R resource, final boolean wantBackup) {
+    assertWithMessage("Attempting to persist a Builder is almost certainly an error in test code")
+        .that(resource)
+        .isNotInstanceOf(Buildable.Builder.class);
     ofy().transact(new VoidWork() {
       @Override
       public void vrun() {