Incremental Server Implementation

[alecholmez]

@snowp here's the implementation for the server part of delta. I'll be fixing up the mux and linear cache PRs as well as the integration test PR asynchronously to this being reviewed. Thanks in advanced!

[snowp]

Thanks for putting this up! Some comments to get you started

I think some comments/docs on the exact contract used by VersionMap would be useful given that it uses "" as a special value to indicate subscription. I think empty string is technically a valid version, so it might be good to use a more explicit sentinel value (e.g. nil with *string values?)

[alecholmez]

@snowp thanks for the review! I'll go through this today

[alecholmez]

@snowp just a heads up I pushed a commit that refactored quite a bit, I haven't addressed all the previous comments so I'll do that now. I just wanted to get your eyes on that code sooner rather than later. The tests are failing and there is an intermittent race condition but I'm looking into it.

[snowp]

Yeah we need to read this to be able to resume sessions, I'm just wondering if we should take care to only try to read this on the first request, as otherwise we'd be resetting resource versions to this value? Maybe error out if this is set for subsequent requests for a given type url, since this seems to be in violation with the spec

[alecholmez]

@snowp just pushed another commit, It seems the tests are failing from the mock CreateDeltaWatch being run exactly 6 times? I have no clue why and I'm currently investigating. Thanks so far for all the help! I'm working to fix the versioning issue with blank "" values so hopefully that should be coming soon.

I also reorganized a bit just so it's easier to track the tests and such so hopefully that helps.

[snowp]

Any progress on fixing the CI issue? Should I review this code as is?

[alecholmez]

Any progress on fixing the CI issue? Should I review this code as is?

@snowp No luck yet, I’ll look again today to see if I can find anything. A review as is might be helpful! The tests are failing to return the correct number of open watches so somethings getting messed up

[alecholmez]

Just found a bug in the server code, working to fix through that now

[alecholmez]

@snowp just pushed some changes, I'm seeing some super weird behaviors with ADS, would you mind taking a look at the test?

I was seeing this in my debugger:

The type url looks really polluted here somehow but when I just logged out in the test I saw nothing out of ordinary. From CI:

=== RUN   TestDeltaAggregatedHandlers
    delta_test.go:410: got 3 messages on the stream, not 5
--- FAIL: TestDeltaAggregatedHandlers (1.00s)

[alecholmez]

Nevermind! I found the problem haha

[snowp]

Any reason for this change?

[alecholmez]

Nope not sure why that happened, my editor might've messed up there. Should I revert?

[snowp]

Might as well, no need to pollute the git history

[snowp]

Indention here makes this a bit tricky to read, maybe put want into a separate variable?

[alecholmez]

Maybe something like?

	want := map[string]int{rsrc.EndpointType: 1, rsrc.ClusterType: 1, rsrc.RouteType: 1, rsrc.ListenerType: 1, rsrc.SecretType: 1}
	if !reflect.DeepEqual(want, config.deltaCounts) {
		t.Errorf("watch counts => got %v, want %v", config.deltaCounts, want)
	}

Or I can keep the long indentation but that does make it a bit hard to read

[alecholmez]

@snowp some of the comments you left didn't have the ability for me to respond too so I'll respond here:

I don't think it makes sense for resource additions/removals to affect the subscription state, but I could be wrong here. I think we need a way to communicate that a resource has been subscribed to within the stream but we haven't been able to provide the resource. Similarly if this resource goes away we want the server to be able to communicate this via removed_resource, while the stream state reverts to the state it was in when this resource was requested but no such resource existed.
I think it would be good to cover this case in a test, seems like a likely edge case.

That's an interesting thought. Do you think this could be somewhat satisfied by using a *string as a way to track the resource version? We define if nil then we want to track the resource but don't have it yet. I think I understand what you're saying but it's a bit unclear what the targeted functionality here would be if we don't want subscriptions to determine state since that is the fundamental basis of the delta. @htuch maybe can weigh in here?

I don't see how the presence of Node affects how we handle the nonce? Nonce isn't part of the node field so why would it be related? Shouldn't be always read the response nonce from the request? In the first request case we shouldn't even need to be bothered with the nonce since the check for existing watches at L198 will always return no watches

I'm assuming you're referring to this:

	var nonce string
	// The node information might only be set on the first incoming delta discovery request, so store it here so we can
	// reset it on subsequent requests that omit it.
	if req.Node != nil {
		node = req.Node
		nonce = req.GetResponseNonce()
	} else {
		req.Node = node

		// If we have no nonce, i.e. this is the first request on a delta stream, set one
		nonce = strconv.FormatInt(streamNonce, 10)
	}

I think it's a matter of how they initialized it? I could be wrong but I'm just as confused by this code as you are. I would've expected (regardless if a Node is there) we would just do:

        nonce := req.GetResponseNonce()
	// The node information might only be set on the first incoming delta discovery request, so store it here so we can
	// reset it on subsequent requests that omit it.
	if req.Node != nil {
		node = req.Node
	} else {
		req.Node = node
	}

Since on the first request that response nonce will be empty anyways.

[htuch]

@alecholmez can you describe the clarification you're after here? The thread has a lot of context that is hard to follow :)

[alecholmez]

@htuch I think we're confused on whether resource additions/removals should affect subscription state? Specifically with what @snowp brought up:

I don't think it makes sense for resource additions/removals to affect the subscription state, but I could be wrong here. I think we need a way to communicate that a resource has been subscribed to within the stream but we haven't been able to provide the resource. Similarly if this resource goes away we want the server to be able to communicate this via removed_resource, while the stream state reverts to the state it was in when this resource was requested but no such resource existed.
I think it would be good to cover this case in a test, seems like a likely edge case.

[htuch]

Additions/removals by the client will generally affect the server state, since the server needs to track what the client is subscribing to in delta.

[alecholmez]

@snowp I just opened this PR: https://github.com/alecholmez/go-control-plane/pull/7/files

I introduced the integration tests for delta and they brought to light some interesting failures. If you're available you might want to take a look. It seems the initial request functions perfectly but afterwards we start closing out streams improperly and failures occur. I'm going to start looking into this but it seems the unit tests haven't caught this.

if you'd like I can eventually merge that integration test code into this PR and merge it all in, or wait, whatever works! I branched off this branch since it has the server implementation.

[alecholmez]

@snowp I think we're actually able to do away with both IsWildcard and IsFirst. We fixed the data race but know ADS seems to fail anytime after the first update so I'll have to look into that. You can see that in the integration tests

aca74f9#diff-616f2ac71437a2f1daab08f6eb5ee33c3ee22cc227f5aca01e8a6a3a46ec60f5R38-R48

^ right here I'm scared that could be resource intensive if the state becomes large. Unfortunately we have to do that though because go treats maps as reference types so we have to copy into the new snapshot. I made sure to only call it once so we don't do the copy often (only when we get a new request or create a new watch).

I might've actually broken the wildcard since I'm going off the version map length but i'm not sure. It really only matters if we check the wildcard on the first request because then everything gets placed in the version tracking map after

[snowp]

I think it would be fine to leave as a copy for now, though having thought about this some more we might be able to get by just passing the reference if we reorder the code so that we cancel the watch before we subscribe/unsubscribe. All the writes to ResourceVersions happens in sub/unsub/responding to a watch, so if we make sure to never write to ResourceVersion while there is an active watch I think we might get away with it?

[snowp]

I think we do this so we can select between requests and watch triggers, maybe rephrase in terms of that?

[alecholmez]

Ah yes i'll make this change

[alecholmez]

@snowp I re-read this after I got a bit confused, wouldn't that only be the case if we were actually calling processDelta inside of the select? I think the the point of the select is to see if we need to close down the request channel if the stream closes itself. Maybe I'm misinterpreting that logic though

[alecholmez]

@snowp ah cool I'll try that. I have an expiremental branch right now I'm messing around in. I'll merge the code into here when I get it stable or have a question and might need your eyes. I think you are right though if we just cancel the watch before we mutate we should be okay

[alecholmez]

@snowp So I think we knocked down the data race. I can't seem to make ADS happy though. I've tried quite a few things and even reverted back to using a wildcard map by type but that didn't work either. I just pushed my latest work, I was able to clean up some things. Thankfully we didn't break xDS just ADS. If you've got time would you mind taking a look?

[snowp]

Seems like the nonce handling was incorrect, this seems to fix it

diff --git a/pkg/server/delta/v3/server.go b/pkg/server/delta/v3/server.go
index c64a6ef7..723293e2 100644
--- a/pkg/server/delta/v3/server.go
+++ b/pkg/server/delta/v3/server.go
@@ -140,19 +140,15 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
                                }
                        }

-                       var nonce string
                        // The node information might only be set on the first incoming delta discovery request, so store it here so we can
                        // reset it on subsequent requests that omit it.
                        if req.Node != nil {
                                node = req.Node
-                               nonce = req.GetResponseNonce()
                        } else {
                                req.Node = node
-
-                               // If we have no nonce, i.e. this is the first request on a delta stream, set one
-                               nonce = strconv.FormatInt(streamNonce, 10)
                        }

+
                        // type URL is required for ADS but is implicit for any other xDS stream
                        if defaultTypeURL == resource.AnyType {
                                if req.TypeUrl == "" {
@@ -177,12 +173,11 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
                                // This means that we're handling the first request for this type on the stream
                                watches.deltaStreamStates[typeURL] = state
                        }
-                       // versionMap := state.GetResourceVersions()

                        // cancel existing watch to (re-)request a newer version
                        watch, ok := watches.deltaResponses[typeURL]
                        // we verify nonce only if nonce is not initialized
-                       if !ok || watch.nonce == nonce {
+                       if !ok || watch.nonce == req.ResponseNonce {
                                // We must signal goroutine termination to prevent a race between the cancel closing the watch
                                // and the producer closing the watch.` `
                                if terminate, exists := watches.deltaTerminations[typeURL]; exists {`

[alecholmez]

hmm @snowp I made that change locally and ran the integration tests, it still seems like ADS is complaining. That nonce logic originally confused me so maybe its a SOTW problem too.

I noticed this after that change:

2021/05/19 22:09:31 request batch 1, ok 4, failed 0, pass true
2021/05/19 22:09:32 request batch 2, ok 4, failed 0, pass true
2021/05/19 22:09:32 request batch 3, ok 4, failed 0, pass true
2021/05/19 22:09:33 request batch 4, ok 4, failed 0, pass true
2021/05/19 22:09:33 server callbacks fetches=0 requests=0
2021/05/19 22:09:33 update snapshot v1
2021/05/19 22:09:33 node: test-id, sending delta response with resources: [name:"cluster-v1-0" type:EDS eds_cluster_config:{eds_config:{ads:{} resource_api_version:V3}} connect_timeout:{seconds:5} name:"cluster-v1-1" type:EDS eds_cluster_config:{eds_config:{ads:{} resource_api_version:V3}} connect_timeout:{seconds:5} name:"cluster-v1-2" type:EDS eds_cluster_config:{eds_config:{ads:{} resource_api_version:V3}} connect_timeout:{seconds:5} name:"cluster-v1-3" type:EDS eds_cluster_config:{eds_config:{ads:{} resource_api_version:V3}} connect_timeout:{seconds:5}] removed resources [cluster-v0-0 cluster-v0-1 cluster-v0-2 cluster-v0-3] wildcard: true
2021/05/19 22:09:33 node: test-id, sending delta response with resources: [] removed resources [cluster-v0-1 cluster-v0-2 cluster-v0-3 cluster-v0-0] wildcard: false
2021/05/19 22:09:33 node: test-id, sending delta response with resources: [name:"listener-9000" address:{socket_address:{address:"127.0.0.1" port_value:9000}} filter_chains:{filters:{name:"envoy.filters.network.http_connection_manager" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager]:{stat_prefix:"http" rds:{config_source:{ads:{} resource_api_version:V3} route_config_name:"route-v1-0"} http_filters:{name:"envoy.filters.http.router"} access_log:{name:"envoy.access_loggers.http_grpc" typed_config:{[type.googleapis.com/envoy.extensions.access_loggers.grpc.v3.HttpGrpcAccessLogConfig]:{common_config:{log_name:"echo" grpc_service:{envoy_grpc:{cluster_name:"xds_cluster"}} transport_api_version:V3}}}}}}}} name:"listener-9001" address:{socket_address:{address:"127.0.0.1" port_value:9001}} filter_chains:{filters:{name:"envoy.filters.network.http_connection_manager" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager]:{stat_prefix:"http" rds:{config_source:{ads:{} resource_api_version:V3} route_config_name:"route-v1-1"} http_filters:{name:"envoy.filters.http.router"} access_log:{name:"envoy.access_loggers.http_grpc" typed_config:{[type.googleapis.com/envoy.extensions.access_loggers.grpc.v3.HttpGrpcAccessLogConfig]:{common_config:{log_name:"echo" grpc_service:{envoy_grpc:{cluster_name:"xds_cluster"}} transport_api_version:V3}}}}}}}} name:"listener-9002" address:{socket_address:{address:"127.0.0.1" port_value:9002}} filter_chains:{filters:{name:"envoy.filters.network.tcp_proxy" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy]:{stat_prefix:"tcp" cluster:"cluster-v1-2"}}}} name:"listener-9003" address:{socket_address:{address:"127.0.0.1" port_value:9003}} filter_chains:{filters:{name:"envoy.filters.network.tcp_proxy" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy]:{stat_prefix:"tcp" cluster:"cluster-v1-3"}}}}] removed resources [] wildcard: true
2021/05/19 22:09:33 node: test-id, sending delta response with resources: [] removed resources [route-v0-1 route-v0-0] wildcard: false
2021/05/19 22:09:33 open delta watch ID:8 for type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment Resources:map[] from nodeID: "test-id", system version "v1"
2021/05/19 22:09:33 request batch 0, ok 4, failed 0, pass true
2021/05/19 22:09:34 open delta watch ID:9 for type.googleapis.com/envoy.config.listener.v3.Listener Resources:map[listener-9000:7259ef2f0651278a788a960b79d5464df1a8680ecc92c005a9f3b2227541889f listener-9001:ae917bd5acfb95fa7a25272cfec9430bbfdebc8af5385d78816410b621d05b0f listener-9002:1aa2c488d22ff64037f6de2730b94feee55def43511ba075283af02da65b7284 listener-9003:fa398e25787969541474d32346e3c050e0cb4c76286a59fad94afa50a47c9054] from nodeID: "test-id", system version "v1"
2021/05/19 22:09:34 open delta watch ID:10 for type.googleapis.com/envoy.config.route.v3.RouteConfiguration Resources:map[] from nodeID: "test-id", system version "v1"
2021/05/19 22:09:34 request batch 1, ok 0, failed 4, pass true
2021/05/19 22:09:35 request batch 2, ok 0, failed 4, pass true
2021/05/19 22:09:35 request batch 3, ok 0, failed 4, pass true
2021/05/19 22:09:36 request batch 4, ok 0, failed 4, pass true
2021/05/19 22:09:36 server callbacks fetches=0 requests=0
2021/05/19 22:09:36 update snapshot v2
2021/05/19 22:09:36 node: test-id, sending delta response with resources: [name:"listener-9002" address:{socket_address:{address:"127.0.0.1" port_value:9002}} filter_chains:{filters:{name:"envoy.filters.network.tcp_proxy" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy]:{stat_prefix:"tcp" cluster:"cluster-v2-2"}}}} name:"listener-9003" address:{socket_address:{address:"127.0.0.1" port_value:9003}} filter_chains:{filters:{name:"envoy.filters.network.tcp_proxy" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy]:{stat_prefix:"tcp" cluster:"cluster-v2-3"}}}} name:"listener-9000" address:{socket_address:{address:"127.0.0.1" port_value:9000}} filter_chains:{filters:{name:"envoy.filters.network.http_connection_manager" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager]:{stat_prefix:"http" rds:{config_source:{ads:{} resource_api_version:V3} route_config_name:"route-v2-0"} http_filters:{name:"envoy.filters.http.router"} access_log:{name:"envoy.access_loggers.http_grpc" typed_config:{[type.googleapis.com/envoy.extensions.access_loggers.grpc.v3.HttpGrpcAccessLogConfig]:{common_config:{log_name:"echo" grpc_service:{envoy_grpc:{cluster_name:"xds_cluster"}} transport_api_version:V3}}}}}}}} name:"listener-9001" address:{socket_address:{address:"127.0.0.1" port_value:9001}} filter_chains:{filters:{name:"envoy.filters.network.http_connection_manager" typed_config:{[type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager]:{stat_prefix:"http" rds:{config_source:{ads:{} resource_api_version:V3} route_config_name:"route-v2-1"} http_filters:{name:"envoy.filters.http.router"} access_log:{name:"envoy.access_loggers.http_grpc" typed_config:{[type.googleapis.com/envoy.extensions.access_loggers.grpc.v3.HttpGrpcAccessLogConfig]:{common_config:{log_name:"echo" grpc_service:{envoy_grpc:{cluster_name:"xds_cluster"}} transport_api_version:V3}}}}}}}}] removed resources [] wildcard: true
2021/05/19 22:09:36 request batch 0, ok 0, failed 4, pass false
2021/05/19 22:09:36 open delta watch ID:11 for type.googleapis.com/envoy.config.listener.v3.Listener Resources:map[listener-9000:a6151a0a3bd651278a1ce357cce0cd4536444b4993f23adf90571e1bb7418571 listener-9001:87c8c9530cf22a0579a7a7f24bc2ef35be591e64f4e40bb9f3b8f9ef130b0349 listener-9002:4b1d43e0ffff3b188ca245591bff0e5167f7da087f49225fc57b6260c37f55df listener-9003:4252672c2113205745dfed9d7d6f6a61e9b31ddb885464227280dce91a08e775] from nodeID: "test-id", system version "v2"
2021/05/19 22:09:37 request batch 1, ok 0, failed 4, pass false
2021/05/19 22:09:37 request batch 2, ok 0, failed 4, pass false
2021/05/19 22:09:38 request batch 3, ok 0, failed 4, pass false
2021/05/19 22:09:38 request batch 4, ok 0, failed 4, pass false
2021/05/19 22:09:39 server callbacks fetches=0 requests=0
2021/05/19 22:09:39 failed all requests in a run 2
make: *** [Makefile:71: integration.ads.delta] Error 1

Note this is for ADS only, xDS is perfectly okay. The unit tests pass no problem though so it's a tough problem to track, maybe some odd behavior in envoy? There seems to be a pattern, the initial request comes through fine the test handlers are able to make their calls appropriately, but as soon as we set a snapshot with a different version it blows up

[snowp]

I was seeing seeing it go green locally earlier I'm pretty sure, but seeing these failures again locally so I guess I'll take another look. The logs from the integration tests are pretty useless for debugging this, so locally im adding a bunch more useful stuff that I can push upstream later on

[snowp]

Seems like removing the nonce check altogether fixes thing locally for me. Seems like requests that are just changes to subscription state and not in response to a response (hah) doesn't set a nonce, but we still need to process these requests and respond with any new resources it might be interested in.

Not sure why this is only a problem on ADS

[alecholmez]

@snowp hm yup that seemed to do the trick. I also noticed the logs seem a lot more logical now in the integration tests. Thanks for the extra eyes! I'll push the change now

[snowp]

Might as well, no need to pollute the git history

[alecholmez]

RE this (not sure why github didn't let me respond):

Oh I misread the size arguments as buffers. Maybe move the comment closer to the channel declaration?

Looking at this I wonder if we can use a single map instead of having three different maps keying of a type URL, but we can > defer that to another PR if you'd prefer

I actually agree, I'll combine the data into the watch struct since we already have something to track information in the server. I agree it's much cleaner to have just one map by type

[snowp]

what case?

[alecholmez]

Ah yes I'll make this a bit more descriptive. We might want to just remove this comment since nonces aren't particularly useful in the delta spec anyways?

[snowp]

Alright I think this is in a good state to ship and iterate, thanks!