client.rs 13.8 KB
Newer Older
1
2
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
Ryan Olson's avatar
Ryan Olson committed
3

4
5
6
7
8
9
10
11
use std::sync::Arc;
use std::{collections::HashMap, time::Duration};

use anyhow::Result;
use arc_swap::ArcSwap;
use futures::StreamExt;
use tokio::net::unix::pipe::Receiver;

12
use crate::{
13
14
    component::{Endpoint, Instance},
    pipeline::async_trait,
15
16
17
18
19
    pipeline::{
        AddressedPushRouter, AddressedRequest, AsyncEngine, Data, ManyOut, PushRouter, RouterMode,
        SingleIn,
    },
    storage::key_value_store::{KeyValueStoreManager, WatchEvent},
20
21
    traits::DistributedRuntimeProvider,
    transports::etcd::Client as EtcdClient,
Ryan Olson's avatar
Ryan Olson committed
22
23
};

24
25
26
27
#[derive(Clone, Debug)]
pub struct Client {
    // This is me
    pub endpoint: Endpoint,
28
29
    // These are the remotes I know about from watching key-value store
    pub instance_source: Arc<tokio::sync::watch::Receiver<Vec<Instance>>>,
30
    // These are the instance source ids less those reported as down from sending rpc
31
    instance_avail: Arc<ArcSwap<Vec<u64>>>,
32
    // These are the instance source ids less those reported as busy (above threshold)
33
    instance_free: Arc<ArcSwap<Vec<u64>>>,
34
35
}

36
impl Client {
37
38
    // Client with auto-discover instances using key-value store
    pub(crate) async fn new(endpoint: Endpoint) -> Result<Self> {
39
40
41
42
        tracing::debug!(
            "Client::new_dynamic: Creating dynamic client for endpoint: {}",
            endpoint.path()
        );
43
        let instance_source = Self::get_or_create_dynamic_instance_source(&endpoint).await?;
44
45
46
47
        tracing::debug!(
            "Client::new_dynamic: Got instance source for endpoint: {}",
            endpoint.path()
        );
48

49
        let client = Client {
50
            endpoint: endpoint.clone(),
51
            instance_source: instance_source.clone(),
52
            instance_avail: Arc::new(ArcSwap::from(Arc::new(vec![]))),
53
            instance_free: Arc::new(ArcSwap::from(Arc::new(vec![]))),
54
        };
55
56
57
58
        tracing::debug!(
            "Client::new_dynamic: Starting instance source monitor for endpoint: {}",
            endpoint.path()
        );
59
        client.monitor_instance_source();
60
61
62
63
        tracing::debug!(
            "Client::new_dynamic: Successfully created dynamic client for endpoint: {}",
            endpoint.path()
        );
64
        Ok(client)
65
66
67
68
69
70
71
72
73
74
75
    }

    pub fn path(&self) -> String {
        self.endpoint.path()
    }

    /// The root etcd path we watch in etcd to discover new instances to route to.
    pub fn etcd_root(&self) -> String {
        self.endpoint.etcd_root()
    }

76
    /// Instances available from watching key-value store
77
    pub fn instances(&self) -> Vec<Instance> {
78
        self.instance_source.borrow().clone()
79
80
    }

81
    pub fn instance_ids(&self) -> Vec<u64> {
82
83
84
        self.instances().into_iter().map(|ep| ep.id()).collect()
    }

85
    pub fn instance_ids_avail(&self) -> arc_swap::Guard<Arc<Vec<u64>>> {
86
87
88
        self.instance_avail.load()
    }

89
    pub fn instance_ids_free(&self) -> arc_swap::Guard<Arc<Vec<u64>>> {
90
91
92
        self.instance_free.load()
    }

93
94
    /// Wait for at least one Instance to be available for this Endpoint
    pub async fn wait_for_instances(&self) -> Result<Vec<Instance>> {
95
96
97
98
        tracing::debug!(
            "wait_for_instances: Starting wait for endpoint: {}",
            self.endpoint.path()
        );
99
100
101
102
103
104
105
106
107
108
109
110
111
        let mut rx = self.instance_source.as_ref().clone();
        // wait for there to be 1 or more endpoints
        let mut iteration = 0;
        let mut instances: Vec<Instance>;
        loop {
            instances = rx.borrow_and_update().to_vec();
            tracing::debug!(
                "wait_for_instances: iteration={}, current_instance_count={}, endpoint={}",
                iteration,
                instances.len(),
                self.endpoint.path()
            );
            if instances.is_empty() {
112
                tracing::debug!(
113
114
115
116
117
118
119
120
121
122
123
                    "wait_for_instances: No instances yet, waiting for change notification for endpoint: {}",
                    self.endpoint.path()
                );
                rx.changed().await?;
                tracing::debug!(
                    "wait_for_instances: Change notification received for endpoint: {}",
                    self.endpoint.path()
                );
            } else {
                tracing::info!(
                    "wait_for_instances: Found {} instance(s) for endpoint: {}",
124
125
126
                    instances.len(),
                    self.endpoint.path()
                );
127
                break;
128
            }
129
            iteration += 1;
130
131
132
133
        }
        Ok(instances)
    }

134
    /// Mark an instance as down/unavailable
135
    pub fn report_instance_down(&self, instance_id: u64) {
136
137
138
139
140
141
        let filtered = self
            .instance_ids_avail()
            .iter()
            .filter_map(|&id| if id == instance_id { None } else { Some(id) })
            .collect::<Vec<_>>();
        self.instance_avail.store(Arc::new(filtered));
142
143
144
145

        tracing::debug!("inhibiting instance {instance_id}");
    }

146
    /// Update the set of free instances based on busy instance IDs
147
    pub fn update_free_instances(&self, busy_instance_ids: &[u64]) {
148
        let all_instance_ids = self.instance_ids();
149
        let free_ids: Vec<u64> = all_instance_ids
150
151
152
153
154
155
            .into_iter()
            .filter(|id| !busy_instance_ids.contains(id))
            .collect();
        self.instance_free.store(Arc::new(free_ids));
    }

156
    /// Monitor the key-value instance source and update instance_avail.
157
158
159
    fn monitor_instance_source(&self) {
        let cancel_token = self.endpoint.drt().primary_token();
        let client = self.clone();
160
161
162
163
164
        let endpoint_path = self.endpoint.path();
        tracing::debug!(
            "monitor_instance_source: Starting monitor for endpoint: {}",
            endpoint_path
        );
165
        tokio::task::spawn(async move {
166
            let mut rx = client.instance_source.as_ref().clone();
167
            let mut iteration = 0;
168
            while !cancel_token.is_cancelled() {
169
                let instance_ids: Vec<u64> = rx
170
171
172
173
                    .borrow_and_update()
                    .iter()
                    .map(|instance| instance.id())
                    .collect();
174

175
176
177
178
179
180
181
182
                tracing::debug!(
                    "monitor_instance_source: iteration={}, instance_count={}, instance_ids={:?}, endpoint={}",
                    iteration,
                    instance_ids.len(),
                    instance_ids,
                    endpoint_path
                );

183
184
                // TODO: this resets both tracked available and free instances
                client.instance_avail.store(Arc::new(instance_ids.clone()));
185
                client.instance_free.store(Arc::new(instance_ids.clone()));
186

187
188
189
190
                tracing::debug!(
                    "monitor_instance_source: instance source updated, endpoint={}",
                    endpoint_path
                );
191
192

                if let Err(err) = rx.changed().await {
193
194
195
196
197
                    tracing::error!(
                        "monitor_instance_source: The Sender is dropped: {}, endpoint={}",
                        err,
                        endpoint_path
                    );
198
199
                    cancel_token.cancel();
                }
200
                iteration += 1;
201
            }
202
203
204
205
            tracing::debug!(
                "monitor_instance_source: Monitor loop exiting for endpoint: {}",
                endpoint_path
            );
206
        });
207
208
209
210
    }

    async fn get_or_create_dynamic_instance_source(
        endpoint: &Endpoint,
211
    ) -> Result<Arc<tokio::sync::watch::Receiver<Vec<Instance>>>> {
212
213
214
215
        let drt = endpoint.drt();
        let instance_sources = drt.instance_sources();
        let mut instance_sources = instance_sources.lock().await;

216
217
218
219
220
        tracing::debug!(
            "get_or_create_dynamic_instance_source: Checking cache for endpoint: {}",
            endpoint.path()
        );

221
222
        if let Some(instance_source) = instance_sources.get(endpoint) {
            if let Some(instance_source) = instance_source.upgrade() {
223
224
225
226
                tracing::debug!(
                    "get_or_create_dynamic_instance_source: Found cached instance source for endpoint: {}",
                    endpoint.path()
                );
227
228
                return Ok(instance_source);
            } else {
229
230
231
232
                tracing::debug!(
                    "get_or_create_dynamic_instance_source: Cached instance source was dropped, removing for endpoint: {}",
                    endpoint.path()
                );
233
234
235
236
                instance_sources.remove(endpoint);
            }
        }

237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
        tracing::debug!(
            "get_or_create_dynamic_instance_source: Creating new instance source for endpoint: {}",
            endpoint.path()
        );

        let discovery = drt.discovery();
        let discovery_query = crate::discovery::DiscoveryQuery::Endpoint {
            namespace: endpoint.component.namespace.name.clone(),
            component: endpoint.component.name.clone(),
            endpoint: endpoint.name.clone(),
        };

        tracing::debug!(
            "get_or_create_dynamic_instance_source: Calling discovery.list_and_watch for query: {:?}",
            discovery_query
        );

        let mut discovery_stream = discovery
            .list_and_watch(discovery_query.clone(), None)
            .await?;

        tracing::debug!(
            "get_or_create_dynamic_instance_source: Got discovery stream for query: {:?}",
            discovery_query
        );

Ryan Olson's avatar
Ryan Olson committed
263
264
        let (watch_tx, watch_rx) = tokio::sync::watch::channel(vec![]);

265
        let secondary = endpoint.component.drt.runtime().secondary().clone();
Ryan Olson's avatar
Ryan Olson committed
266
267

        secondary.spawn(async move {
268
269
270
            tracing::debug!("endpoint_watcher: Starting for discovery query: {:?}", discovery_query);
            let mut map: HashMap<u64, Instance> = HashMap::new();
            let mut event_count = 0;
Ryan Olson's avatar
Ryan Olson committed
271
272

            loop {
273
                let discovery_event = tokio::select! {
Ryan Olson's avatar
Ryan Olson committed
274
                    _ = watch_tx.closed() => {
275
                        tracing::debug!("endpoint_watcher: all watchers have closed; shutting down for discovery query: {:?}", discovery_query);
Ryan Olson's avatar
Ryan Olson committed
276
277
                        break;
                    }
278
279
280
281
282
283
284
285
286
287
288
                    discovery_event = discovery_stream.next() => {
                        tracing::debug!("endpoint_watcher: Received stream event for discovery query: {:?}", discovery_query);
                        match discovery_event {
                            Some(Ok(event)) => {
                                tracing::debug!("endpoint_watcher: Got Ok event: {:?}", event);
                                event
                            },
                            Some(Err(e)) => {
                                tracing::error!("endpoint_watcher: discovery stream error: {}; shutting down for discovery query: {:?}", e, discovery_query);
                                break;
                            }
Ryan Olson's avatar
Ryan Olson committed
289
                            None => {
290
                                tracing::debug!("endpoint_watcher: watch stream has closed; shutting down for discovery query: {:?}", discovery_query);
Ryan Olson's avatar
Ryan Olson committed
291
292
293
294
295
296
                                break;
                            }
                        }
                    }
                };

297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
                event_count += 1;
                tracing::debug!("endpoint_watcher: Processing event #{} for discovery query: {:?}", event_count, discovery_query);

                match discovery_event {
                    crate::discovery::DiscoveryEvent::Added(discovery_instance) => {
                        match discovery_instance {
                            crate::discovery::DiscoveryInstance::Endpoint(instance) => {
                                tracing::debug!(
                                    "endpoint_watcher: Added endpoint instance_id={}, namespace={}, component={}, endpoint={}",
                                    instance.instance_id,
                                    instance.namespace,
                                    instance.component,
                                    instance.endpoint
                                );
                                map.insert(instance.instance_id, instance);
                            }
                            _ => {
                                tracing::debug!("endpoint_watcher: Ignoring non-endpoint instance (Model, etc.) for discovery query: {:?}", discovery_query);
Ryan Olson's avatar
Ryan Olson committed
315
316
                            }
                        }
317
318
319
320
321
322
323
324
                    }
                    crate::discovery::DiscoveryEvent::Removed(instance_id) => {
                        tracing::debug!(
                            "endpoint_watcher: Removed instance_id={} for discovery query: {:?}",
                            instance_id,
                            discovery_query
                        );
                        map.remove(&instance_id);
Ryan Olson's avatar
Ryan Olson committed
325
326
327
                    }
                }

328
                let instances: Vec<Instance> = map.values().cloned().collect();
329
330
331
332
333
                tracing::debug!(
                    "endpoint_watcher: Current map size={}, sending update for discovery query: {:?}",
                    instances.len(),
                    discovery_query
                );
Ryan Olson's avatar
Ryan Olson committed
334

335
                if watch_tx.send(instances).is_err() {
336
                    tracing::debug!("endpoint_watcher: Unable to send watch updates; shutting down for discovery query: {:?}", discovery_query);
Ryan Olson's avatar
Ryan Olson committed
337
338
339
340
                    break;
                }
            }

341
            tracing::debug!("endpoint_watcher: Completed for discovery query: {:?}, total events processed: {}", discovery_query, event_count);
Ryan Olson's avatar
Ryan Olson committed
342
343
344
            let _ = watch_tx.send(vec![]);
        });

345
        let instance_source = Arc::new(watch_rx);
346
        instance_sources.insert(endpoint.clone(), Arc::downgrade(&instance_source));
347
348
349
350
        tracing::debug!(
            "get_or_create_dynamic_instance_source: Successfully created and cached instance source for endpoint: {}",
            endpoint.path()
        );
351
        Ok(instance_source)
352
    }
Ryan Olson's avatar
Ryan Olson committed
353
}