LogDog: Support per-namespace expired archival.

appengine/logdog/coordinator/endpoints/services/registerStream.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"
         "errors"
+        "time"
 
         ds "github.com/luci/gae/service/datastore"
         "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"
@@ skipping 70 matching lines @@
         }
         switch {
         case req.Desc.Prefix != string(prefix):
                 return nil, grpcutil.Errf(codes.InvalidArgument, "Descriptor prefix does not match path (%s != %s)",
                         req.Desc.Prefix, prefix)
         case req.Desc.Name != string(name):
                 return nil, grpcutil.Errf(codes.InvalidArgument, "Descriptor name does not match path (%s != %s)",
                         req.Desc.Name, name)
         }
 
+        // Load our config and archive expiration.
+        _, cfg, err := coordinator.GetServices(c).Config(c)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to load configuration.")
+                return nil, grpcutil.Internal
+        }
+
+        archiveDelayMax := cfg.Coordinator.ArchiveDelayMax.Duration()
+        if archiveDelayMax < 0 {
+                log.Fields{
+                        "archiveDelayMax": archiveDelayMax,
+                }.Errorf(c, "Must have positive maximum archive delay.")
+                return nil, grpcutil.Internal
+        }
+
         // Already registered? (Non-transactional).
         ls := coordinator.LogStreamFromPath(path)
         switch err := ds.Get(c).Get(ls); err {
         case nil:
                 // We want this to be idempotent, so validate that it matches the current
                 // configuration and return accordingly.
                 if err := matchesLogStream(req, ls); err != nil {
                         return nil, grpcutil.Errf(codes.AlreadyExists, "Log stream is already incompatibly registered: %v", err)
                 }
 
         case ds.ErrNoSuchEntity:
                 // The registration is valid, so retain it.
-                if err := tumble.RunMutation(c, &registerStreamMutation{ls, req}); err != nil {
+                err = tumble.RunMutation(c, &registerStreamMutation{
+                        LogStream:    ls,
+                        req:          req,
+                        archiveDelay: archiveDelayMax,
+                })
+                if err != nil {
                         log.Fields{
                                 log.ErrorKey: err,
                         }.Errorf(c, "Failed to register LogStream.")
                         return nil, filterError(err)
                 }
 
         default:
                 log.WithError(err).Errorf(c, "Failed to check for log stream.")
                 return nil, grpcutil.Internal
         }
@@ skipping 11 matching lines @@
         case grpc.Code(err) == codes.Unknown:
                 return grpcutil.Internal
         default:
                 return err
         }
 }
 
 type registerStreamMutation struct {
         *coordinator.LogStream
 
-        req *logdog.RegisterStreamRequest
+        req          *logdog.RegisterStreamRequest
+        archiveDelay time.Duration
 }
 
 func (m registerStreamMutation) RollForward(c context.Context) ([]tumble.Mutation, error) {
         di := ds.Get(c)
 
         // Already registered? (Transactional).
         switch err := di.Get(m.LogStream); err {
         case ds.ErrNoSuchEntity:
                 break
 
@@ skipping 27 matching lines @@
         m.Created = ds.RoundTime(clock.Now(c).UTC())
         m.TerminalIndex = -1
 
         if err := di.Put(m.LogStream); err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                 }.Errorf(c, "Failed to Put() LogStream.")
                 return nil, grpcutil.Internal
         }
 
+        // Add a named delayed mutation to archive this stream if it's not archived
+        // yet. We will cancel this in terminateStream once we dispatch an immediate
+        // archival task.
+        archiveExpiredMutation := mutations.CreateArchiveTask{
+                Path:       m.Path(),
+                Expiration: clock.Now(c).Add(m.archiveDelay),
+        }
+        aeParent, aeName := archiveExpiredMutation.TaskName(di)
+        err := tumble.PutNamedMutations(c, aeParent, map[string]tumble.Mutation{
+                aeName: &archiveExpiredMutation,
+        })
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to load named mutations.")
+                return nil, grpcutil.Internal
+        }
+
         return []tumble.Mutation{
                 &mutations.PutHierarchyMutation{
                         Path: m.Path(),
                 },
         }, nil
 }
 
 func (m registerStreamMutation) Root(c context.Context) *ds.Key {
         return ds.Get(c).KeyForObj(m.LogStream)
 }