LogDog: Update to archival V2.

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/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/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"
 )
 
 // 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
         }
 
+        cfg, err := config.Load(c)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to load configuration.")
+                return nil, grpcutil.Internal
+        }
+        ccfg := cfg.GetCoordinator() // config.Load ensures this is not nil.
+
         log.Fields{
-                "path": req.Path,
-        }.Infof(c, "Marking log stream as archived.")
+                "path":          req.Path,
+                "complete":      req.Complete(),
+                "terminalIndex": req.TerminalIndex,
+                "logEntryCount": req.LogEntryCount,
+                "error":         req.Error,
+        }.Infof(c, "Received archival request.")
 
         // Verify that the request is minimially valid.
         path := types.StreamPath(req.Path)
         if err := path.Validate(); err != nil {
                 return nil, grpcutil.Errf(codes.InvalidArgument, "invalid log stream path: %v", err)
         }
 
         switch {
         case req.IndexUrl == "":
                 return nil, grpcutil.Errf(codes.InvalidArgument, "missing required index archive URL")
         case req.StreamUrl == "":
                 return nil, grpcutil.Errf(codes.InvalidArgument, "missing required stream archive URL")
         }
 
         ls := coordinator.LogStreamFromPath(path)
 
-        // (Non-transactional) Is the log stream already archived?
-        switch err := ds.Get(c).Get(ls); err {
-        case nil:
-                if ls.Archived() {
-                        log.Infof(c, "Log stream already marked as archived (non-transactional).")
-                        return &google.Empty{}, nil
-                }
-
-        case ds.ErrNoSuchEntity:
-                break
-
-        default:
-                log.WithError(err).Errorf(c, "Failed to check for log stream archvial state.")
-                return nil, grpcutil.Internal
-        }
+        log.Fields{
+                "id": ls.HashID(),
+        }.Infof(c, "Log stream ID.")
 
         // Post the archival results to the Coordinator.
         now := clock.Now(c).UTC()
-        err := ds.Get(c).RunInTransaction(func(c context.Context) error {
+        var ierr error

[Vadim Sh., 2016/04/07 01:21:32]

nit: reset ierr to nil at the beginning of a transaction (since it can be
retried)

[dnj, 2016/04/11 17:20:03]

Done.

+        err = ds.Get(c).RunInTransaction(func(c context.Context) error {
+                // Note that within this transaction, we have two return values:
+                // - Non-nil to abort the transaction.
+                // - Specific error via "ierr".
                 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.")
+
+                // If our log stream is not in LSArchiveTasked, we will reject this archive
+                // request with FailedPrecondition.
+                switch {
+                case ls.Archived():
+                        // Return nil if the log stream is already archived (idempotent).
+                        log.Warningf(c, "Log stream is already archived.")
                         return nil
+
+                case ls.State != coordinator.LSArchiveTasked:
+                        log.Fields{
+                                "state": ls.State,
+                        }.Errorf(c, "Log stream is not in archival tasked state.")
+                        ierr = grpcutil.Errf(codes.FailedPrecondition, "Log stream has not tasked an archival.")
+                        return ierr
+                }
+
+                // If this request contained an error, we will reject it with Aborted status
+                // if we are below our error threshold.
+                if req.Error {
+                        // Increment our error count.
+                        ls.ArchiveErrors++
+                        if err := di.Put(ls); err != nil {
+                                log.WithError(err).Errorf(c, "Failed to update log stream error count.")
+                                return err
+                        }
+
+                        retries := int(ccfg.ArchiveRetries)
+                        if ls.ArchiveErrors < retries {
+                                log.Fields{
+                                        "path":           req.Path,
+                                        "errorCount":     ls.ArchiveErrors,
+                                        "archiveRetries": retries,
+                                }.Warningf(c, "Rejecting failed archival: below error threshold.")
+
+                                // Fail this RPC call to keep the archival task in the queue.
+                                //
+                                // We return this via "ierr" because we want the transaction to succeed.
+                                ierr = grpcutil.Errf(codes.Aborted, "below error threshold (%d < %d)", ls.ArchiveErrors, retries)
+                                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,
+                                "archiveRetries": retries,
+                        }.Warningf(c, "Log stream has exceeded archive error threshold. Archiving empty stream.")
+
+                        req.TerminalIndex = -1
+                        req.LogEntryCount = 0
                 }
 
                 // Update archival information. Make sure this actually marks the stream as
                 // archived.
-                ls.Updated = now
                 ls.State = coordinator.LSArchived
-                ls.ArchiveWhole = req.Complete
+                ls.ArchivedTime = now
+
+                if ls.TerminalIndex < 0 {
+                        // Also set the terminated time.
+                        ls.TerminatedTime = now
+                }
                 ls.TerminalIndex = req.TerminalIndex
+
+                ls.ArchiveLogEntryCount = req.LogEntryCount
                 ls.ArchiveStreamURL = req.StreamUrl
                 ls.ArchiveStreamSize = req.StreamSize
                 ls.ArchiveIndexURL = req.IndexUrl
                 ls.ArchiveIndexSize = req.IndexSize
                 ls.ArchiveDataURL = req.DataUrl
                 ls.ArchiveDataSize = req.DataSize
 
                 // Update the log stream.
-                if err := ls.Put(di); err != nil {
+                if err := di.Put(ls); 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 ierr != nil {
+                log.WithError(err).Errorf(c, "Failed to mark stream as archived.")
+                return nil, ierr
+        }
         if err != nil {
-                log.WithError(err).Errorf(c, "Failed to mark stream as archived.")
+                log.WithError(err).Errorf(c, "Internal error.")
                 return nil, grpcutil.Internal
         }
 
         return &google.Empty{}, nil
 }