LogDog: Implement prefix expiration.

appengine/logdog/coordinator/endpoints/registration/registerPrefix.go

 // Copyright 2016 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 registration
 
 import (
+        "time"
+
         ds "github.com/luci/gae/service/datastore"
         "golang.org/x/net/context"
         "google.golang.org/grpc/codes"
 
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         "github.com/luci/luci-go/appengine/logdog/coordinator/hierarchy"
         "github.com/luci/luci-go/common/api/logdog_coordinator/registration/v1"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/cryptorand"
         "github.com/luci/luci-go/common/gcloud/pubsub"
         "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"
 )
 
 func (s *server) RegisterPrefix(c context.Context, req *logdog.RegisterPrefixRequest) (*logdog.RegisterPrefixResponse, error) {
         log.Fields{
                 "project":    req.Project,
                 "prefix":     req.Prefix,
                 "source":     req.SourceInfo,
                 "expiration": req.Expiration.Duration(),
         }.Debugf(c, "Registering log prefix.")
 
@@ skipping 20 matching lines @@
         }
 
         // Load our service and project configurations.
         svcs := coordinator.GetServices(c)
         cfg, err := svcs.Config(c)
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to load service configuration.")
                 return nil, grpcutil.Internal
         }
 
+        pcfg, err := coordinator.CurrentProjectConfig(c)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to load project configuration.")
+                return nil, grpcutil.Internal
+        }
+
+        // Determine our prefix expiration. This must be > 0, else there will be no
+        // window when log streams can be registered and this prefix is useless.
+        //
+        // We will choose the shortest expiration window defined by our request and
+        // our project and service configurations.
+        expiration := resolveExpiration(req.Expiration, cfg.Coordinator.PrefixExpiration, pcfg.PrefixExpiration)
+        if expiration <= 0 {
+                log.Errorf(c, "Refusing to register prefix in expired state.")
+                return nil, grpcutil.Errf(codes.InvalidArgument, "no prefix expiration defined")
+        }
+
         // Determine our Pub/Sub topic.
         cfgTransport := cfg.Transport
         if cfgTransport == nil {
                 log.Errorf(c, "Missing transport configuration.")
                 return nil, grpcutil.Internal
         }
 
         cfgTransportPubSub := cfgTransport.GetPubsub()
         if cfgTransportPubSub == nil {
                 log.Errorf(c, "Missing transport Pub/Sub configuration.")
@@ skipping 55 matching lines @@
                 case exists:
                         log.Errorf(c, "The prefix is already registered (transactional).")
                         return grpcutil.AlreadyExists
                 }
 
                 // The Prefix is not registered, so let's register it.
                 pfx.Created = now
                 pfx.Prefix = string(prefix)
                 pfx.Source = req.SourceInfo
                 pfx.Secret = []byte(secret)
+                pfx.Expiration = now.Add(expiration)
 
                 if err := di.Put(pfx); err != nil {
                         log.WithError(err).Errorf(c, "Failed to register prefix.")
                         return grpcutil.Internal
                 }
 
                 log.Infof(c, "The prefix was successfully registered.")
                 return nil
         }, nil)
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to register prefix (transactional).")
                 return nil, err
         }
 
         return &logdog.RegisterPrefixResponse{
                 Secret:         []byte(secret),
                 LogBundleTopic: string(pubsubTopic),
         }, nil
 }
+
+func resolveExpiration(candidates ...*google.Duration) (exp time.Duration) {
+        for _, c := range candidates {
+                if cd := c.Duration(); cd > 0 && (exp <= 0 || cd < exp) {
+                        exp = cd
+                }
+        }
+        return
+}