[logdog] Replace Tumble with push queues.

logdog/appengine/coordinator/endpoints/services/terminateStream.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
 package services
 
 import (
         "crypto/subtle"
 
         "github.com/golang/protobuf/ptypes/empty"
+
         ds "github.com/luci/gae/service/datastore"
+
         "github.com/luci/luci-go/common/clock"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/grpc/grpcutil"
         "github.com/luci/luci-go/logdog/api/config/svcconfig"
         "github.com/luci/luci-go/logdog/api/endpoints/coordinator/services/v1"
         "github.com/luci/luci-go/logdog/appengine/coordinator"
         "github.com/luci/luci-go/logdog/appengine/coordinator/config"
         "github.com/luci/luci-go/logdog/appengine/coordinator/endpoints"
         "github.com/luci/luci-go/logdog/appengine/coordinator/mutations"
+        "github.com/luci/luci-go/logdog/appengine/coordinator/tasks"
         "github.com/luci/luci-go/tumble"
+
         "golang.org/x/net/context"
         "google.golang.org/grpc/codes"
 )
 
 // TerminateStream is an idempotent stream state terminate operation.
 func (s *server) TerminateStream(c context.Context, req *logdog.TerminateStreamRequest) (*empty.Empty, error) {
         log.Fields{
                 "project":       req.Project,
                 "id":            req.Id,
                 "terminalIndex": req.TerminalIndex,
@@ skipping 66 matching lines @@
                         lst.TerminalIndex = req.TerminalIndex
                         lst.TerminatedTime = now
 
                         if err := ds.Put(c, lst); err != nil {
                                 log.Fields{
                                         log.ErrorKey: err,
                                 }.Errorf(c, "Failed to Put() LogStream.")
                                 return grpcutil.Internal
                         }
 
-                        // Replace the pessimistic archive expiration mutation scheduled in
-                        // RegisterStream with an optimistic archival mutation.
-                        cat := mutations.CreateArchiveTask{
-                                ID: id,
+                        // Replace the pessimistic archive expiration task scheduled in
+                        // RegisterStream with an optimistic archival task.
+                        if err := tasks.CreateArchivalTask(c, id, logdog.ArchiveDispatchTask_TERMINATED,
+                                params.SettleDelay, params); err != nil {
 
-                                // Optimistic parameters.
-                                SettleDelay:    params.SettleDelay,
-                                CompletePeriod: params.CompletePeriod,
-
-                                // Schedule this mutation to execute after our settle delay.
-                                Expiration: now.Add(params.SettleDelay),
+                                log.WithError(err).Errorf(c, "Failed to create terminated archival task.")
+                                return grpcutil.Internal
                         }
 
-                        aeParent, aeName := ds.KeyForObj(c, lst), cat.TaskName(c)
-                        if err := tumble.PutNamedMutations(c, aeParent, map[string]tumble.Mutation{aeName: &cat}); err != nil {
-                                log.WithError(err).Errorf(c, "Failed to replace archive expiration mutation.")
-                                return grpcutil.Internal
+                        if err := tasks.DeleteArchiveStreamExpiredTask(c, id); err != nil {
+                                // If we can't delete this task, it will just run, notice that the
+                                // stream is archived, and quit. No big deal.
+                                log.WithError(err).Warningf(c, "(Non-fatal) Failed to delete expired archival task.")
+                        }
+
+                        // In case the stream was *registered* with Tumble, but is now being
+                        // processed with task queue code, clear the Tumble archival mutation.
+                        //
+                        // TODO(dnj): Remove this once Tumble is drained.
+                        archiveMutation := mutations.CreateArchiveTask{ID: id}
+                        if err := tumble.CancelNamedMutations(c, archiveMutation.Root(c), archiveMutation.TaskName(c)); err != nil {
+                                log.WithError(err).Warningf(c, "(Non-fatal) Failed to cancel archive mutation.")
                         }
 
                         log.Fields{
                                 "terminalIndex":  lst.TerminalIndex,
-                                "settleDelay":    cat.SettleDelay,
-                                "completePeriod": cat.CompletePeriod,
-                                "scheduledAt":    cat.Expiration,
-                        }.Debugf(c, "Terminal index was set, and archival mutation was scheduled.")
+                                "settleDelay":    params.SettleDelay,
+                                "completePeriod": params.CompletePeriod,
+                        }.Debugf(c, "Terminal index was set, and archival task was scheduled.")
                         return nil
                 }
         }, nil)
         if err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                 }.Errorf(c, "Failed to update LogStream.")
                 return nil, err
         }
 
         return &empty.Empty{}, nil
 }
 
 func standardArchivalParams(cfg *config.Config, pcfg *svcconfig.ProjectConfig) *coordinator.ArchivalParams {
         return &coordinator.ArchivalParams{
                 SettleDelay:    google.DurationFromProto(cfg.Coordinator.ArchiveSettleDelay),
                 CompletePeriod: endpoints.MinDuration(cfg.Coordinator.ArchiveDelayMax, pcfg.MaxStreamAge),
         }
 }