Initial distributor implementation

appengine/cmd/dm/deps/service.go

Index: appengine/cmd/dm/deps/service.go
diff --git a/appengine/cmd/dm/deps/service.go b/appengine/cmd/dm/deps/service.go
index d20981a6e39f739b6adf97cdfab87da9de6cc53d..99edd18b4f23a439eb807085a65828becc32f8aa 100644
--- a/appengine/cmd/dm/deps/service.go
+++ b/appengine/cmd/dm/deps/service.go
@@ -6,6 +6,7 @@ package deps
 
 import (
 	"github.com/golang/protobuf/proto"
+	"github.com/luci/luci-go/appengine/cmd/dm/distributor"
 	"github.com/luci/luci-go/appengine/tumble"
 	dm "github.com/luci/luci-go/common/api/dm/service/v1"
 	"github.com/luci/luci-go/common/grpcutil"
@@ -22,29 +23,35 @@ type deps struct{}
 
 var _ dm.DepsServer = (*deps)(nil)
 
-func depsServerPrelude(c context.Context, methodName string, req proto.Message) (context.Context, error) {
-	// Many of the DM request messages can be Normalize'd. This checks them for
-	// basic validity and normalizes cases where multiple representations can mean
-	// the same thing so that the service handlers only need to check for the
-	// canonical representation.
-	if norm, ok := req.(interface {
-		Normalize() error
-	}); ok {
-		if err := norm.Normalize(); err != nil {
-			return nil, grpcutil.MaybeLogErr(c, err, codes.InvalidArgument, "invalid request")
+func depsServerPrelude(reg distributor.Registry) func(context.Context, string, proto.Message) (context.Context, error) {
+	return func(c context.Context, methodName string, req proto.Message) (context.Context, error) {
+		// Many of the DM request messages can be Normalize'd. This checks them for
+		// basic validity and normalizes cases where multiple representations can mean
+		// the same thing so that the service handlers only need to check for the
+		// canonical representation.
+		if norm, ok := req.(interface {
+			Normalize() error
+		}); ok {
+			if err := norm.Normalize(); err != nil {
+				return nil, grpcutil.MaybeLogErr(c, err, codes.InvalidArgument, "invalid request")
+			}
 		}
+		c = distributor.WithRegistry(c, reg)
+		return c, nil
 	}
-	return c, nil
 }
 
-func newDecoratedDeps() dm.DepsServer {
-	return &dm.DecoratedDeps{Service: &deps{}, Prelude: depsServerPrelude}
+func newDecoratedDeps(reg distributor.Registry) dm.DepsServer {
+	return &dm.DecoratedDeps{
+		Service: &deps{},
+		Prelude: depsServerPrelude(reg),
+	}
 }
 
 // RegisterDepsServer registers an implementation of the dm.DepsServer with
 // the provided Registrar.
-func RegisterDepsServer(svr prpc.Registrar) {
-	dm.RegisterDepsServer(svr, newDecoratedDeps())
+func RegisterDepsServer(svr prpc.Registrar, reg distributor.Registry) {
+	dm.RegisterDepsServer(svr, newDecoratedDeps(reg))
 }
 
 // tumbleNow will run the mutation immediately, converting any non grpc errors
@@ -55,5 +62,6 @@ func tumbleNow(c context.Context, m tumble.Mutation) error {
 		logging.WithError(err).Errorf(c, "unknown error while applying mutation %v", m)
 		err = grpcutil.Internal
 	}
+	logging.Fields{"root": m.Root(c)}.Infof(c, "tumbleNow success")
 	return err
 }