LogDog: Implement prefix expiration.

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

Index: appengine/logdog/coordinator/endpoints/registration/registerPrefix.go
diff --git a/appengine/logdog/coordinator/endpoints/registration/registerPrefix.go b/appengine/logdog/coordinator/endpoints/registration/registerPrefix.go
index 775d91d77ab8eb5db725a4493f7386b46e2fd50f..34753e4563a73449edd36afebe27c1ecba6a7972 100644
--- a/appengine/logdog/coordinator/endpoints/registration/registerPrefix.go
+++ b/appengine/logdog/coordinator/endpoints/registration/registerPrefix.go
@@ -5,6 +5,8 @@
 package registration
 
 import (
+	"time"
+
 	ds "github.com/luci/gae/service/datastore"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
@@ -18,6 +20,7 @@ import (
 	"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) {
@@ -58,6 +61,23 @@ func (s *server) RegisterPrefix(c context.Context, req *logdog.RegisterPrefixReq
 		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 {
@@ -133,6 +153,7 @@ func (s *server) RegisterPrefix(c context.Context, req *logdog.RegisterPrefixReq
 		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.")
@@ -152,3 +173,12 @@ func (s *server) RegisterPrefix(c context.Context, req *logdog.RegisterPrefixReq
 		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
+}