Iterate archive query alongside task queue.

appengine/logdog/coordinator/backend/archiveCron.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package backend
 
 import (
-        "errors"
         "fmt"
         "net/http"
         "time"
 
         "github.com/julienschmidt/httprouter"
         ds "github.com/luci/gae/service/datastore"
         tq "github.com/luci/gae/service/taskqueue"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         "github.com/luci/luci-go/appengine/logdog/coordinator/config"
         "github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
         "github.com/luci/luci-go/common/clock"
+        "github.com/luci/luci-go/common/errors"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/logdog/svcconfig"
         "golang.org/x/net/context"
 )
 
 // archiveTaskQueueName returns the task queue name for archival, or an error
 // if it's not configured.
 func archiveTaskQueueName(cfg *svcconfig.Config) (string, error) {
         q := cfg.GetCoordinator().ArchiveTaskQueue
         if q == "" {
@@ skipping 71 matching lines @@
                 log.WithError(err).Errorf(c, "Failed to load configuration.")
                 return err
         }
 
         queueName, err := archiveTaskQueueName(cfg)
         if err != nil {
                 log.Errorf(c, "Failed to get task queue name.")
                 return err
         }
 
+        // If the log stream has a terminal index, and its Updated time is less than
+        // the maximum archive delay, require this archival to be complete (no
+        // missing LogEntry).
+        //
+        // If we're past maximum archive delay, settle for any (even empty) archival.
+        // This is a failsafe to prevent logs from sitting in limbo forever.
+        maxDelay := cfg.GetCoordinator().ArchiveDelayMax.Duration()
+
         now := clock.Now(c).UTC()
         q := ds.NewQuery("LogStream")
 
         var threshold time.Duration
         if complete {
                 threshold = cfg.GetCoordinator().ArchiveDelay.Duration()
                 q = q.Eq("State", coordinator.LSTerminated)
         } else {
                 threshold = cfg.GetCoordinator().ArchiveDelayMax.Duration()
                 q = q.Eq("State", coordinator.LSPending)
         }
         q = q.Lte("Updated", now.Add(-threshold))
 
-        // Query and dispatch our tasks.
-        var ierr error
-        count, err := b.multiTask(c, queueName, func(taskC chan<- *tq.Task) {
-                ierr = ds.Get(c).Run(q, func(ls *coordinator.LogStream) error {
+        // We will enqueue tasks in batches, and there's no need to retrieve more than
+        // our batch size.
+        batch := b.getMultiTaskBatchSize()
+        tasks := make([]*tq.Task, 0, batch)
+        q = q.Limit(int32(batch))
+
+        // Perform an iterative query, dispatching tasks with each round.
+        di := ds.Get(c)
+        ti := tq.Get(c)
+
+        totalScheduledTasks := 0
+
+        var next ds.Cursor
+        for {
+                iterQuery := q
+                if next != nil {
+                        iterQuery = iterQuery.Start(next)
+                        next = nil
+                }
+
+                err = di.Run(iterQuery, func(ls *coordinator.LogStream, cb ds.CursorCB) error {

[iannucci, 2016/03/30 19:04:27]

I would just loop in the query and dispatch the tasks from inside the query body
as batches.

[dnj, 2016/03/30 19:10:46]

Done.

                         log.Fields{
                                 "id":      ls.HashID(),
                                 "updated": ls.Updated.String(),
                         }.Infof(c, "Identified log stream ready for archival.")
 
-                        // If the log stream has a terminal index, and its Updated time is less than
-                        // the maximum archive delay, require this archival to be complete (no
-                        // missing LogEntry).
-                        //
-                        // If we're past maximum archive delay, settle for any (even empty) archival.
-                        // This is a failsafe to prevent logs from sitting in limbo forever.
-                        maxDelay := cfg.GetCoordinator().ArchiveDelayMax.Duration()
                         requireComplete := !now.After(ls.Updated.Add(maxDelay))
                         if !requireComplete {
                                 log.Fields{
                                         "path":             ls.Path(),
                                         "updatedTimestamp": ls.Updated,
                                         "maxDelay":         maxDelay,
                                 }.Warningf(c, "Log stream is past maximum archival delay. Dropping completeness requirement.")
                         }
 
                         task, err := createArchiveTask(cfg.GetCoordinator(), ls, requireComplete)
                         if err != nil {
                                 log.Fields{
                                         log.ErrorKey: err,
                                         "path":       ls.Path(),
                                 }.Errorf(c, "Failed to create archive task.")
                                 return err
                         }
 
-                        taskC <- task
+                        tasks = append(tasks, task)
+
+                        // If we're at or over our batch size, break and enqueue the tasks.
+                        if len(tasks) >= batch {
+                                next, err = cb()
+                                if err != nil {
+                                        return fmt.Errorf("failed to get cursor: %v", err)
+                                }
+                                return ds.Stop
+                        }
                         return nil
                 })
-        })
-        if err != nil || ierr != nil {
-                log.Fields{
-                        log.ErrorKey: err,
-                        "queryErr":   ierr,
-                        "taskCount":  count,
-                }.Errorf(c, "Failed to dispatch archival tasks.")
-                return errors.New("failed to dispatch archival tasks")
+                if err != nil {
+                        log.Fields{
+                                log.ErrorKey:         err,
+                                "scheduledTaskCount": totalScheduledTasks,
+                        }.Errorf(c, "Failed to query for archival tasks.")
+                        return errors.New("failed to dispatch archival tasks")
+                }
+
+                // Dispatch the accumulated tasks.
+                if len(tasks) > 0 {
+                        if err := errors.Filter(ti.AddMulti(tasks, queueName), tq.ErrTaskAlreadyAdded); err != nil {
+                                log.Fields{
+                                        log.ErrorKey:         err,
+                                        "queue":              queueName,
+                                        "numTasks":           len(tasks),
+                                        "scheduledTaskCount": totalScheduledTasks,
+                                }.Errorf(c, "Failed to add tasks to task queue.")
+                                return errors.New("failed to add tasks to task queue")
+                        }
+
+                        totalScheduledTasks += len(tasks)
+                        tasks = tasks[:0]
+                }
+
+                // If there is no query cursor, we're done.
+                if next == nil {
+                        break
+                }
         }
 
         log.Fields{
-                "taskCount": count,
+                "scheduledTaskCount": totalScheduledTasks,
         }.Debugf(c, "Archive sweep completed successfully.")
         return nil
 }