LogDog: Handle archive failures.

appengine/logdog/coordinator/endpoints/services/archiveStream.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 services
 
 import (
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         "github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/grpcutil"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/google"
         "golang.org/x/net/context"
         "google.golang.org/grpc/codes"
 )
 
+// maxArchiveErrors is the maximum number of archival errors that we will
+// tolerate before accepting a failed archival.
+const maxArchiveErrors = 2
+
 // ArchiveStream implements the logdog.ServicesServer interface.
 func (b *Server) ArchiveStream(c context.Context, req *logdog.ArchiveStreamRequest) (*google.Empty, error) {
         if err := Auth(c); err != nil {
                 return nil, err
         }
 
         log.Fields{
                 "path": req.Path,
         }.Infof(c, "Marking log stream as archived.")
 
@@ skipping 23 matching lines @@
         case ds.ErrNoSuchEntity:
                 break
 
         default:
                 log.WithError(err).Errorf(c, "Failed to check for log stream archvial state.")
                 return nil, grpcutil.Internal
         }
 
         // Post the archival results to the Coordinator.
         now := clock.Now(c).UTC()
+        var ierr error
         err := ds.Get(c).RunInTransaction(func(c context.Context) error {
                 di := ds.Get(c)
                 if err := di.Get(ls); err != nil {
                         return err
                 }
                 if ls.Archived() {
                         log.Infof(c, "Log stream already marked as archived.")
                         return nil
                 }
 
+                // If this request contained an error, we will reject it with
+                // FailedPrecondition status if we are below our error threshold.
+                if req.Error {
+                        if ls.ArchiveErrors < maxArchiveErrors {
+                                log.Fields{
+                                        "path":             req.Path,
+                                        "errorCount":       ls.ArchiveErrors,
+                                        "maxArchiveErrors": maxArchiveErrors,
+                                }.Warningf(c, "Rejecting failed archival: below error threshold.")
+
+                                // Increment our error count.
+                                ls.ArchiveErrors++
+                                if err := ls.Put(di); err != nil {
+                                        log.WithError(err).Errorf(c, "Failed to update log stream error count.")
+                                        return err
+                                }
+
+                                // Fail this RPC call to keep the archival task in the queue.
+                                ierr = grpcutil.Errf(codes.FailedPrecondition, "below error threshold (%d < %d)", ls.ArchiveErrors, maxArchiveErrors)
+                                return nil
+                        }
+
+                        // We have exceeded our error threshold, so we will continue to archive
+                        // this log stream.
+                        log.Fields{
+                                "path":             req.Path,
+                                "errorCount":       ls.ArchiveErrors,
+                                "maxArchiveErrors": maxArchiveErrors,
+                        }.Warningf(c, "Log stream has exceeded archive error threshold.")
+                }
+
                 // Update archival information. Make sure this actually marks the stream as
                 // archived.
                 ls.Updated = now
                 ls.State = coordinator.LSArchived
                 ls.ArchiveWhole = req.Complete
                 ls.TerminalIndex = req.TerminalIndex
                 ls.ArchiveStreamURL = req.StreamUrl
                 ls.ArchiveStreamSize = req.StreamSize
                 ls.ArchiveIndexURL = req.IndexUrl
                 ls.ArchiveIndexSize = req.IndexSize
                 ls.ArchiveDataURL = req.DataUrl
                 ls.ArchiveDataSize = req.DataSize
 
+                // Determine archive state.
+                switch {
+                case req.Error:
+                        ls.ArchiveState = coordinator.ArchivedWithErrors
+                case !req.Complete:
+                        ls.ArchiveState = coordinator.ArchivedPartially
+                default:
+                        ls.ArchiveState = coordinator.Archived
+                }
+
                 // Update the log stream.
                 if err := ls.Put(di); err != nil {
                         log.WithError(err).Errorf(c, "Failed to update log stream.")
                         return err
                 }
 
                 log.Infof(c, "Successfully marked stream as archived.")
                 return nil
         }, nil)
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to mark stream as archived.")
                 return nil, grpcutil.Internal
         }
+        if ierr != nil {
+                log.WithError(ierr).Errorf(c, "Failed to mark stream as archived (inner).")
+                return nil, ierr
+        }
 
         return &google.Empty{}, nil
 }