ct-fetch: pipeline cache insertions (#363)

* certdatabase: move cache-modifying SerialCacheWriter methods to CertDatabase

* storage: add SetInsertMany

* storage: use built-in batching for SRem

* ct-fetch: queue set member additions in insertCTWorker

* storage: remove unused cache member from SerialCacheKey

Author: jschanck

go/cmd/ct-fetch/ct-fetch.go

@@ -194,24 +194,20 @@ func (ld *LogSyncEngine) Wait() {
 	ld.ThreadWaitGroup.Wait()
 }
 
-func (ld *LogSyncEngine) tryUpdate(ep *LogSyncMessage) bool {
-	entry := *ep
-
-	if entry.Certificate != nil && entry.Issuer != nil {
-		err := ld.database.Store(entry.Certificate, entry.Issuer)
-		if err != nil {
-			glog.Errorf("Problem inserting certificate (serial=%s): %s", types.NewSerial(entry.Certificate).String(), err)
-			return false
-		}
+func (ld *LogSyncEngine) tryUpdate(queue []storage.SetMemberWithExpiry, logState *types.CTLogState) bool {
+	err := ld.database.Store(queue)
+	if err != nil {
+		glog.Errorf("Problem inserting certificates: %s", err)
+		return false
 	}
 
-	if entry.LogState != nil {
-		err := ld.database.SaveLogState(entry.LogState)
+	if logState != nil {
+		err = ld.database.SaveLogState(logState)
 		if err != nil {
-			glog.Errorf("Problem saving log state (%s): %s", entry.LogState, err)
+			glog.Errorf("Problem saving log state (%s): %s", logState, err)
 			return false
 		}
-		glog.Infof("[%s] Saved log state: %s", entry.LogState.ShortURL, entry.LogState)
+		glog.Infof("[%s] Saved log state: %s", logState.ShortURL, logState)
 	}
 
 	return true
@@ -226,6 +222,9 @@ func (ld *LogSyncEngine) insertCTWorker(ctx context.Context) {
 	healthStatusTicker := time.NewTicker(healthStatusPeriod)
 	defer healthStatusTicker.Stop()
 
+	batchSize := *ctconfig.BatchSize
+	queue := make([]storage.SetMemberWithExpiry, 0, batchSize)
+
 	for ep := range ld.entryChan {
 		select { // Taking something off the queue is useful work.
 		// So indicate server health when requested.
@@ -234,8 +233,20 @@ func (ld *LogSyncEngine) insertCTWorker(ctx context.Context) {
 		default:
 		}
 
+		if ep.Certificate != nil && ep.Issuer != nil {
+			item := ld.database.PrepareSetMember(ep.Certificate, ep.Issuer)
+			queue = append(queue, item)
+		}
+
+		// Only dispatch the queue if it is full or if we're updating
+		// the log state
+		if uint64(len(queue)+1) < batchSize && ep.LogState == nil {
+			continue
+		}
+
 		for {
-			if ld.tryUpdate(&ep) {
+			if ld.tryUpdate(queue, ep.LogState) {
+				queue = queue[:0]
 				break
 			}
 

go/storage/certdatabase.go

@@ -276,14 +276,15 @@ func (db *CertDatabase) GetLogState(aUrl *url.URL) (*types.CTLogState, error) {
 	}, nil
 }
 
-func (db *CertDatabase) Store(aCert *x509.Certificate, aIssuer *x509.Certificate) error {
-	expDate := types.NewExpDateFromTime(aCert.NotAfter)
+func (db *CertDatabase) PrepareSetMember(aCertificate, aIssuer *x509.Certificate) SetMemberWithExpiry {
+	expDate := types.NewExpDateFromTime(aCertificate.NotAfter)
 	issuer := types.NewIssuer(aIssuer)
-	serialWriter := db.GetSerialCacheAccessor(expDate, issuer)
-
-	serial := types.NewSerial(aCert)
+	serial := types.NewSerial(aCertificate)
+	return db.GetSerialCacheKey(expDate, issuer).NewMember(serial)
+}
 
-	_, err := serialWriter.Insert(serial)
+func (db *CertDatabase) Store(items []SetMemberWithExpiry) error {
+	err := db.cache.SetInsertMany(items)
 	if err != nil {
 		return err
 	}
@@ -330,15 +331,15 @@ func (db *CertDatabase) GetCTLogsFromStorage() ([]types.CTLogState, error) {
 	return ctLogList, nil
 }
 
-func (db *CertDatabase) GetSerialCacheAccessor(aExpDate types.ExpDate, aIssuer types.Issuer) *SerialCacheWriter {
-	var kc *SerialCacheWriter
+func (db *CertDatabase) GetSerialCacheKey(aExpDate types.ExpDate, aIssuer types.Issuer) *SerialCacheKey {
+	var kc *SerialCacheKey
 
 	id := aIssuer.ID() + aExpDate.ID()
 
 	cacheObj, err := db.cacheAccessors.GetIFPresent(id)
 	if err != nil {
 		if err == gcache.KeyNotFoundError {
-			kc = NewSerialCacheWriter(aExpDate, aIssuer, db.cache)
+			kc = NewSerialCacheKey(aExpDate, aIssuer)
 			err = db.cacheAccessors.Set(id, kc)
 			if err != nil {
 				glog.Fatalf("Couldn't set into the cache expDate=%s issuer=%s from cache: %s",
@@ -349,7 +350,7 @@ func (db *CertDatabase) GetSerialCacheAccessor(aExpDate types.ExpDate, aIssuer t
 				aExpDate, aIssuer.ID(), err)
 		}
 	} else {
-		kc = cacheObj.(*SerialCacheWriter)
+		kc = cacheObj.(*SerialCacheKey)
 	}
 
 	if kc == nil {
@@ -359,8 +360,7 @@ func (db *CertDatabase) GetSerialCacheAccessor(aExpDate types.ExpDate, aIssuer t
 }
 
 func (db *CertDatabase) ReadSerialsFromCache(aExpDate types.ExpDate, aIssuer types.Issuer) []types.Serial {
-	accessor := db.GetSerialCacheAccessor(aExpDate, aIssuer)
-	return accessor.List()
+	return db.List(db.GetSerialCacheKey(aExpDate, aIssuer))
 }
 
 func (db *CertDatabase) ReadSerialsFromStorage(aExpDate types.ExpDate, aIssuer types.Issuer) ([]types.Serial, error) {
@@ -434,8 +434,8 @@ func (db *CertDatabase) moveOneBinOfCachedSerialsToStorage(aTmpDir string, aExpD
 	}
 
 	// It's now safe to remove cachedSerials from the cache.
-	cacheWriter := db.GetSerialCacheAccessor(aExpDate, aIssuer)
-	err = cacheWriter.RemoveMany(cachedSerials)
+	key := db.GetSerialCacheKey(aExpDate, aIssuer)
+	err = db.RemoveMany(key, cachedSerials)
 	if err != nil {
 		glog.Warningf("Failed to remove serial from cache: %s", err)
 	}
@@ -698,3 +698,68 @@ func (db *CertDatabase) Commit(aProofOfLock string) error {
 func (db *CertDatabase) AddPreIssuerAlias(aPreIssuer types.Issuer, aIssuer types.Issuer) error {
 	return db.cache.AddPreIssuerAlias(aPreIssuer, aIssuer)
 }
+
+// Returns true if this serial was unknown. Subsequent calls with the same serial
+// will return false, as it will be known then.
+func (db *CertDatabase) Insert(k *SerialCacheKey, aSerial types.Serial) (bool, error) {
+	result, err := db.cache.SetInsert(k.ID(), aSerial.BinaryString())
+	if err != nil {
+		return false, err
+	}
+
+	if !k.expirySet {
+		expireTime := k.expDate.ExpireTime()
+		if err := db.cache.ExpireAt(k.ID(), expireTime); err != nil {
+			glog.Errorf("Couldn't set expiration time %v for serials %s: %v", expireTime, k.ID(), err)
+		} else {
+			k.expirySet = true
+		}
+	}
+
+	return result, nil
+}
+
+func (db *CertDatabase) RemoveMany(k *SerialCacheKey, aSerials []types.Serial) error {
+	// Removing an element of a set may leave the set empty. Redis
+	// automatically deletes empty sets, so assume that we need to reset
+	// the ExpireAt time for this set on the next Insert call.
+	k.expirySet = false
+	serialStrings := make([]string, len(aSerials))
+	for i := 0; i < len(aSerials); i++ {
+		serialStrings[i] = aSerials[i].BinaryString()
+	}
+	return db.cache.SetRemove(k.ID(), serialStrings)
+}
+
+func (db *CertDatabase) List(k *SerialCacheKey) []types.Serial {
+	// Redis' scan methods regularly provide duplicates. The duplication
+	// happens at this level, pulling from SetToChan, so we make a hash-set
+	// here to de-duplicate when the memory impacts are the most minimal.
+	serials := make(map[string]struct{})
+	var count int
+
+	strChan := make(chan string)
+	go func() {
+		err := db.cache.SetToChan(k.ID(), strChan)
+		if err != nil {
+			glog.Fatalf("Error obtaining list of known certificates: %v", err)
+		}
+	}()
+
+	for str := range strChan {
+		serials[str] = struct{}{}
+		count += 1
+	}
+
+	serialList := make([]types.Serial, 0, count)
+	for str := range serials {
+		bs, err := types.NewSerialFromBinaryString(str)
+		if err != nil {
+			glog.Errorf("Failed to populate serial str=[%s] %v", str, err)
+			continue
+		}
+		serialList = append(serialList, bs)
+	}
+
+	return serialList
+}

go/storage/certdatabase_test.go

@@ -42,8 +42,7 @@ func cacheSerial(t *testing.T, db CertDatabase, expDateStr string, issuerStr str
 	}
 	serial := types.NewSerialFromHex(serialStr)
 
-	kc := db.GetSerialCacheAccessor(expDate, issuer)
-	_, err = kc.Insert(serial)
+	_, err = db.Insert(db.GetSerialCacheKey(expDate, issuer), serial)
 	if err != nil {
 		t.Error(err)
 	}
@@ -57,8 +56,9 @@ func isCached(t *testing.T, db CertDatabase, expDateStr string, issuerStr string
 	}
 	serial := types.NewSerialFromHex(serialStr)
 
-	kc := db.GetSerialCacheAccessor(expDate, issuer)
-	cached, err := kc.Contains(serial)
+	kc := db.GetSerialCacheKey(expDate, issuer)
+
+	cached, err := db.cache.SetContains(kc.ID(), serial.BinaryString())
 	if err != nil {
 		t.Error(err)
 	}

go/storage/mockcache.go

@@ -70,6 +70,16 @@ func (ec *MockRemoteCache) SetInsert(key string, entry string) (bool, error) {
 	return true, nil
 }
 
+func (ec *MockRemoteCache) SetInsertMany(items []SetMemberWithExpiry) error {
+	for _, item := range items {
+		_, err := ec.SetInsert(item.Key, item.Value)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (ec *MockRemoteCache) setRemove(key string, entry string) error {
 	ec.mu.Lock()
 	defer ec.mu.Unlock()

go/storage/rediscache.go

@@ -80,6 +80,26 @@ func (rc *RedisCache) SetInsert(key string, entry string) (bool, error) {
 	return added == 1, err
 }
 
+func (rc *RedisCache) SetInsertMany(items []SetMemberWithExpiry) error {
+	_, err := rc.client.Pipelined(func(pipe redis.Pipeliner) error {
+		for _, item := range items {
+			err := pipe.SAdd(item.Key, item.Value).Err()
+			if err != nil {
+				return err
+			}
+			err = pipe.ExpireAt(item.Key, item.Expiry).Err()
+			if err != nil {
+				glog.Errorf("Couldn't set expiration time %v for serials %s: %v", item.Expiry, item.Key, err)
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func (rc *RedisCache) SetRemove(key string, entries []string) error {
 	batchSize := 1024
 	for batchStart := 0; batchStart < len(entries); batchStart += batchSize {
@@ -88,15 +108,7 @@ func (rc *RedisCache) SetRemove(key string, entries []string) error {
 			batchEnd = len(entries)
 		}
 		batch := entries[batchStart:batchEnd]
-		_, err := rc.client.Pipelined(func(pipe redis.Pipeliner) error {
-			for _, entry := range batch {
-				err := pipe.SRem(key, entry).Err()
-				if err != nil {
-					return err
-				}
-			}
-			return nil
-		})
+		err := rc.client.SRem(key, batch).Err()
 		if err != nil {
 			return err
 		}