LogDog: Implement RegisterPrefix RPC.

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"
         "time"
 
         "github.com/golang/protobuf/proto"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         "github.com/luci/luci-go/appengine/logdog/coordinator/hierarchy"
         "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"
@@ skipping 77 matching lines @@
 
         pcfg, err := coordinator.CurrentProjectConfig(c)
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to load current project configuration.")
                 return nil, grpcutil.Internal
         }
 
         // Determine the archival expiration.
         archiveDelayMax := resolveArchiveDelay(cfg.Coordinator, pcfg)
 
-        // Register our Prefix.
-        //
-        // This will also verify that our request secret matches the registered one,
-        // if one is registered.
-        //
-        // Note: This step will not be necessary once a "register prefix" RPC call is
-        // implemented.
-        lsp := logStreamPrefix{
-                prefix: string(prefix),
-                secret: req.Secret,
+        // Load our Prefix. It must be registered.
+        di := ds.Get(c)
+
+        pfx := &coordinator.LogPrefix{ID: coordinator.LogPrefixID(prefix)}
+        if err := di.Get(pfx); err != nil {
+                log.Fields{
+                        log.ErrorKey: err,
+                        "id":         pfx.ID,
+                        "prefix":     prefix,
+                }.Errorf(c, "Failed to load log stream prefix.")
+                if err == ds.ErrNoSuchEntity {
+                        return nil, grpcutil.Errf(codes.FailedPrecondition, "prefix is not registered")
+                }
+                return nil, grpcutil.Internal
         }
-        pfx, err := registerPrefix(c, &lsp)
-        if err != nil {
-                log.Errorf(c, "Failed to register/validate log stream prefix.")
-                return nil, err
+
+        // The prefix secret must match the request secret. If it does, we know this
+        // is a legitimate registration attempt.
+        if subtle.ConstantTimeCompare(pfx.Secret, req.Secret) != 1 {
+                log.Errorf(c, "Request secret does not match prefix secret.")
+                return nil, grpcutil.Errf(codes.InvalidArgument, "invalid secret")
         }
-        log.Fields{
-                "prefix":        pfx.Prefix,
-                "prefixCreated": pfx.Created,
-        }.Debugf(c, "Loaded log stream prefix.")
 
-        di := ds.Get(c)
+        // Check for registration, and that the prefix did not expire
+        // (non-transactional).
         ls := &coordinator.LogStream{ID: coordinator.LogStreamID(path)}
         lst := ls.State(di)
 
-        // Check for registration (non-transactional).
-        if err := checkRegisterStream(di, ls, lst); err != nil {
+        if err := di.GetMulti([]interface{}{ls, lst}); err != nil {
                 if !anyNoSuchEntity(err) {
                         log.WithError(err).Errorf(c, "Failed to check for log stream.")
                         return nil, err
                 }
 
                 // The stream is not registered. Perform a transactional registration via
                 // mutation.
                 //
                 // Determine which hierarchy components we need to add.
                 comps := hierarchy.Components(path)
                 if comps, err = hierarchy.Missing(di, comps); err != nil {
                         log.WithError(err).Warningf(c, "Failed to probe for missing hierarchy components.")
                 }
 
                 // Before we go into transaction, try and put these entries. This should not
                 // be contested, since components don't share an entity root.
                 if err := hierarchy.PutMulti(di, comps); err != nil {
-                        log.WithError(err).Infof(c, "Failed to add missing hierarchy components.")
+                        log.WithError(err).Errorf(c, "Failed to add missing hierarchy components.")
                         return nil, grpcutil.Internal
                 }
 
                 // The stream does not exist. Proceed with transactional registration.
                 err = tumble.RunMutation(c, &registerStreamMutation{
                         RegisterStreamRequest: req,
                         desc:         &desc,
                         pfx:          pfx,
                         lst:          lst,
                         ls:           ls,
                         archiveDelay: archiveDelayMax,
                 })
                 if err != nil {
                         log.Fields{
                                 log.ErrorKey: err,
                         }.Errorf(c, "Failed to register LogStream.")
                         return nil, err
                 }
         }
 
         return &logdog.RegisterStreamResponse{
                 Id:    string(ls.ID),
                 State: buildLogStreamState(ls, lst),
         }, nil
 }
 
-func checkRegisterStream(di ds.Interface, ls *coordinator.LogStream, lst *coordinator.LogStreamState) error {
-        // Load the existing log stream state.
-        //
-        // We have already verified that the secrets match when the Prefix was
-        // registered. This will have to verify secrets when prefix registration
-        // moves its own RPC.
-        return di.GetMulti([]interface{}{ls, lst})
-}
-
 type registerStreamMutation struct {
         *logdog.RegisterStreamRequest
 
         desc         *logpb.LogStreamDescriptor
         pfx          *coordinator.LogPrefix
         ls           *coordinator.LogStream
         lst          *coordinator.LogStreamState
         archiveDelay time.Duration
 }
 
 func (m *registerStreamMutation) RollForward(c context.Context) ([]tumble.Mutation, error) {
         di := ds.Get(c)
 
-        // Check if our stream is registered (transactional).
-        if err := checkRegisterStream(di, m.ls, m.lst); err != nil {
+        // Load our state and stream (transactional).
+        if err := di.GetMulti([]interface{}{m.ls, m.lst}); err != nil {
                 if !anyNoSuchEntity(err) {
                         log.WithError(err).Errorf(c, "Failed to check for stream registration (transactional).")
                         return nil, err
                 }
 
                 // The stream is not yet registered.
                 log.Infof(c, "Registering new log stream.")
 
                 now := clock.Now(c).UTC()
 
@@ skipping 42 matching lines @@
                         return nil, grpcutil.Internal
                 }
         }
 
         return nil, nil
 }
 
 func (m *registerStreamMutation) Root(c context.Context) *ds.Key {
         return ds.Get(c).KeyForObj(m.ls)
 }