Introduce xDS module and SotW stream #5342
Conversation
jrhee17
force-pushed
the
feature/xds-endpoint-step1
branch
from
ee1eb37
to
a4c49c5
Compare
xds/src/main/java/com/linecorp/armeria/xds/ClusterResourceParser.java
Outdated
Show resolved
Hide resolved
xds/src/main/java/com/linecorp/armeria/xds/EndpointResourceParser.java
Outdated
Show resolved
Hide resolved
xds/src/main/java/com/linecorp/armeria/xds/EndpointResourceParser.java
Outdated
Show resolved
Hide resolved
xds/src/main/java/com/linecorp/armeria/xds/ListenerResourceParser.java
Outdated
Show resolved
Hide resolved
|
|
||
| import com.linecorp.armeria.common.util.SafeCloseable; | ||
|
|
||
| class XdsStreamSubscriber implements SafeCloseable { |
There was a problem hiding this comment.
So the purpose of this class is to keep the reference?
Are there any other features to be added to this class?
There was a problem hiding this comment.
Later, I imagine class will be responsible for:
- Scheduling timeouts
- Caching values
- Notifying the watcher storage of updates
Lines 106 to 108 in 90a9349
xds/src/test/java/com/linecorp/armeria/xds/SotwXdsStreamTest.java
Outdated
Show resolved
Hide resolved
| interface XdsResponseHandler { | ||
|
|
||
| <T extends Message> void handleResponse( | ||
| ResourceParser<T> resourceParser, DiscoveryResponse value, SotwXdsStream sender); |
There was a problem hiding this comment.
Is resourceParser necessary? DiscoveryResponse has the type in it so we can use the type to get the corresponding ResourceParser.
There was a problem hiding this comment.
Do you mean there is no need to pass the resourceParser and we can re-run XdsResourceParserUtil.fromTypeUrl?
We could, but what would be the advantage of doing so?
There was a problem hiding this comment.
In my mind, the Stream layer will do all the heavy lifting to ensure communication with the control plane is maintained and up-to-date.
For this reason I would rather keep nonceMap, versionsMap in the XdsStream layer.
In order to do so, I think it makes sense that the ResourceType (or ResourceParser) is determined inside SotwXdsStream. I didn't see a need to re-compute this so I just passed it via parameter. I'd rather keep this the way it is, but let me know if you feel strongly about this 😄
There was a problem hiding this comment.
Understood. That makes sense. 👍
| <T extends Message> void handleResponse( | ||
| ResourceParser<T> resourceParser, DiscoveryResponse value, SotwXdsStream sender); | ||
|
|
||
| void handleReset(XdsStream sender); |
There was a problem hiding this comment.
Could you explain the definition of reset? 🤔
There was a problem hiding this comment.
Whenever the stream thinks its SubscriberStorage may not be synchronized with the control plane, then a reset is done. This can be the case if a connection is closed unexpectedly, or if a stream is doing a cold start.
When a reset is performed, all of the currently watched types and resources are sent to the control plane server. Note that this is a package-private interface, and is exposed for testing purposes.
ref:
Codecov ReportAttention:
Additional details and impacted files@@ Coverage Diff @@
## main #5342 +/- ##
============================================
+ Coverage 73.93% 73.96% +0.03%
- Complexity 20098 20159 +61
============================================
Files 1730 1741 +11
Lines 74142 74353 +211
Branches 9461 9481 +20
============================================
+ Hits 54815 54998 +183
- Misses 14852 14869 +17
- Partials 4475 4486 +11 ☔ View full report in Codecov by Sentry. |
jrhee17
force-pushed
the
feature/xds-endpoint-step1
branch
from
2ed3ae3
to
bfd4d90
Compare
| Thread.sleep(100); | ||
| assertThat(responseHandler.getResponses()).isEmpty(); |
There was a problem hiding this comment.
Global comment: Could we use Awaitility?
| import com.google.common.base.MoreObjects; | ||
| import com.google.protobuf.Message; | ||
|
|
||
| abstract class ResourceParser<T extends Message> { |
There was a problem hiding this comment.
I'm not sure the generic type parameter provides type safety in xDS implementation.
As the input is dynamically determined by typeUrl at runtime, the compile time constrain does not seem to work.
https://github.com/line/armeria/pull/5342/files#diff-463524dcc65fabda54bb06752b0e248879c7bc1e5b367d0e8885bfbd23aa785bR209
What do you think of removing <T extends Message> and checking the type with XdsType only?
Motivation:
We would like to introduce a xDS integration module.
In order to fetch resources from xDS module, I would like to introduce a
Streamobject which represents a single persistent connection to a remote control plane server. TheStreamwill query aSubscriberStorageto check which resources are subscribed and update the request accordingly.Once a response is recieved, the type will be used to fetch a
ResourceParserwhich will later parse the response. Afterwards, the parameters will be sent over to aResponseHandlerwhich will handle the response and send either anackResponseornackResponseto keep the feedback loop open. Also, aresourcesUpdatedcan be sent to indicate the subscribed resources have been changed.Modifications:
SotwXdsStreamwhich represents a persistent connection to a control plane serverResourceParserwhich will be responsible for parsing a response in a format readily sendable in a nack responseXdsStreamSubscriberandSubscriberStoragewhich represent the currently subscribed resourcesResult: