xds_server/services/
eds.rs

1//! Endpoint Discovery Service (EDS) implementation.
2//!
3//! EDS provides endpoint configuration to Envoy proxies.
4
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use tokio::sync::mpsc;
9use tokio_stream::wrappers::ReceiverStream;
10use tonic::{Request, Response, Status, Streaming};
11use tracing::{debug, error, info, instrument};
12
13use xds_cache::ShardedCache;
14use xds_core::{NodeHash, ResourceRegistry, TypeUrl};
15
16use crate::delta::{delta_response_to_proto, ClientResourceState, DeltaHandler};
17use crate::sotw::SotwHandler;
18use crate::stream::StreamContext;
19
20// Re-export the data-plane-api types
21pub use xds_types::envoy::service::discovery::v3::{DiscoveryRequest, DiscoveryResponse};
22pub use xds_types::envoy::service::endpoint::v3::endpoint_discovery_service_server::{
23    EndpointDiscoveryService, EndpointDiscoveryServiceServer,
24};
25pub use xds_types::envoy::service::discovery::v3::{
26    DeltaDiscoveryRequest, DeltaDiscoveryResponse,
27};
28
29/// Endpoint Discovery Service.
30#[derive(Debug, Clone)]
31pub struct EdsService {
32    /// Shared cache.
33    cache: Arc<ShardedCache>,
34    /// Resource registry.
35    registry: Arc<ResourceRegistry>,
36    /// SotW handler.
37    sotw_handler: Arc<SotwHandler>,
38    /// Delta handler.
39    delta_handler: Arc<DeltaHandler>,
40}
41
42impl EdsService {
43    /// Create a new EDS service.
44    pub fn new(cache: Arc<ShardedCache>, registry: Arc<ResourceRegistry>) -> Self {
45        let sotw_handler = Arc::new(SotwHandler::new(Arc::clone(&cache), Arc::clone(&registry)));
46        let delta_handler = Arc::new(DeltaHandler::new(Arc::clone(&cache), Arc::clone(&registry)));
47        Self {
48            cache,
49            registry,
50            sotw_handler,
51            delta_handler,
52        }
53    }
54
55    /// Create a new EDS service with shared protocol handlers.
56    ///
57    /// This allows sharing handlers across multiple services for better
58    /// resource efficiency.
59    pub fn new_with_handlers(
60        cache: Arc<ShardedCache>,
61        registry: Arc<ResourceRegistry>,
62        sotw_handler: Arc<SotwHandler>,
63        delta_handler: Arc<DeltaHandler>,
64    ) -> Self {
65        Self {
66            cache,
67            registry,
68            sotw_handler,
69            delta_handler,
70        }
71    }
72
73    /// Get the type URL for endpoints.
74    #[inline]
75    pub fn type_url() -> &'static str {
76        TypeUrl::ENDPOINT
77    }
78
79    /// Get a reference to the cache.
80    #[allow(dead_code)]
81    pub fn cache(&self) -> &ShardedCache {
82        &self.cache
83    }
84
85    /// Get a reference to the registry.
86    #[allow(dead_code)]
87    pub fn registry(&self) -> &ResourceRegistry {
88        &self.registry
89    }
90
91    /// Convert this service into a tonic service for use with Server::add_service.
92    pub fn into_service(self) -> EndpointDiscoveryServiceServer<Self> {
93        EndpointDiscoveryServiceServer::new(self)
94    }
95
96    /// Convert a SotW response to a proto DiscoveryResponse.
97    ///
98    /// Returns an error if any resource fails to encode, rather than
99    /// silently dropping it.
100    fn convert_response(
101        &self,
102        response: crate::sotw::SotwResponse,
103    ) -> Result<DiscoveryResponse, Status> {
104        use xds_types::google::protobuf::Any;
105
106        let resources: Vec<Any> = response
107            .resources
108            .iter()
109            .map(|r| {
110                r.encode().map(|encoded| Any {
111                    type_url: encoded.type_url.clone(),
112                    value: encoded.value.clone(),
113                })
114            })
115            .collect::<Result<Vec<_>, _>>()
116            .map_err(|e| Status::internal(format!("failed to encode resource: {}", e)))?;
117
118        Ok(DiscoveryResponse {
119            version_info: response.version_info,
120            resources,
121            type_url: TypeUrl::ENDPOINT.to_string(),
122            nonce: response.nonce,
123            canary: false,
124            control_plane: None,
125            resource_errors: vec![],
126        })
127    }
128}
129
130/// Response stream type for EDS.
131pub type EdsResponseStream = ReceiverStream<Result<DiscoveryResponse, Status>>;
132
133/// Delta response stream type for EDS.
134pub type EdsDeltaResponseStream = ReceiverStream<Result<DeltaDiscoveryResponse, Status>>;
135
136#[async_trait]
137impl EndpointDiscoveryService for EdsService {
138    type StreamEndpointsStream = EdsResponseStream;
139
140    #[instrument(skip(self, request), name = "eds_stream")]
141    async fn stream_endpoints(
142        &self,
143        request: Request<Streaming<DiscoveryRequest>>,
144    ) -> Result<Response<Self::StreamEndpointsStream>, Status> {
145        let mut stream = request.into_inner();
146        let (tx, rx) = mpsc::channel(16);
147
148        let service = self.clone();
149        let mut ctx = StreamContext::new();
150
151        info!(stream = %ctx.id(), "EDS stream started");
152
153        tokio::spawn(async move {
154            let mut node_hash: Option<NodeHash> = None;
155
156            while let Some(result) = tokio_stream::StreamExt::next(&mut stream).await {
157                match result {
158                    Ok(request) => {
159                        // Validate type URL
160                        if !request.type_url.is_empty() && request.type_url != TypeUrl::ENDPOINT {
161                            error!(
162                                stream = %ctx.id(),
163                                expected = TypeUrl::ENDPOINT,
164                                got = %request.type_url,
165                                "invalid type URL for EDS"
166                            );
167                            continue;
168                        }
169
170                        // Extract node info
171                        if node_hash.is_none() {
172                            if let Some(ref node) = request.node {
173                                let hash = NodeHash::from_id(&node.id);
174                                ctx.set_node(node.id.clone(), hash);
175                                node_hash = Some(hash);
176                            }
177                        }
178
179                        let hash = match node_hash {
180                            Some(h) => h,
181                            None => {
182                                // First request must include node information
183                                error!(
184                                    stream = %ctx.id(),
185                                    "first request missing required node information"
186                                );
187                                let _ = tx.send(Err(Status::invalid_argument(
188                                    "first request must include node information"
189                                ))).await;
190                                break;
191                            }
192                        };
193
194                        // Process request
195                        match service.sotw_handler.process_request(
196                            &ctx,
197                            TypeUrl::ENDPOINT.into(),
198                            &request.version_info,
199                            &request.resource_names,
200                            hash,
201                        ) {
202                            Ok(Some(response)) => {
203                                match service.convert_response(response) {
204                                    Ok(discovery_response) => {
205                                        if tx.send(Ok(discovery_response)).await.is_err() {
206                                            break;
207                                        }
208                                    }
209                                    Err(e) => {
210                                        error!(stream = %ctx.id(), error = %e, "failed to convert response");
211                                        let _ = tx.send(Err(e)).await;
212                                        break;
213                                    }
214                                }
215                            }
216                            Ok(None) => {}
217                            Err(e) => {
218                                error!(stream = %ctx.id(), error = %e, "EDS request failed");
219                                break;
220                            }
221                        }
222                    }
223                    Err(e) => {
224                        error!(stream = %ctx.id(), error = %e, "stream error");
225                        break;
226                    }
227                }
228            }
229
230            info!(stream = %ctx.id(), "EDS stream ended");
231        });
232
233        Ok(Response::new(ReceiverStream::new(rx)))
234    }
235
236    type DeltaEndpointsStream = EdsDeltaResponseStream;
237
238    #[instrument(skip(self, request), name = "eds_delta_stream")]
239    async fn delta_endpoints(
240        &self,
241        request: Request<Streaming<DeltaDiscoveryRequest>>,
242    ) -> Result<Response<Self::DeltaEndpointsStream>, Status> {
243        let mut stream = request.into_inner();
244        let (tx, rx) = mpsc::channel(16);
245
246        let service = self.clone();
247        let mut ctx = StreamContext::new();
248        info!(stream = %ctx.id(), "Delta EDS stream started");
249
250        tokio::spawn(async move {
251            let mut node_hash: Option<NodeHash> = None;
252            let mut client_state = ClientResourceState::new();
253
254            while let Some(result) = tokio_stream::StreamExt::next(&mut stream).await {
255                match result {
256                    Ok(request) => {
257                        if !request.type_url.is_empty() && request.type_url != TypeUrl::ENDPOINT {
258                            error!(
259                                stream = %ctx.id(),
260                                expected = TypeUrl::ENDPOINT,
261                                got = %request.type_url,
262                                "invalid type URL for delta EDS"
263                            );
264                            continue;
265                        }
266
267                        if node_hash.is_none() {
268                            if let Some(ref node) = request.node {
269                                let hash = NodeHash::from_id(&node.id);
270                                ctx.set_node(node.id.clone(), hash);
271                                node_hash = Some(hash);
272                            }
273                        }
274
275                        let hash = match node_hash {
276                            Some(h) => h,
277                            None => {
278                                error!(
279                                    stream = %ctx.id(),
280                                    "first delta request missing required node information"
281                                );
282                                let _ = tx
283                                    .send(Err(Status::invalid_argument(
284                                        "first request must include node information",
285                                    )))
286                                    .await;
287                                break;
288                            }
289                        };
290
291                        if !request.response_nonce.is_empty() {
292                            if let Some(ref err) = request.error_detail {
293                                service.delta_handler.handle_nack(
294                                    &ctx,
295                                    TypeUrl::ENDPOINT.into(),
296                                    &request.response_nonce,
297                                    &err.message,
298                                );
299                            } else {
300                                service.delta_handler.handle_ack(
301                                    &ctx,
302                                    TypeUrl::ENDPOINT.into(),
303                                    &request.response_nonce,
304                                );
305                            }
306                        }
307
308                        match service.delta_handler.process_request(
309                            &ctx,
310                            TypeUrl::ENDPOINT.into(),
311                            &mut client_state,
312                            request.resource_names_subscribe,
313                            request.resource_names_unsubscribe,
314                            hash,
315                        ) {
316                            Ok(Some(response)) => match delta_response_to_proto(response) {
317                                Ok(proto_response) => {
318                                    if tx.send(Ok(proto_response)).await.is_err() {
319                                        break;
320                                    }
321                                }
322                                Err(e) => {
323                                    error!(stream = %ctx.id(), error = %e, "failed to encode delta response");
324                                    let _ = tx.send(Err(e)).await;
325                                    break;
326                                }
327                            },
328                            Ok(None) => {}
329                            Err(e) => {
330                                error!(stream = %ctx.id(), error = %e, "delta EDS request failed");
331                                break;
332                            }
333                        }
334                    }
335                    Err(e) => {
336                        error!(stream = %ctx.id(), error = %e, "delta stream error");
337                        break;
338                    }
339                }
340            }
341
342            info!(stream = %ctx.id(), "Delta EDS stream ended");
343            drop(tx);
344        });
345
346        Ok(Response::new(ReceiverStream::new(rx)))
347    }
348
349    #[instrument(skip(self, request), name = "eds_fetch")]
350    async fn fetch_endpoints(
351        &self,
352        request: Request<DiscoveryRequest>,
353    ) -> Result<Response<DiscoveryResponse>, Status> {
354        let request = request.into_inner();
355        let ctx = StreamContext::new();
356
357        let node_hash = request
358            .node
359            .as_ref()
360            .map(|n| NodeHash::from_id(&n.id))
361            .ok_or_else(|| Status::invalid_argument("node is required"))?;
362
363        debug!(node = ?node_hash, "EDS fetch request");
364
365        let response = self
366            .sotw_handler
367            .process_request(
368                &ctx,
369                TypeUrl::ENDPOINT.into(),
370                &request.version_info,
371                &request.resource_names,
372                node_hash,
373            )
374            .map_err(|e| Status::internal(e.to_string()))?
375            .ok_or_else(|| Status::not_found("no endpoints available"))?;
376
377        Ok(Response::new(self.convert_response(response)?))
378    }
379}
380
381#[cfg(test)]
382mod tests {
383    use super::*;
384
385    #[test]
386    fn eds_service_creation() {
387        let cache = Arc::new(ShardedCache::new());
388        let registry = Arc::new(ResourceRegistry::new());
389        let _service = EdsService::new(cache, registry);
390    }
391
392    #[test]
393    fn eds_type_url() {
394        assert_eq!(EdsService::type_url(), TypeUrl::ENDPOINT);
395    }
396}
xds_server/services/eds.rs

xds_server/services/
eds.rs
