Process /api/v2/remove-ineligible-domains in shards (#279)

Processing all domains at once is failing to complete. Adding query
parameters to the endpoint to have it process a subset of the domains on
each run should result in each run using less resources and taking less
time. The cron configuration is changed to split the work into 4
approximately evenly sized shards.

Author: nharper

api/domain_handlers.go

@@ -404,7 +404,7 @@ func (api API) Remove(w http.ResponseWriter, r *http.Request) {
 
 type DomainStateWithIssues struct {
 	DomainState database.DomainState
-	Issues hstspreload.Issues
+	Issues      hstspreload.Issues
 }
 
 // RemoveIneligibleDomains runs eligibility checks on domains present in the
@@ -429,14 +429,22 @@ func (api API) RemoveIneligibleDomains(w http.ResponseWriter, r *http.Request) {
 	var deleteEligibleDomains []string
 
 	api.logger.Print("Fetching domains...")
-	// Get all domains
-	domains, err := api.database.AllDomainStates()
+	var start, end string
+	if s, ok := r.URL.Query()["start"]; ok && len(s) > 0 {
+		start = s[0]
+	}
+	if e, ok := r.URL.Query()["end"]; ok && len(e) > 0 {
+		end = e[0]
+	}
+	// Get domains
+	api.logger.Printf("using start %q, end %q", start, end)
+	domains, err := api.database.DomainStatesInRange(start, end)
 	if err != nil {
 		msg := fmt.Sprintf("Internal error: could not retrieve domains. (%s)\n", err)
 		http.Error(w, msg, http.StatusInternalServerError)
 		return
 	}
-	api.logger.Print("Filtering domains...")
+	api.logger.Printf("Filtering %d domains...", len(domains))
 
 	// Filter Domains
 	for _, d := range domains {
@@ -446,7 +454,7 @@ func (api API) RemoveIneligibleDomains(w http.ResponseWriter, r *http.Request) {
 	}
 	api.logger.Print("Getting ineligible domain states...")
 
-	// call GetIneligibleDomainStates, add to map
+	// call GetIneligibleDomainStates and store them in a map by domain name
 	states := make(map[string]database.IneligibleDomainState)
 	state, err := api.database.GetAllIneligibleDomainStates()
 	if err != nil {
@@ -455,10 +463,15 @@ func (api API) RemoveIneligibleDomains(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// delete domains that exist in the ineligible database but not
-	// on the preload list
 	for _, s := range state {
+		// ignore IneligibleDomainStates for domain names not in the [start, end)
+		// range we're processing.
+		if (start != "" && s.Name < start) || (end != "" && s.Name >= end) {
+			continue
+		}
 		states[s.Name] = s
+		// Delete IneligibleDomainStates for names that are no longer on the
+		// preload list.
 		if _, ok := policyStates[s.Name]; !ok {
 			deleteEligibleDomains = append(deleteEligibleDomains, s.Name)
 		}
@@ -488,7 +501,7 @@ func (api API) RemoveIneligibleDomains(w http.ResponseWriter, r *http.Request) {
 			i++
 			wg.Add(1)
 			domainStates <- d
-			if i % 1000 == 0 {
+			if i%1000 == 0 {
 				api.logger.Printf("Sent %d domains to workers\n", i)
 			}
 		}

api/domain_handlers_test.go

@@ -363,3 +363,130 @@ func TestDeletionAndStatusChange(t *testing.T) {
 		t.Errorf("Status changed")
 	}
 }
+
+func TestRemoveIneligibleDomainsSharding(t *testing.T) {
+	api, _, mockHstspreload, mockPreloadlist := mockAPI(0 * time.Second)
+
+	testPreloadlist := preloadlist.PreloadList{Entries: []preloadlist.Entry{
+		{Name: "a.test", Mode: preloadlist.ForceHTTPS, IncludeSubDomains: true, Policy: preloadlist.Bulk1Year},
+		{Name: "n.test", Mode: preloadlist.ForceHTTPS, IncludeSubDomains: true, Policy: preloadlist.Bulk1Year},
+		{Name: "z.test", Mode: preloadlist.ForceHTTPS, IncludeSubDomains: true, Policy: preloadlist.Bulk1Year},
+	}}
+	testEligibleResponses := map[string]hstspreload.Issues{
+		"a.test": issuesWithErrors,
+		"n.test": issuesWithErrors,
+		"z.test": issuesWithErrors,
+	}
+	mockPreloadlist.list = testPreloadlist
+	mockHstspreload.eligibleResponses = testEligibleResponses
+
+	w := httptest.NewRecorder()
+	w.Body = &bytes.Buffer{}
+
+	r, err := http.NewRequest("GET", "", nil)
+	if err != nil {
+		t.Fatalf("[%s] %s", "NewRequest Failed", err)
+	}
+
+	api.Update(w, r)
+
+	// These test cases are structured to be run in this specific order and
+	// each case depends on the behavior of the previous ones.
+	tests := []struct {
+		name           string
+		query          string
+		expectedCounts map[string]int
+	}{
+		{
+			// Start by running RemoveIneligibleDomains with no query
+			// parameters - it should process every domain.
+			"no range specified",
+			"",
+			map[string]int{
+				"a.test": 1,
+				"n.test": 1,
+				"z.test": 1,
+			},
+		},
+		{
+			// Specifying an end of "n" (the [start, end) interval is half-open)
+			// should result in only a.test being processed. Every time a domain
+			// is processed, the number of scans in its IneligibleDomainState
+			// increases.
+			"query range only has end",
+			"end=n",
+			map[string]int{
+				"a.test": 2,
+				"n.test": 1,
+				"z.test": 1,
+			},
+		},
+		{
+			// With an interval of ["n","z"), only n.test should match.
+			"start and end",
+			"start=n&end=z",
+			map[string]int{
+				"a.test": 2,
+				"n.test": 2,
+				"z.test": 1,
+			},
+		},
+		{
+			// A start of "z" with no end should only match z.test from the
+			// test preload list.
+			"only start",
+			"start=z",
+			map[string]int{
+				"a.test": 2,
+				"n.test": 2,
+				"z.test": 2,
+			},
+		},
+		{
+			// A bad range (start after end) does nothing.
+			"bad range",
+			"start=b&end=a",
+			map[string]int{
+				"a.test": 2,
+				"n.test": 2,
+				"z.test": 2,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		// Make request with the specified query
+		r, err := http.NewRequest("GET", "", nil)
+		if err != nil {
+			t.Fatalf("[%s] %s", "NewRequest Failed", err)
+		}
+		r = toAppEngineHttpRequest(r)
+		r.URL.RawQuery = test.query
+		w := httptest.NewRecorder()
+		w.Body = &bytes.Buffer{}
+		api.RemoveIneligibleDomains(w, r)
+
+		// Look at the IneligibleDomainStates created or updated by
+		// RemoveIneligibleDomains and check that the number of scans for
+		// each domain matches the expected count.
+		states, err := api.database.GetAllIneligibleDomainStates()
+		if err != nil {
+			t.Fatalf("Couldn't get the states of all domains in the database.")
+		}
+		seenNames := make(map[string]bool)
+		for _, state := range states {
+			expectedCount, found := test.expectedCounts[state.Name]
+			if !found {
+				t.Errorf("[%s] found unexpected domain %q in IneligibleDomainStates list", test.name, state.Name)
+				continue
+			}
+			if len(state.Scans) != expectedCount {
+				t.Errorf("[%s] Unexpected number of scans for domain %q: got %d, want %d", test.name, state.Name, len(state.Scans), expectedCount)
+			}
+			seenNames[state.Name] = true
+		}
+		if len(seenNames) != len(test.expectedCounts) {
+			t.Errorf("[%s] Wrong number of IneligibleDomainStates: got %d, want %d", test.name, len(seenNames), len(test.expectedCounts))
+		}
+	}
+}

cron.yaml

@@ -1,6 +1,17 @@
 cron: 
-- description: "Removes ineligible domains"
-  url: "/api/v2/remove-ineligible-domains"
-  schedule: every monday 09:00
+- description: "Remove ineligible domains ['','e')"
+  url: "/api/v2/remove-ineligible-domains?end=e"
+  schedule: every monday 9:00
+  timezone: America/New_York
+- description: "Remove ineligible domains ['e','l')"
+  url: "/api/v2/remove-ineligible-domains?start=e&end=l"
+  schedule: every monday 11:00
+  timezone: America/New_York
+- description: "Remove ineligible domains ['l','s')"
+  url: "/api/v2/remove-ineligible-domains?start=l&end=s"
+  schedule: every monday 13:00
+  timezone: America/New_York
+- description: "Remove ineligible domains ['s','')"
+  url: "/api/v2/remove-ineligible-domains?start=s"
+  schedule: every monday 15:00
   timezone: America/New_York
-

database/database.go

@@ -27,6 +27,7 @@ type Database interface {
 	StateForDomain(string) (DomainState, error)
 	StatesForDomains([]string) ([]DomainState, error)
 	AllDomainStates() ([]DomainState, error)
+	DomainStatesInRange(start, end string) ([]DomainState, error)
 	StatesWithStatus(PreloadStatus) ([]DomainState, error)
 	GetIneligibleDomainStates(domains []string) (states []IneligibleDomainState, err error)
 	SetIneligibleDomainStates(updates []IneligibleDomainState, logf func(format string, args ...interface{})) error
@@ -208,6 +209,17 @@ func (db DatastoreBacked) AllDomainStates() (states []DomainState, err error) {
 	return db.statesForQuery(datastore.NewQuery("DomainState"))
 }
 
+func (db DatastoreBacked) DomainStatesInRange(start, end string) ([]DomainState, error) {
+	query := datastore.NewQuery(domainStateKind)
+	if start != "" {
+		query = query.FilterField("__key__", ">=", datastore.NameKey(domainStateKind, start, nil))
+	}
+	if end != "" {
+		query = query.FilterField("__key__", "<", datastore.NameKey(domainStateKind, end, nil))
+	}
+	return db.statesForQuery(query)
+}
+
 // StatesWithStatus returns the states of domains with the given status in the database.
 func (db DatastoreBacked) StatesWithStatus(status PreloadStatus) (domains []DomainState, err error) {
 	return db.statesForQuery(

database/database_test.go

@@ -276,6 +276,100 @@ func TestStatesWithStatus(t *testing.T) {
 	}
 }
 
+func TestDomainStatesInRange(t *testing.T) {
+	resetDB()
+
+	testNames := []string{
+		"1.test",
+		"a.test",
+		"aaa.test",
+		"b.test",
+		"g.test",
+		"h.test",
+		"y.test",
+		"z.test",
+	}
+	for _, name := range testNames {
+		state := DomainState{Name: name}
+		if err := testDB.PutState(state); err != nil {
+			t.Fatalf("failed to set test state for TestDomainStatesInRange: %v", err)
+		}
+	}
+
+	tests := []struct {
+		name          string
+		start         string
+		end           string
+		expectedNames []string
+	}{
+		{
+			"empty start and end",
+			"",
+			"",
+			testNames,
+		},
+		{
+			"empty start",
+			"",
+			"a",
+			[]string{"1.test"},
+		},
+		{
+			"empty end",
+			"z",
+			"",
+			[]string{"z.test"},
+		},
+		{
+			"intervals are half-open",
+			"a",
+			"b",
+			[]string{"a.test", "aaa.test"},
+		},
+		{
+			"larger interval",
+			"a",
+			"h",
+			[]string{"a.test", "aaa.test", "b.test", "g.test"},
+		},
+		{
+			"same non-empty start and end returns empty list",
+			"a.test",
+			"a.test",
+			nil,
+		},
+		{
+			"start after end returns empty list",
+			"b",
+			"a",
+			nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			states, err := testDB.DomainStatesInRange(test.start, test.end)
+			if err != nil {
+				t.Fatalf("DomainStatesInRange unexpectedly failed: %v", err)
+			}
+			// check that the names in the returned states match the ones in test.expectedNames
+			nameSet := make(map[string]bool)
+			for _, state := range states {
+				nameSet[state.Name] = true
+			}
+			for _, name := range test.expectedNames {
+				if !nameSet[name] {
+					t.Errorf("%q was not returned by DomainStatesInRange but should have been", name)
+				}
+				delete(nameSet, name)
+			}
+			for name := range nameSet {
+				t.Errorf("%q was returned by DomainStatesInRange but not expected", name)
+			}
+		})
+	}
+}
+
 func TestSetPendingAutomatedRemoval(t *testing.T) {
 	resetDB()
 

database/gcd/gcd.go

@@ -3,6 +3,7 @@ package gcd
 import (
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"os"
@@ -102,6 +103,7 @@ func NewLocalBackend() (db LocalBackend, shutdown func() error, err error) {
 		"--testing",
 	)
 	db.cmd = cmd
+	stderr, _ := cmd.StderrPipe()
 
 	err = cmd.Start()
 	if err != nil {
@@ -127,6 +129,19 @@ func NewLocalBackend() (db LocalBackend, shutdown func() error, err error) {
 		}
 	}
 
+	// try to read some from stderr to see if we can get a more useful error message:
+	if stderr != nil {
+		msg := make([]byte, 2048)
+		n, err := stderr.Read(msg)
+		if err != nil && err != io.EOF {
+			return db, shutdown, fmt.Errorf("failed to connect, failure reading stderr: %v", err)
+		}
+		msg = msg[:n]
+		if strings.Contains(string(msg), "Unable to locate a Java Runtime.") {
+			return db, shutdown, fmt.Errorf("failed to connect, unable to locate Java runtime to run datastore emulator")
+		}
+	}
+
 	return db, shutdown, fmt.Errorf("could not connect")
 }