Create new shards during domain creation

google · Sep 25, 2018 · b0d86e4 · b0d86e4
1 parent 7584613
commit b0d86e4
Show file tree

Hide file tree

Showing 4 changed files with 44 additions and 21 deletions.
diff --git a/cmd/keytransparency-sequencer/main.go b/cmd/keytransparency-sequencer/main.go
@@ -113,7 +113,7 @@ func main() {
 	keygen := func(ctx context.Context, spec *keyspb.Specification) (proto.Message, error) {
 		return der.NewProtoFromSpec(spec)
 	}
-	adminServer := adminserver.New(tlog, tmap, logAdmin, mapAdmin, domainStorage, keygen)
+	adminServer := adminserver.New(tlog, tmap, logAdmin, mapAdmin, domainStorage, mutations, keygen)
 	glog.Infof("Signer starting")
 
 	// Run servers

diff --git a/core/adminserver/admin_server.go b/core/adminserver/admin_server.go
@@ -75,14 +75,21 @@ var (
 	}
 )
 
+// QueueAdmin controls the lifecycle and scaling of mutation queues.
+type QueueAdmin interface {
+	// AddShards creates and adds new shards for queue writing to a domain.
+	AddShards(ctx context.Context, domainID string, shardIDs ...int64) error
+}
+
 // Server implements pb.KeyTransparencyAdminServer
 type Server struct {
-	tlog     tpb.TrillianLogClient
-	tmap     tpb.TrillianMapClient
-	logAdmin tpb.TrillianAdminClient
-	mapAdmin tpb.TrillianAdminClient
-	domains  domain.Storage
-	keygen   keys.ProtoGenerator
+	tlog       tpb.TrillianLogClient
+	tmap       tpb.TrillianMapClient
+	logAdmin   tpb.TrillianAdminClient
+	mapAdmin   tpb.TrillianAdminClient
+	domains    domain.Storage
+	queueAdmin QueueAdmin
+	keygen     keys.ProtoGenerator
 }
 
 // New returns a KeyTransparencyAdmin implementation.
@@ -91,15 +98,17 @@ func New(
 	tmap tpb.TrillianMapClient,
 	logAdmin, mapAdmin tpb.TrillianAdminClient,
 	domains domain.Storage,
+	queueAdmin QueueAdmin,
 	keygen keys.ProtoGenerator,
 ) *Server {
 	return &Server{
-		tlog:     tlog,
-		tmap:     tmap,
-		logAdmin: logAdmin,
-		mapAdmin: mapAdmin,
-		domains:  domains,
-		keygen:   keygen,
+		tlog:       tlog,
+		tmap:       tmap,
+		logAdmin:   logAdmin,
+		mapAdmin:   mapAdmin,
+		domains:    domains,
+		queueAdmin: queueAdmin,
+		keygen:     keygen,
 	}
 }
 
@@ -249,6 +258,7 @@ func (s *Server) CreateDomain(ctx context.Context, in *pb.CreateDomainRequest) (
 			err, logTree.TreeId, delLogErr, mapTree.TreeId, delMapErr)
 	}
 
+	// Create domain - {log, map} binding.
 	if err := s.domains.Write(ctx, &domain.Domain{
 		DomainID:    in.GetDomainId(),
 		MapID:       mapTree.TreeId,
@@ -260,6 +270,13 @@ func (s *Server) CreateDomain(ctx context.Context, in *pb.CreateDomainRequest) (
 	}); err != nil {
 		return nil, fmt.Errorf("adminserver: domains.Write(): %v", err)
 	}
+
+	// Create shards for queue.
+	shardIDs := []int64{1}
+	if err := s.queueAdmin.AddShards(ctx, in.GetDomainId(), shardIDs...); err != nil {
+		return nil, fmt.Errorf("adminserver: AddShards(%v): %v", shardIDs, err)
+	}
+
 	d := &pb.Domain{
 		DomainId:    in.GetDomainId(),
 		Log:         logTree,

diff --git a/core/adminserver/admin_server_test.go b/core/adminserver/admin_server_test.go
@@ -85,6 +85,12 @@ func (e *miniEnv) Close() {
 	e.stopMockServer()
 }
 
+type fakeQueueAdmin struct{}
+
+func (*fakeQueueAdmin) AddShards(ctx context.Context, domainID string, shardIDs ...int64) error {
+	return nil
+}
+
 func TestCreateDomain(t *testing.T) {
 	for _, tc := range []struct {
 		desc     string
@@ -171,7 +177,7 @@ func TestCreateRead(t *testing.T) {
 		t.Fatalf("Failed to create trillian log server: %v", err)
 	}
 
-	svr := New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, storage, vrfKeyGen)
+	svr := New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, storage, &fakeQueueAdmin{}, vrfKeyGen)
 
 	for _, tc := range []struct {
 		domainID                 string
@@ -223,7 +229,7 @@ func TestDelete(t *testing.T) {
 		t.Fatalf("Failed to create trillian log server: %v", err)
 	}
 
-	svr := New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, storage, vrfKeyGen)
+	svr := New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, storage, &fakeQueueAdmin{}, vrfKeyGen)
 
 	for _, tc := range []struct {
 		domainID                 string
@@ -287,7 +293,7 @@ func TestListDomains(t *testing.T) {
 		t.Fatalf("Failed to create trillian log server: %v", err)
 	}
 
-	svr := New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, storage, vrfKeyGen)
+	svr := New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, storage, &fakeQueueAdmin{}, vrfKeyGen)
 
 	for _, tc := range []struct {
 		domainIDs []string

diff --git a/impl/integration/env.go b/impl/integration/env.go
@@ -141,7 +141,11 @@ func NewEnv(ctx context.Context) (*Env, error) {
 	if err != nil {
 		return nil, fmt.Errorf("env: failed to create domain storage: %v", err)
 	}
-	adminSvr := adminserver.New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, domainStorage, vrfKeyGen)
+	mutations, err := mutationstorage.New(db)
+	if err != nil {
+		return nil, fmt.Errorf("env: Failed to create mutations object: %v", err)
+	}
+	adminSvr := adminserver.New(logEnv.Log, mapEnv.Map, logEnv.Admin, mapEnv.Admin, domainStorage, mutations, vrfKeyGen)
 	cctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 	domainPB, err := adminSvr.CreateDomain(cctx, &pb.CreateDomainRequest{
@@ -158,10 +162,6 @@ func NewEnv(ctx context.Context) (*Env, error) {
 	glog.V(5).Infof("Domain: %# v", pretty.Formatter(domainPB))
 
 	// Common data structures.
-	mutations, err := mutationstorage.New(db)
-	if err != nil {
-		return nil, fmt.Errorf("env: Failed to create mutations object: %v", err)
-	}
 	authFunc := authentication.FakeAuthFunc
 	authz := &authorization.AuthzPolicy{}