initial port

scanner/fetcher.go

@@ -130,7 +130,7 @@ func (f *Fetcher) Prepare(ctx context.Context) (*ct.SignedTreeHead, error) {
 // Run performs fetching of the Log. Blocks until scanning is complete, the
 // passed in context is canceled, or Stop is called (and pending work is
 // finished). For each successfully fetched batch, runs the fn callback.
-func (f *Fetcher) Run(ctx context.Context, fn func(EntryBatch)) error {
+func (f *Fetcher) Run(ctx context.Context, fn func(EntryBatch), maxNewEntries func() uint64) error {
 	klog.V(1).Infof("%s: Starting up Fetcher...", f.uri)
 	if _, err := f.Prepare(ctx); err != nil {
 		return err
@@ -146,7 +146,7 @@ func (f *Fetcher) Run(ctx context.Context, fn func(EntryBatch)) error {
 	// Use a separately-cancelable context for the range generator, so we can
 	// close it down (in Stop) but still let the fetchers below run to
 	// completion.
-	ranges := f.genRanges(cctx)
+	ranges := f.genRanges(cctx, maxNewEntries)
 
 	// Run fetcher workers.
 	var wg sync.WaitGroup
@@ -177,7 +177,7 @@ func (f *Fetcher) Stop() {
 // genRanges returns a channel of ranges to fetch, and starts a goroutine that
 // sends things down this channel. The goroutine terminates when all ranges
 // have been generated, or if context is cancelled.
-func (f *Fetcher) genRanges(ctx context.Context) <-chan fetchRange {
+func (f *Fetcher) genRanges(ctx context.Context, maxNewEntries func() uint64) <-chan fetchRange {
 	batch := int64(f.opts.BatchSize)
 	ranges := make(chan fetchRange)
 
@@ -198,7 +198,7 @@ func (f *Fetcher) genRanges(ctx context.Context) <-chan fetchRange {
 				end = f.opts.EndIndex
 			}
 
-			batchEnd := start + min(end-start, batch)
+			batchEnd := start + min(min(end-start, batch), int64(maxNewEntries()))
 			next := fetchRange{start, batchEnd - 1}
 			select {
 			case <-ctx.Done():

scanner/scanlog/scanlog.go

@@ -33,6 +33,7 @@ import (
 	"github.com/google/certificate-transparency-go/jsonclient"
 	"github.com/google/certificate-transparency-go/scanner"
 	"github.com/google/certificate-transparency-go/x509"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -184,6 +185,7 @@ func createMatcherFromFlags(logClient *client.LogClient) (interface{}, error) {
 }
 
 func main() {
+	klog.InitFlags(nil)
 	flag.Parse()
 
 	logClient, err := client.New(*logURI, &http.Client{
@@ -219,12 +221,15 @@ func main() {
 	s := scanner.NewScanner(logClient, opts)
 
 	ctx := context.Background()
+	maxNewEntries := func() uint64 {
+		return ^uint64(0)
+	}
 	if *printChains {
-		if err := s.Scan(ctx, logFullChain, logFullChain); err != nil {
+		if err := s.Scan(ctx, logFullChain, logFullChain, maxNewEntries); err != nil {
 			log.Fatal(err)
 		}
 	} else {
-		if err := s.Scan(ctx, logCertInfo, logPrecertInfo); err != nil {
+		if err := s.Scan(ctx, logCertInfo, logPrecertInfo, maxNewEntries); err != nil {
 			log.Fatal(err)
 		}
 	}

scanner/scanner.go

@@ -251,13 +251,14 @@ func (s *Scanner) logThroughput(treeSize int64, stop <-chan bool) {
 // LogEntry, which includes the index of the entry and the certificate.
 // For each precert found, calls foundPrecert with the corresponding LogEntry,
 // which includes the index of the entry and the precert.
-func (s *Scanner) Scan(ctx context.Context, foundCert func(*ct.RawLogEntry), foundPrecert func(*ct.RawLogEntry)) error {
-	_, err := s.ScanLog(ctx, foundCert, foundPrecert)
+func (s *Scanner) Scan(ctx context.Context, foundCert func(*ct.RawLogEntry), foundPrecert func(*ct.RawLogEntry), maxNewEntries func() uint64) error {
+	fmt.Println("HHHSDFSDFSADFASDFAS")
+	_, err := s.ScanLog(ctx, foundCert, foundPrecert, maxNewEntries)
 	return err
 }
 
 // ScanLog performs a scan against the Log, returning the count of scanned entries.
-func (s *Scanner) ScanLog(ctx context.Context, foundCert func(*ct.RawLogEntry), foundPrecert func(*ct.RawLogEntry)) (int64, error) {
+func (s *Scanner) ScanLog(ctx context.Context, foundCert func(*ct.RawLogEntry), foundPrecert func(*ct.RawLogEntry), maxNewEntries func() uint64) (int64, error) {
 	klog.V(1).Infof("Starting up Scanner...")
 	s.certsProcessed = 0
 	s.certsMatched = 0
@@ -296,7 +297,7 @@ func (s *Scanner) ScanLog(ctx context.Context, foundCert func(*ct.RawLogEntry),
 			entries <- entryInfo{index: b.Start + int64(i), entry: e}
 		}
 	}
-	err = s.fetcher.Run(ctx, flatten)
+	err = s.fetcher.Run(ctx, flatten, maxNewEntries)
 	close(entries) // Causes matcher workers to terminate.
 	wg.Wait()      // Wait until they terminate.
 	if err != nil {

scanner/scanner_test.go

@@ -198,6 +198,9 @@ func TestScannerEndToEnd(t *testing.T) {
 	var matchedPrecerts list.List
 
 	ctx := context.Background()
+	maxNewEntries := func() uint64 {
+		return ^uint64(0)
+	}
 	err = scanner.Scan(ctx, func(re *ct.RawLogEntry) {
 		// Annoyingly we can't t.Fatal() in here, as this is run in another go
 		// routine
@@ -212,7 +215,7 @@ func TestScannerEndToEnd(t *testing.T) {
 			return
 		}
 		matchedPrecerts.PushBack(*e.Precert)
-	})
+	}, maxNewEntries)
 
 	if err != nil {
 		t.Fatal(err)