SharedStore provider #1147

danrspencer · 2023-02-21T10:16:36Z

Motivation

For additional context see: #1128

In large controllers you can end up with multiple instances of the same store, this has the potential to case a split brain problem where different bits of the application have a different view of the cluster.

This can be solved by creating stores at the top level of the application and passing them down; but the definition of the stores is then required to be at a distance from the usage sites, smearing responsibilities around the codebase.

Solution

The SharedStore struct will always return a clone of the same Store for a given scope and ListParams. The aim is to minimise the knowledge that must exist at a distance (only the types must be known, which can be cleanly asked for in controller interfaces), while still allowing separate parts of the application to share stores.

…event subscribers Signed-off-by: Dan Spencer <[email protected]>

1. Remove expensive cloning of Kubernetes objects 2. Allow propogation of err events Signed-off-by: Dan Spencer <[email protected]>

Co-authored-by: Eirik A <[email protected]> Signed-off-by: Dan Spencer <[email protected]>

Signed-off-by: Dan Spencer <[email protected]>

Also renamed WatchStreamExt::subscribe to WatchStreamExt::stream_subscribe. The compiler was unable to tell if we were trying to call WatchStreamExt::subscribe or StreamSubscribe::subscribe when they were named the same. e.g. this code would not compile: let stream_subscribe = stream.subscribe(); let subscription = stream_subscribe.subscribe(); Signed-off-by: Dan Spencer <[email protected]>

Signed-off-by: Dan Spencer <[email protected]>

Co-authored-by: Eirik A <[email protected]> Signed-off-by: Dan Spencer <[email protected]>

Signed-off-by: Dan Spencer <[email protected]>

… feature Signed-off-by: Dan Spencer <[email protected]>

…or the same list params for a given type. Also allows subscriptions to those stores. Signed-off-by: Dan Spencer <[email protected]>

danrspencer · 2023-02-21T10:18:38Z

@clux I've pulled out just the SharedStore bit of my bigger PR (selfishly because this is the most useful bit for us). Obviously we're missing docs and some tests here but before I spend more time on it I wanted to discuss if this is a direction worth pursuing.

codecov-commenter · 2023-02-21T10:28:36Z

Codecov Report

Merging #1147 (7c1cee1) into main (53ad9ee) will decrease coverage by 1.00%.
The diff coverage is 0.00%.

📣 This organization is not using Codecov’s GitHub App Integration. We recommend you install it so Codecov can continue to function properly for your repositories. Learn more

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1147      +/-   ##
==========================================
- Coverage   72.79%   71.80%   -1.00%     
==========================================
  Files          66       68       +2     
  Lines        5066     5136      +70     
==========================================
  Hits         3688     3688              
- Misses       1378     1448      +70

Impacted Files	Coverage Δ
kube-core/src/params.rs	`78.52% <ø> (ø)`
kube-runtime/src/reflector/mod.rs	`100.00% <ø> (ø)`
kube-runtime/src/reflector/shared_store.rs	`0.00% <0.00%> (ø)`
kube-runtime/src/utils/mod.rs	`64.15% <ø> (ø)`
kube-runtime/src/utils/stream_subscribe.rs	`0.00% <0.00%> (ø)`
kube-runtime/src/utils/watch_ext.rs	`0.00% <0.00%> (ø)`

clux

some initial comments. i do really like the exploration here and the map you've set up definitely has potential. you'll have to excuse me while I think long term about this for a little bit.

kube-runtime/src/reflector/shared_store.rs

clux · 2023-02-23T08:53:39Z

kube-runtime/src/reflector/shared_store.rs

+pub struct SharedStore<K, W = WatcherFactory<K>>
+where
+    K: 'static + Resource + Clone + DeserializeOwned + Debug + Send,
+    K::DynamicType: Hash + Eq,
+    W: CreateWatcher<K>,
+{
+    watcher_provider: W,


Suggested change

pub struct SharedStore<K, W = WatcherFactory<K>>

where

K: 'static + Resource + Clone + DeserializeOwned + Debug + Send,

K::DynamicType: Hash + Eq,

W: CreateWatcher<K>,

{

watcher_provider: W,

pub struct SharedStore<K>

where

K: 'static + Resource + Clone + DeserializeOwned + Debug + Send,

K::DynamicType: Hash + Eq,

{

I feel like we can get away with not having a second generic parameter for this. The WatcherFactory's impls have exactly the same constraints as the SharedStore methods that construct them. Did you have a particular use case in mind for the WatcherFactory, or is it mainly for tests?

It's purely for the tests. I don't particularly like test induced damage to non-test code like this ... but I also don't like not unit testing stuff 😅. Do you have a better suggestion on how to test something like this?

Possibly worth noting, users cannot create an instance of the SharedStore where W is anything other than a WatcherFactory since non of the properties are public and the only constructor we expose is for the variation using a WatcherFactory. I don't think end users ever need to know or care about the W generic 🤔.

hm, yeah, it's probably not that bad, I would also ideally have the damage confined to the tests if possible. was thinking there might be some ways to mock just one of the functions (like setup_reflector) used in this module, but maybe it's only mocktopus that lets you do functions (and that's nightly only).

i think it probably is OK. mocking watch streams is hard to do raw. i would otherwise suggest the mocked client with tower_test as is used in client/api.

clux · 2023-02-24T09:00:23Z

kube-runtime/src/reflector/shared_store.rs

+    watcher: BoxStream<'static, watcher::Result<Event<K>>>,
+) -> (Store<K>, BoxStreamSubscribe<K>)
+where
+    K: Resource + Clone + Send + Sync,


Where does the Sync requirement come from? It's propagated everywhere here, and neither watcher nor reflector should have more than Send required AFAIKT.

clux · 2023-02-24T09:06:47Z

kube-runtime/src/reflector/shared_store.rs

+        async fn spawn(self) {
+            tokio::spawn(async move {
+                self.run().for_each(|_| async {}).await;
+            });
+
+            tokio::task::yield_now().await;
+        }


I think this call should be explicitly awaited in tests without spawning (and can probably be done in a one-liner) because there's nothing we actually wait for having mocked the provider - so there shouldn't be a need for a task.

clux · 2023-02-24T09:08:51Z

kube-runtime/src/reflector/shared_store.rs

+pub struct SharedStore<K, W = WatcherFactory<K>>
+where
+    K: 'static + Resource + Clone + DeserializeOwned + Debug + Send,
+    K::DynamicType: Hash + Eq,
+    W: CreateWatcher<K>,
+{
+    watcher_provider: W,


hm, yeah, it's probably not that bad, I would also ideally have the damage confined to the tests if possible. was thinking there might be some ways to mock just one of the functions (like setup_reflector) used in this module, but maybe it's only mocktopus that lets you do functions (and that's nightly only).

i think it probably is OK. mocking watch streams is hard to do raw. i would otherwise suggest the mocked client with tower_test as is used in client/api.

clux · 2023-02-24T09:09:57Z

kube-runtime/src/reflector/shared_store.rs

+    #[derive(Clone, Debug)]
+    struct TestContext {
+        reconciled: Arc<Mutex<bool>>,
+    }


TestContext is unused

clux · 2023-02-24T09:10:48Z

kube-runtime/src/reflector/shared_store.rs

+            stream::unfold(events, |mut events| async move {
+                match events.pop_front() {
+                    Some(event) => Some((Ok(event), events)),
+                    // if there's nothing left we block to simulate waiting for a change
+                    None => futures::future::pending().await,
+                }
+            })


i think you can just convert the vec to a stream here and return that without doing an unfold

clux · 2023-02-24T09:11:56Z

kube-runtime/src/reflector/shared_store.rs

+    enum TestError {
+        TestError,
+    }
+
+    impl Debug for TestError {
+        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+            unimplemented!()
+        }
+    }
+
+    impl Display for TestError {
+        fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+            unimplemented!()
+        }
+    }
+
+    impl std::error::Error for TestError {}


TestError also unused ^
EDIT: also ClonedState

kube-runtime/src/utils/watch_ext.rs

clux · 2023-02-24T09:18:48Z

I think the basic idea of this struct is good, but it's harder to see where to go optimally here without seeing how you hook it up to a controller (because it doesn't help much to generate stores up front when Controller ultimately creates their own stores).

The potential setup you have here can work, but I can also see it having potential as a more global stream manager that maintains watcher streams for all types regardless of kind. The latter might be an easier thing to integrate into Controller as an endgame for #1080. At any rate, would be interested in hearing your thoughts on integration here.

danrspencer · 2023-02-24T13:17:51Z

I think the basic idea of this struct is good, but it's harder to see where to go optimally here without seeing how you hook it up to a controller (because it doesn't help much to generate stores up front when Controller ultimately creates their own stores).

I haven't had chance to play around with this yet, but the idea in my head is something along the lines of:

let lp = ListParams::Default();
// I'm still in two minds if the SharedStore functions should just always return a store and a stream
// and you just drop one if you don't care about it
let store = deploy_shared_store.all(lp);
let stream = shared_store.subscribe_all(lp);

let pod_stream = pod_shared_store.subscribe_all(lp);

Controller::new_from_store(store, stream)
     .watches_stream(pod_stream, |pod| {
            Some(ObjectRef::new(&pod.name_unchecked()))
     });

The potential setup you have here can work, but I can also see it having potential as a more global stream manager that maintains watcher streams for all types regardless of kind. The latter might be an easier thing to integrate into Controller as an endgame for #1080. At any rate, would be interested in hearing your thoughts on integration here.

I did have a go at creating an "all types" version of this, but I couldn't think of a way of creating it without losing compile time type safety. The only way I can think of would be to change the reflectors property to be indexed by something like (ApiResource, Option<String>, ListParams) then strongly typing them on the way out to the user. That feels horrible to me, but I guess the streams will always come into the Rust untyped and must be cast to a known type at some point.

danrspencer · 2023-03-29T09:23:11Z

Just wanted to drive by and say this isn't forgotten about, I just haven't had time to work on it recently. Hopefully I'll be able to bring some focus back to it soon.

clux · 2023-03-30T07:01:49Z

Appreciate it! We are early-stage exploring the other side of the coin here with the controller interface. See #1163 for some context.

clux · 2023-06-07T17:35:22Z

Hey @danrspencer , I know this might not be on your radar anymore, just wondering what you ended up doing or if you still have interest in this.

It is certain that we still need at least two more things to close out #1080;

waiting for reflectors to fully init before starting the controller
hooking up your streamsubscribe (that has made it in under unstable in your other PR) through some arc-wrapping

if you have solutions elsewhere it'd be good to know about, and i'll open other issues

danrspencer · 2023-06-07T19:34:55Z

Hey,

Yep it’s still on my radar. The work I was doing that I could consider controller improvements part of got punted down the backlog but we should be picking it up again soon. We’ve got at least a couple of concrete cases in our codebase where not having shared stores is causing race condition bugs so I should be able to find some time to get back to this as part of solving those.

danrspencer · 2023-09-08T14:04:34Z

@clux I'm going to close this PR. I'm moving jobs and realistically I'm probably not going to get chance to pick this up again. Thanks for your help and feedback getting it this far. Hopefully some of the ideas and discussions will help solve the problem.

clux · 2023-09-08T14:23:25Z

Ok 👍

Thanks for your efforts and best of luck!

danrspencer and others added 14 commits February 17, 2023 13:25

Added stream_subscribe which allows watch streams to have additional …

8efdc15

…event subscribers Signed-off-by: Dan Spencer <[email protected]>

StreamSubscribe now wraps items in arcs so that:

8d77e0a

1. Remove expensive cloning of Kubernetes objects 2. Allow propogation of err events Signed-off-by: Dan Spencer <[email protected]>

Renamed watch_ext subscribable to subscribe

934faec

Co-authored-by: Eirik A <[email protected]> Signed-off-by: Dan Spencer <[email protected]>

StreamSubscribe now allows subscribers how to handle lagging

a85fe3f

Signed-off-by: Dan Spencer <[email protected]>

Fixed clippy errors in StreamSubscribe

c3e6663

Signed-off-by: Dan Spencer <[email protected]>

Fixed grammar in StreamSubscribe docs

0899436

Signed-off-by: Dan Spencer <[email protected]>

Fixed rustfmt errors in StreamSubscribe

a57184a

Signed-off-by: Dan Spencer <[email protected]>

Put StreamSubscribe behind a feature flag unstable_runtime_subscribe

8b28a9b

Signed-off-by: Dan Spencer <[email protected]>

Update kube-runtime/src/utils/mod.rs

efb4166

Co-authored-by: Eirik A <[email protected]> Signed-off-by: Dan Spencer <[email protected]>

Fixed rustfmt error in kube-runtime utils mod.rs

a320b19

Signed-off-by: Dan Spencer <[email protected]>

Fixed incorrect feature flag usage for the unstable-runtime-subscribe…

e152543

… feature Signed-off-by: Dan Spencer <[email protected]>

Created a SharedStore which can return the same instance of a store f…

3b67d6d

…or the same list params for a given type. Also allows subscriptions to those stores. Signed-off-by: Dan Spencer <[email protected]>

Merge branch 'main' into shared_stores_2

7c1cee1

Merge branch 'main' into shared_stores_2

4f0418c

clux reviewed Feb 22, 2023

View reviewed changes

kube-runtime/src/reflector/shared_store.rs Show resolved Hide resolved

kube-runtime/src/reflector/shared_store.rs Show resolved Hide resolved

clux mentioned this pull request Feb 22, 2023

Create ListParams union + filter logic for watcher streams #1148

Open

2 tasks

clux reviewed Feb 23, 2023

View reviewed changes

clux reviewed Feb 24, 2023

View reviewed changes

kube-runtime/src/utils/watch_ext.rs Show resolved Hide resolved

danrspencer closed this Feb 24, 2023

danrspencer reopened this Feb 24, 2023

clux mentioned this pull request Apr 7, 2023

Sharing watch streams and caches between Controllers #1080

Open

7 tasks

danrspencer closed this Sep 8, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SharedStore provider #1147

SharedStore provider #1147

danrspencer commented Feb 21, 2023

danrspencer commented Feb 21, 2023

codecov-commenter commented Feb 21, 2023

clux left a comment

clux Feb 23, 2023 •

edited

Loading

danrspencer Feb 23, 2023

danrspencer Feb 23, 2023

clux Feb 24, 2023

clux Feb 24, 2023

clux Feb 24, 2023

clux Feb 24, 2023

clux Feb 24, 2023

clux Feb 24, 2023

clux Feb 24, 2023 •

edited

Loading

clux commented Feb 24, 2023

danrspencer commented Feb 24, 2023 •

edited

Loading

danrspencer commented Mar 29, 2023

clux commented Mar 30, 2023

clux commented Jun 7, 2023 •

edited

Loading

danrspencer commented Jun 7, 2023

danrspencer commented Sep 8, 2023

clux commented Sep 8, 2023

SharedStore provider #1147

SharedStore provider #1147

Conversation

danrspencer commented Feb 21, 2023

Motivation

Solution

danrspencer commented Feb 21, 2023

codecov-commenter commented Feb 21, 2023

Codecov Report

clux left a comment

Choose a reason for hiding this comment

clux Feb 23, 2023 • edited Loading

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

clux Feb 24, 2023 • edited Loading

Choose a reason for hiding this comment

clux commented Feb 24, 2023

danrspencer commented Feb 24, 2023 • edited Loading

danrspencer commented Mar 29, 2023

clux commented Mar 30, 2023

clux commented Jun 7, 2023 • edited Loading

danrspencer commented Jun 7, 2023

danrspencer commented Sep 8, 2023

clux commented Sep 8, 2023

clux Feb 23, 2023 •

edited

Loading

clux Feb 24, 2023 •

edited

Loading

danrspencer commented Feb 24, 2023 •

edited

Loading

clux commented Jun 7, 2023 •

edited

Loading