Scale mutation queue with multiple shards #1048

gdbelvin · 2018-09-25T13:57:30Z

Tables with an incrementing value in the lower bits of the primary key are notoriously hard to scale because all the writes will go to a single server. This PR allows the queue to scale by adding write support for multiple queues.

Depends on: read by high/low watermarks #1045
Part of: retry-able mutations #1044

golangcibot · 2018-09-25T13:59:14Z

impl/sql/mutationstorage/queue.go

+	}
+	for rows.Next() {
+		var shardID int64
+		rows.Scan(&shardID)


G104: Errors unhandled.

golangcibot · 2018-09-25T13:59:14Z

impl/sql/mutationstorage/queue.go

+}
+
+// ReadQueue reads all mutations that are still in the queue up to batchSize.
+func (m *Mutations) ReadQueue(ctx context.Context, domainID string, shardID, low, high int64) ([]*mutator.QueueMessage, error) {


line is 128 characters

codecov · 2018-09-25T14:04:32Z

Codecov Report

Merging #1048 into master will increase coverage by 0.01%.
The diff coverage is 68.42%.

@@            Coverage Diff             @@
##           master    #1048      +/-   ##
==========================================
+ Coverage   65.24%   65.26%   +0.01%     
==========================================
  Files          39       39              
  Lines        2728     2764      +36     
==========================================
+ Hits         1780     1804      +24     
- Misses        630      633       +3     
- Partials      318      327       +9

Impacted Files	Coverage Δ
impl/sql/mutationstorage/mutations.go	`60.78% <ø> (ø)`	⬆️
impl/integration/env.go	`73.21% <50%> (ø)`	⬆️
impl/sql/mutationstorage/queue.go	`58.4% <61.01%> (+3.52%)`	⬆️
core/adminserver/admin_server.go	`67.02% <80%> (-0.38%)`	⬇️
core/sequencer/server.go	`63.73% <86.36%> (+0.18%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update a70c371...69e197c. Read the comment docs.

Send - Write to multiple shards HighWatermarks - return multiple high watermarks

golangcibot · 2018-10-05T13:00:55Z

impl/sql/mutationstorage/queue_test.go

+	if err != nil {
+		t.Fatalf("Failed to create Mutations: %v", err)
+	}
+	domainID := "foo"


string foo has 3 occurrences, make it a constant

golangcibot · 2018-10-05T13:00:55Z

impl/sql/mutationstorage/queue.go

+	}
+	for rows.Next() {
+		var shardID int64
+		rows.Scan(&shardID)


G104: Errors unhandled.

golangcibot · 2018-10-05T13:13:53Z

impl/sql/mutationstorage/queue.go

+}
+
+// randShard returns a random, enabled shard for domainID.
+func (m *Mutations) randShard(ctx context.Context, domainID string) (int64, error) {


47-73 lines are duplicate of impl/sql/mutationstorage/queue.go:92-118

golangcibot · 2018-10-05T13:13:53Z

impl/sql/mutationstorage/queue.go

+}
+
+// randomShard returns a random shard from the list of active shards for domainID.
+func (m *Mutations) randomShard(ctx context.Context, domainID string) (int64, error) {


92-118 lines are duplicate of impl/sql/mutationstorage/queue.go:47-73

gdbelvin · 2018-10-09T16:12:35Z

Updated and nice and green. PTAL.
Happy to walk through the code with you

pav-kv · 2018-10-10T17:57:14Z

core/adminserver/admin_server.go

@@ -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}


Who is supposed to create shards if they are more than 1? This looks like a place for some TODO.

Created issue #1048

You probably meant #1063
(putting it to create a link between github issues)

pav-kv · 2018-10-10T18:01:12Z

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 {


Can the receiver simply be taken by value?

Yep, that would simplify things a bit.

pav-kv · 2018-10-10T18:03:15Z

core/sequencer/sequencer_api.proto

@@ -36,8 +36,9 @@ message MapMetadata {
      // of logged items MUST be monotonically increasing.
      int64 highest_watermark = 2;
   }
-   // source defines the range of inputs used for this map revision.
-   SourceSlice source = 1;
+   reserved 1;


Are you at a point where you can't afford breaking compatibility by simply removing this field?

Fortunately we're early enough that we're happy to break compatibility.
This PR is also introducing backwards-incompatible database schema changes as well.

Yeah, that was my point. Why using reserved 1 instead of just dropping / reassigning the tag to the new field?

Good habits I suppose :-)

core/sequencer/sequencer_api.proto

impl/sql/mutationstorage/queue_test.go

gdbelvin · 2018-10-11T15:20:24Z

PTAL

pav-kv

LGTM % nits.

core/sequencer/server.go

pav-kv · 2018-10-11T15:44:36Z

core/sequencer/server.go

-	ReadQueue(ctx context.Context, domainID string, low, high int64) ([]*mutator.QueueMessage, error)
+	// HighWatermark returns the highest timestamp in the mutations table.
+	HighWatermarks(ctx context.Context, domainID string) (map[int64]int64, error)
+	// Read returns up to batchSize messages for domainID.


I think yes. How about renaming the ReadQueue method to just Read, and nicely naming the interface (e.g. Logs as you say)? I think that a more verbose interface name like ShardedMutationLog or similar could be fine as well.

core/sequencer/server.go

impl/sql/mutationstorage/mutations.go

impl/sql/mutationstorage/queue.go

impl/sql/mutationstorage/queue_test.go

impl/sql/mutationstorage/queue.go

gdbelvin · 2018-10-11T17:03:13Z

Thanks very much for the review!

gdbelvin added the blocked PR cannot be merged until another PR is merged. label Sep 25, 2018

golangcibot reviewed Sep 25, 2018

View reviewed changes

gdbelvin force-pushed the f/retry/multi-queue branch from b0d86e4 to 3c088f0 Compare September 26, 2018 11:26

gdbelvin added 6 commits October 5, 2018 10:14

Add Shards to Queue

7173c8a

Read Queue by shardID

5437f9b

Write to multiple shards

fc8c64f

Send - Write to multiple shards HighWatermarks - return multiple high watermarks

HighWatermarks metadata proto

055ac62

Read from multiple shards

39122ea

Create new shards during domain creation

0748802

gdbelvin force-pushed the f/retry/multi-queue branch from 3c088f0 to bf5e4e2 Compare October 5, 2018 12:59

golangcibot reviewed Oct 5, 2018

View reviewed changes

gdbelvin force-pushed the f/retry/multi-queue branch from bf5e4e2 to 75969ea Compare October 5, 2018 13:12

golangcibot reviewed Oct 5, 2018

View reviewed changes

gdbelvin added 2 commits October 5, 2018 14:16

SQL independent randomShard

2b398b2

Use constant for repeated string

6ebf36c

gdbelvin force-pushed the f/retry/multi-queue branch from 75969ea to 6ebf36c Compare October 5, 2018 13:16

gdbelvin added for review PR is ready for review and removed blocked PR cannot be merged until another PR is merged. labels Oct 5, 2018

gdbelvin requested a review from pav-kv October 5, 2018 13:19

Merge branch 'master' into f/retry/multi-queue

f6a17ed

pav-kv suggested changes Oct 10, 2018

View reviewed changes

Cleanup comments

b48a1c1

pav-kv reviewed Oct 11, 2018

View reviewed changes

pav-kv approved these changes Oct 11, 2018

View reviewed changes

Nits and improved error handling

69e197c

gdbelvin merged commit e869a3b into google:master Oct 11, 2018

gdbelvin deleted the f/retry/multi-queue branch October 11, 2018 17:12

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Scale mutation queue with multiple shards #1048

Scale mutation queue with multiple shards #1048

gdbelvin commented Sep 25, 2018

golangcibot Sep 25, 2018

golangcibot Sep 25, 2018

codecov bot commented Sep 25, 2018 •

edited

Loading

golangcibot Oct 5, 2018

golangcibot Oct 5, 2018

golangcibot Oct 5, 2018

golangcibot Oct 5, 2018

gdbelvin commented Oct 9, 2018

pav-kv Oct 10, 2018

gdbelvin Oct 11, 2018

pav-kv Oct 11, 2018

pav-kv Oct 10, 2018

gdbelvin Oct 11, 2018

pav-kv Oct 10, 2018

gdbelvin Oct 11, 2018

pav-kv Oct 11, 2018

gdbelvin Oct 11, 2018

gdbelvin commented Oct 11, 2018

pav-kv left a comment

pav-kv Oct 11, 2018

gdbelvin commented Oct 11, 2018

Scale mutation queue with multiple shards #1048

Scale mutation queue with multiple shards #1048

Conversation

gdbelvin commented Sep 25, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Sep 25, 2018 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gdbelvin commented Oct 9, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gdbelvin commented Oct 11, 2018

pav-kv left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gdbelvin commented Oct 11, 2018

codecov bot commented Sep 25, 2018 •

edited

Loading