Snapshot streaming by anshulpundir · Pull Request #2458 · moby/swarmkit

anshulpundir · 2017-11-20T19:58:13Z

Based on proposal #2416

Signed-off-by: Anshul Pundir anshul.pundir@docker.com

stevvooe · 2017-11-20T22:17:20Z

+	// RaftMessageStream returns a gRPC stream that can be used to stream raft messages
+	// to be processed on a raft member.
+	// It is called from the RaftMember willing to send a message to its destination ('To' field)
+	rpc RaftMessageStream(stream ProcessRaftMessageRequest) returns (ProcessRaftMessageResponse) {


This needs a response type wrapper.

This seems confusing. The doc says that it returns a stream, but it only seems to take a stream as input. Who is the caller and callee in this scenario?

stevvooe · 2017-11-20T22:18:58Z

 func DefaultRaftConfig() api.RaftConfig {
 	return api.RaftConfig{
 		KeepOldSnapshots:           0,
-		SnapshotInterval:           10000,


Why the big change? Was this for testing?

stevvooe · 2017-11-20T22:21:22Z

+
+		// Append the received snapshot data.
+		if recvdMsg.Message.Type == raftpb.MsgSnap {
+			assembledMessage.Message.Snapshot.Data = append(assembledMessage.Message.Snapshot.Data, recvdMsg.Message.Snapshot.Data...)


Doesn't this arrive at the same problem? This buffers the whole thing into a single buffer. How does this address the issue?

The problem was being able to stream large snapshots without arbitrarily large gRPC message limits and timeout outs. This addresses that problem. Memory constraint is not the problem we're trying to solve, at least yet.

@anshulpundir Fair enough. :)

This could double (or more!) memory usage.

Here's how we can decide: Do we have to load the entire snapshot into memory to apply it? If yes, then this is fine, but if not, we should buffer this to disk (tmp is fine).

Buffering to disk won't be any better would it ? to apply we'd still need to load the snapshot into memory before passing it to etcd/raft.

Also, I think MemoryStorage::ApplySnapshot() doesn't actually do a memory copy, so we should still just need 1x memory.

There are three components:

"The bytes" of the snapshot itself.

The structure that gets applied to the store.

The store itself (not raft, but the database).

With the current model, I think the minimum is a 2x bump. Ideally, we can stream it in and avoid 1 and 2. Buffering to disk avoids 1, if 2 is necessary.

Looking at the code, there are two components: the entries and the snapshot data, a []byte. From the looks of it, the snapshot is applied always as a []byte, so we can never avoid 2. If a snapshot is applied in this fashion, there may exist a snapshot that has a similar size, which can contribute to bloat.

So, I think I agree with you: this code is probably fine as is, as long as this doesn't get copied (and it doesn't look like it does). Following that, I don't think there is a easy way to avoid the 2x memory bump by applying a snapshot in a stream.

i might be mistaken, but I don't think this code is any worse off than the regular raft message rpc call, except for the allocations and copies required to expand the array from the successive appends.

that said, maybe it's advantageous to take the couple of extra bytes to send the total message length with each stream request, so we can on the first message allocate an appropriately sized byte slice.

I did consider that, but streaming snapshots should not happen that frequently, so this is probably OK for now. We can always make the change to send the size later, if needed.

stevvooe · 2017-11-20T22:23:19Z

+// Function to check if streaming is needed.
+// Returns true if the message type is MsgSnap/MsgApp
+// and size larger than MaxRaftMsgSize.
+func needsStreaming(m *raftpb.Message) bool {


I would advocate for always using streaming unless the other end doesn't support it.

Can you please elaborate on this ?

I thought a bit more about this. If the message is smaller than GrpcMaxMsgSize, it's less efficient to first try to stream and then fall back to a basic send if unsupported.

Having said that, I do see the value in always streaming, and eventually deprecating the other endpoint, unless there is a performance overhead to streaming a single message.

Can you please also elaborate on this ?

@stevvooe

I also like the idea of always using streaming. It gives the streaming version of the RPC more real-world coverage instead of only using it in unusual corner cases. It would let us eventually deprecate the non-streaming version.

I don't think the mixed-version scenario is important to optimize.

I agree with the point about giving this more real-world coverage. I'll make this change.

stevvooe

Reviews inline.

This still seems to buffer the entire snapshot into the Rss of the swarm process. It would be best to buffer into a local file or apply the snapshot to a temp location as it comes in.

codecov · 2017-11-20T23:24:56Z

Codecov Report

Merging #2458 into master will decrease coverage by 2.09%.
The diff coverage is 93.67%.

@@            Coverage Diff            @@
##           master    #2458     +/-   ##
=========================================
- Coverage   63.74%   61.64%   -2.1%     
=========================================
  Files          64      128     +64     
  Lines       11793    21076   +9283     
=========================================
+ Hits         7517    12993   +5476     
- Misses       3664     6680   +3016     
- Partials      612     1403    +791

aaronlehmann · 2017-11-24T15:05:10Z

 	"errors"
 	"fmt"
+	"github.com/coreos/etcd/raft/raftpb"
+	"github.com/docker/swarmkit/manager/state/raft/transport"


Can you put these below with the other non-stdlib imports?

aaronlehmann · 2017-11-24T15:06:46Z

 	}
 }
+
+func GetSnapshotMessage(from, to uint64, size int) *raftpb.Message {


Put this in manager/state/raft/testutils instead of duplicating it in two places.

aaronlehmann · 2017-11-24T15:08:44Z

+	// Get the max payload size.
+	payloadSize := raftMessagePayloadSize(m)
+
+	// split the snpashot into smaller messages.


aaronlehmann · 2017-11-24T15:10:00Z

+	// split the snpashot into smaller messages.
+	for snapDataIndex := 0; snapDataIndex < size; {
+		remainingSize := size - snapDataIndex
+		chunkSize := int(math.Min(float64(remainingSize), float64(payloadSize)))


Using floating point math here doesn't feel right.

holy shit I know right!

you'd prefer just doing if/else ? @aaronlehmann

instead of math.Min why not just do min by hand for integers, so there's no round trip through floats?

chunkSize = payloadSize if remainingSize < payloadSize { chunkSize = remainingSize }

aaronlehmann · 2017-11-24T15:15:15Z

+		for chunkIndex := 0; chunkIndex < chunkSize; chunkIndex++ {
+			raftMsg.Snapshot.Data[chunkIndex] = m.Snapshot.Data[snapDataIndex]
+			snapDataIndex++
+		}


Use the builtin function copy instead of this byte-by-byte loop.

i don't believe we're mutating the bytes, so there's probably a way to avoid copying and leverage the same underlying storage array for all of the slices.

this would need to be documented, so that someone didn't come later on and break that assumption.

i think i misunderstood this code actually...

Avoiding copy here would be ideal, but I'm not sure if there is a way to do this. I'll make the change if it is.

I think it should be possible to avoid copy @dperny

yeah, we don't need to do any copying at all. instead of this loop, just do

raftMsg.Snapshot.Data = m.Snapshot.Data[snapDataIndex:(snapDataIndex+chunkSize)]

And then you can change the for loop above to be something like

var chunkSize int for snapDataIndex := 0; snapDataIndex < size; snapDataIndex = snapDataIndex + chunkSize { // do the processing in here }

aaronlehmann · 2017-11-24T15:18:57Z

+// Function to check if streaming is needed.
+// Returns true if the message type is MsgSnap/MsgApp
+// and size larger than MaxRaftMsgSize.
+func needsStreaming(m *raftpb.Message) bool {


I also like the idea of always using streaming. It gives the streaming version of the RPC more real-world coverage instead of only using it in unusual corner cases. It would let us eventually deprecate the non-streaming version.

I don't think the mixed-version scenario is important to optimize.

anshulpundir · 2017-11-27T21:16:45Z

Also ping @dperny @nishanttotla for review

dperny

Lot of comments. Not all of these are blockers. I'll defer to your judgement on them.

dperny · 2017-11-27T21:36:12Z

+		if err == io.EOF {
+			break
+		} else if err != nil {
+			log.G(context.Background()).WithError(err).Error("error while reading from stream")


nit: stream.Context(), instead of context.Background(), unless there's some reason I'm not aware of to do so.

dperny · 2017-11-27T21:37:39Z

+		if recvdMsg.Message.Type == raftpb.MsgSnap {
+			assembledMessage.Message.Snapshot.Data = append(assembledMessage.Message.Snapshot.Data, recvdMsg.Message.Snapshot.Data...)
+		} else {
+			log.G(context.Background()).Errorf("Ignoring unexpected message type received on stream: %d", recvdMsg.Message.Type)


same as above, use stream context.

dperny · 2017-11-27T21:40:56Z

+	// split the snpashot into smaller messages.
+	for snapDataIndex := 0; snapDataIndex < size; {
+		remainingSize := size - snapDataIndex
+		chunkSize := int(math.Min(float64(remainingSize), float64(payloadSize)))


instead of math.Min why not just do min by hand for integers, so there's no round trip through floats?

chunkSize = payloadSize if remainingSize < payloadSize { chunkSize = remainingSize }

dperny · 2017-11-27T21:41:36Z


+const (
+	// GrpcMaxMsgSize is the max allowed gRPC message size.
+	GrpcMaxMsgSize = 4 << 20


why represent this as 4 << 20? just curious.

Someone suggested it as being cleaner/easier to read than 4 * 1024 * 1024.

lol i disagree strongly with someone but that's bikeshedding.

so this part looks fine to me.

Its a little shorter, but I think its fine either way.

dperny · 2017-11-27T21:43:41Z

+		for chunkIndex := 0; chunkIndex < chunkSize; chunkIndex++ {
+			raftMsg.Snapshot.Data[chunkIndex] = m.Snapshot.Data[snapDataIndex]
+			snapDataIndex++
+		}


i don't believe we're mutating the bytes, so there's probably a way to avoid copying and leverage the same underlying storage array for all of the slices.

this would need to be documented, so that someone didn't come later on and break that assumption.

dperny · 2017-11-27T22:14:32Z

+// Returns the max allowable payload based on MaxRaftMsgSize and
+// the struct size for the given raftpb.Message and
+func raftMessagePayloadSize(m *raftpb.Message) int {
+	return GrpcMaxMsgSize - raftMessageStructSize(m)


dumb question: is GrpcMaxMsgSize actually the optimal message size? Or would a smaller message perform better?

GrpcMaxMsgSize is probably fine for now. I can't think of any reason to treat streamed messages differently than the regular messages.

dperny · 2017-11-27T22:17:49Z

+	assert.Error(t, err, "Received unexpected error EOF")
+
+	_, err = stream.CloseAndRecv()
+	errStr := fmt.Sprintf("rpc error: code = Internal desc = grpc: received message length %d exceeding the max size %d", msg.Size(), transport.GrpcMaxMsgSize)


instead of building an error string, you might want to use golang.google.com/grpc/status.FromError, and compare the code and message separately?

Sounds reasonable

dperny · 2017-11-27T22:19:03Z

+
+	// We should have the complete snapshot. Verify and process.
+	if err == io.EOF {
+		_, err = n.ProcessRaftMessage(context.Background(), assembledMessage)


Unsure if I think we should use the stream context here or not.

maybe it's better not to discard the first return value, and instead use it as the return value for stream.SendAndClose below?

Unsure if I think we should use the stream context here or not.

Yea, I'm unclear on this also. I'll think more about this.

dperny · 2017-11-27T22:38:19Z

+
+		// Append the received snapshot data.
+		if recvdMsg.Message.Type == raftpb.MsgSnap {
+			assembledMessage.Message.Snapshot.Data = append(assembledMessage.Message.Snapshot.Data, recvdMsg.Message.Snapshot.Data...)


i might be mistaken, but I don't think this code is any worse off than the regular raft message rpc call, except for the allocations and copies required to expand the array from the successive appends.

dperny · 2017-11-27T22:42:05Z

+
+		// Append the received snapshot data.
+		if recvdMsg.Message.Type == raftpb.MsgSnap {
+			assembledMessage.Message.Snapshot.Data = append(assembledMessage.Message.Snapshot.Data, recvdMsg.Message.Snapshot.Data...)


that said, maybe it's advantageous to take the couple of extra bytes to send the total message length with each stream request, so we can on the first message allocate an appropriately sized byte slice.

anshulpundir · 2017-11-28T23:11:35Z

 )

+// Build a snapshot message where each byte in the data is of the value (index % sizeof(byte))
+func getSnapshotMessage(from uint64, to uint64) raftpb.Message {


Trying to use this from testutils causes a circular dependency and moving this test to a transport_test package was just too problematic since unexported fields have been used extensively in the tests. Leaving as is for now, will find a way to address this in the future.

why do you need to use this from testutils?

Keeping it in once place instead of duplicating it. It's also used in raft_test.go

anshulpundir · 2017-11-28T23:56:14Z

Working on making the change to always stream. Just to clarify: this change will only stream snapshot messages larger than the max grpc message size. Support for splitting other message types, e.g. append entires, may be supported later.

anshulpundir · 2017-11-29T19:32:17Z

Addressed all review comments. ping @dperny @stevvooe @aaronlehmann for another pass

dperny

Couple of new comments but no blockers. LGTM.

dperny · 2017-11-29T22:04:44Z

 	// if a snapshot is being sent, set timeout to LargeSendTimeout because
 	// sending snapshots can take more time than other messages sent between peers.
 	// The same applies to AppendEntries as well, where messages can get large.
+	// TODO(anshul) remove when streaming change ready to merge.


what's this?

The large timeout/grpc message size is no longer needed once we have snapshot streaming. I guess that can be a separate PR.

dperny · 2017-11-29T22:07:29Z

+			chunkSize = payloadSize
+		}
+
+		raftMsg := *m


to be clear, you're derefing m to make a copy of it, yeah?

Yes, since I want to keep the entire message the same except for the fields to be split.

dperny · 2017-11-29T22:19:28Z

+	msgs := splitSnapshotData(ctx, &raftMsg)
+	assert.Equal(t, numMsgs, len(msgs), "unexpected number of messages")
+
+	raftMsg.Snapshot.Data = make([]byte, raftMessagePayloadSize)


nit, not a blocker. change if you want to, or don't. either way.

check := func(size, expectedMessages int) { raftMsg.Snapshot.Data = make([]byte, size) msgs := splitSnapshotData(ctx, &raftMsg) assert.Equal(t, expectedMesages, len(msgs), "unexpected number of raft messages") } numMsgs := int(math.Ceil(float64(snapshotSize) / float64(raftMessagePayloadSize))) check(raftMessagePayloadSize, numMsgs check(raftMessagePayloadSize-1, 1) check(raftMessagePayloadSize*2, 2) check(0, 0)

stevvooe · 2017-11-30T19:24:41Z

+	// RaftMessageStream accepts a stream of raft messages to be processed on a raft member, 
+	// returning a ProcessRaftMessageResponse when processing of the streamed messages is complete.
+	// It is called from the Raft leader, which uses it to stream messages to a raft member.
+	rpc RaftMessageStream(stream ProcessRaftMessageRequest) returns (ProcessRaftMessageResponse) {


These need to have rpc specific types. You can't just reuse the existing request type.

Offline conversation with @stevvooe This is a generally a d good practice, to help separate out the two rpcs and enable them to evolve independently.

stevvooe · 2017-11-30T19:27:11Z

+// GetSnapshotMessage creates and returns a raftpb.Message of type MsgSnap
+// where the snapshot data is of the given size and the value of each byte
+// is (index of the byte) % 256.
+func GetSnapshotMessage(from, to uint64, size int) *raftpb.Message {


This is not a getter. This should be NewSnapshotMessage.

stevvooe · 2017-11-30T19:43:51Z

 )

+const (
+	// GrpcMaxMsgSize is the max allowed gRPC message size.


GRPCMaxMsgSize.

stevvooe · 2017-11-30T19:55:21Z

+			log.G(stream.Context()).WithError(err).Error("error while reading from stream")
+			return err
+		}
+


Validate that each message belongs to the same index and are to be concatenated.

stevvooe · 2017-11-30T19:58:00Z


+	// RaftMessageStream accepts a stream of raft messages to be processed on a raft member, 
+	// returning a ProcessRaftMessageResponse when processing of the streamed messages is complete.
+	// It is called from the Raft leader, which uses it to stream messages to a raft member.


Explain here that a single stream call corresponds to a single, disassembled message.

Signed-off-by: Anshul Pundir <anshul.pundir@docker.com>

stevvooe · 2017-12-01T01:11:28Z

LGTM

GordonTheTurtle added the dco/no label Nov 20, 2017

anshulpundir force-pushed the snap2 branch from 769eb95 to 5eb7bc8 Compare November 20, 2017 20:07

GordonTheTurtle removed the dco/no label Nov 20, 2017

nishanttotla added area/raft kind/enhancement labels Nov 20, 2017

anshulpundir changed the title ~~Snapshot streaming.~~ Snapshot streaming Nov 20, 2017

stevvooe reviewed Nov 20, 2017

View reviewed changes

stevvooe changed the title ~~Snapshot streaming~~ [WIP] Snapshot streaming Nov 20, 2017

stevvooe suggested changes Nov 20, 2017

View reviewed changes

anshulpundir force-pushed the snap2 branch from 5eb7bc8 to 34088af Compare November 20, 2017 23:24

anshulpundir force-pushed the snap2 branch from 34088af to 4ffe82a Compare November 20, 2017 23:28

thaJeztah mentioned this pull request Nov 21, 2017

[17.09] Fix reapTime logic in NetworkDB + handle cleanup DNS for attachable container moby/libnetwork#2017

Merged

anshulpundir force-pushed the snap2 branch from 4ffe82a to efbe770 Compare November 21, 2017 22:53

aaronlehmann reviewed Nov 24, 2017

View reviewed changes

dperny requested changes Nov 27, 2017

View reviewed changes

anshulpundir force-pushed the snap2 branch from efbe770 to 8a73b5d Compare November 28, 2017 23:08

anshulpundir commented Nov 28, 2017

View reviewed changes

anshulpundir force-pushed the snap2 branch 2 times, most recently from 2d95f3a to b7c27fa Compare November 29, 2017 19:31

dperny approved these changes Nov 29, 2017

View reviewed changes

stevvooe reviewed Nov 30, 2017

View reviewed changes

anshulpundir force-pushed the snap2 branch 4 times, most recently from 614d598 to a6ceb09 Compare December 1, 2017 00:47

anshulpundir added 3 commits November 30, 2017 16:53

Protoc changes for raft message streaming.

58e42ec

Signed-off-by: Anshul Pundir <anshul.pundir@docker.com>

Snapshot streaming.

16301ee

Signed-off-by: Anshul Pundir <anshul.pundir@docker.com>

gRPC streaming by default for all messages.

6a0cf8c

Signed-off-by: Anshul Pundir <anshul.pundir@docker.com>

anshulpundir force-pushed the snap2 branch from a6ceb09 to 6a0cf8c Compare December 1, 2017 00:54

stevvooe approved these changes Dec 1, 2017

View reviewed changes

anshulpundir changed the title ~~[WIP] Snapshot streaming~~ Snapshot streaming Dec 1, 2017

moby deleted a comment from GordonTheTurtle Dec 1, 2017

stevvooe merged commit 889f1a3 into moby:master Dec 1, 2017

andrewhsu mentioned this pull request Dec 5, 2017

Vendor swarmkit to 4429c763 moby/moby#35698

Merged

anshulpundir mentioned this pull request Dec 6, 2017

Large snapshots prevent the addition of new managers to the cluster #2374

Closed

dperny mentioned this pull request Jan 12, 2018

Raft log streaming compaction #363

Closed

thaJeztah mentioned this pull request Feb 5, 2018

Regenerate protobuf for 3.5 (again) #2503

Merged

cyli mentioned this pull request May 29, 2018

Bumps GRPC to 1.12.0 #2649

Merged

xinfengliu mentioned this pull request Jan 31, 2023

Large snapshot causes adding a new manager to fail #3113

Closed

Conversation

anshulpundir commented Nov 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anshulpundir Nov 21, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anshulpundir Nov 27, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anshulpundir Nov 22, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevvooe left a comment

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented Nov 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anshulpundir commented Nov 27, 2017

Uh oh!

dperny left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anshulpundir commented Nov 20, 2017 •

edited

Loading

anshulpundir Nov 21, 2017 •

edited

Loading

anshulpundir Nov 27, 2017 •

edited

Loading

anshulpundir Nov 22, 2017 •

edited

Loading

codecov Bot commented Nov 20, 2017 •

edited

Loading

anshulpundir Nov 27, 2017 •

edited

Loading

anshulpundir Nov 28, 2017 •

edited

Loading

anshulpundir Nov 27, 2017 •

edited

Loading