LogDog: Support per-namespace expired archival.

appengine/logdog/coordinator/endpoints/services/terminateStream.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 (
         "crypto/subtle"
 
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/info"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
+        "github.com/luci/luci-go/appengine/logdog/coordinator/mutations"
+        "github.com/luci/luci-go/appengine/tumble"
         "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"
 )
 
@@ skipping 32 matching lines @@
 
         // Initialize our archival parameters.
         params := coordinator.ArchivalParams{
                 RequestID:      info.Get(c).RequestID(),
                 SettleDelay:    cfg.Coordinator.ArchiveSettleDelay.Duration(),
                 CompletePeriod: cfg.Coordinator.ArchiveDelayMax.Duration(),
         }
 
         // Transactionally validate and update the terminal index.
         err = ds.Get(c).RunInTransaction(func(c context.Context) error {
-                if err := ds.Get(c).Get(ls); err != nil {
+                di := ds.Get(c)
+
+                if err := di.Get(ls); err != nil {
                         if err == ds.ErrNoSuchEntity {
                                 log.Debugf(c, "LogEntry not found.")
                                 return grpcutil.Errf(codes.NotFound, "Log stream %q is not registered", req.Path)
                         }
 
                         log.WithError(err).Errorf(c, "Failed to load LogEntry.")
                         return grpcutil.Internal
                 }
 
                 switch {
@@ skipping 26 matching lines @@
                         if err := params.PublishTask(c, ap, ls); err != nil {
                                 if err == coordinator.ErrArchiveTasked {
                                         log.Warningf(c, "Archival has already been tasked for this stream.")
                                         return nil
                                 }
 
                                 log.WithError(err).Errorf(c, "Failed to create archive task.")
                                 return grpcutil.Internal
                         }
 
-                        if err := ds.Get(c).Put(ls); err != nil {
+                        if err := di.Put(ls); err != nil {
                                 log.Fields{
                                         log.ErrorKey: err,
                                 }.Errorf(c, "Failed to Put() LogStream.")
                                 return grpcutil.Internal
                         }
 
+                        // Delete the archive expiration mutation, since we have just dispatched
+                        // an archive request.
+                        aeParent, aeName := (&mutations.CreateArchiveTask{Path: path}).TaskName(di)
+                        if err := tumble.CancelNamedMutations(c, aeParent, aeName); err != nil {
+                                log.WithError(err).Errorf(c, "Failed to cancel archive expiration mutation.")
+                                return grpcutil.Internal
+                        }
+
                         log.Fields{
                                 "terminalIndex": ls.TerminalIndex,
                         }.Infof(c, "Terminal index was set and archival was dispatched.")
                         return nil
                 }
         }, nil)
         if err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                 }.Errorf(c, "Failed to update LogStream.")
                 return nil, err
         }
 
         return &google.Empty{}, nil
 }