1use crate::{
4 config::{invalid, validate_name, ConnectionPolicy, Limits, RegistrationPolicy},
5 providers::LocalProvider,
6 ConnectionSpec, DiscoveryOwner, Eligibility, OwnerBuilder, RegistrationSpec,
7};
8use nadisc::{
9 managed::{ManagedProvider, RegistrationPlan, Result},
10 DnsService, Instance,
11};
12use serde::{
13 de::{Error, MapAccess, Visitor},
14 Deserialize, Serialize,
15};
16use std::{
17 collections::{BTreeMap, HashMap},
18 sync::Arc,
19};
20
21#[derive(Clone, Default, Deserialize, Serialize)]
23#[serde(default, deny_unknown_fields)]
24pub struct DiscoveryConfig {
25 pub enabled: bool,
27 pub eligibility_probe: bool,
29 #[serde(skip_serializing_if = "Option::is_none")]
31 pub default_connection: Option<String>,
32 pub limits: Limits,
34 #[serde(deserialize_with = "unique_map")]
36 pub connections: BTreeMap<String, ConnectionConfig>,
37 #[serde(deserialize_with = "unique_map")]
39 pub registrations: BTreeMap<String, RegistrationConfig>,
40}
41
42#[derive(Clone, Deserialize, Serialize)]
44#[serde(default, deny_unknown_fields)]
45pub struct ConnectionConfig {
46 pub provider: String,
48 pub startup_required: bool,
50 pub request_timeout_ms: u64,
52 pub poll_interval_ms: u64,
54 pub idle_timeout_ms: u64,
56 pub snapshot: SnapshotPolicy,
58 pub retry: RetryPolicy,
60 #[serde(deserialize_with = "optional_unique_value")]
62 pub nacos: Option<serde_json::Value>,
63 #[serde(deserialize_with = "optional_unique_value")]
65 pub consul: Option<serde_json::Value>,
66 #[serde(deserialize_with = "optional_unique_value")]
68 pub kubernetes: Option<serde_json::Value>,
69 #[serde(rename = "static")]
71 pub static_config: Option<StaticConfig>,
72 pub dns: Option<DnsConfig>,
74 pub rest: Option<RestPolicy>,
76}
77
78#[derive(Clone, Deserialize, Serialize)]
80#[serde(default, deny_unknown_fields)]
81pub struct RestPolicy {
82 pub request_timeout_ms: u64,
84 pub connect_timeout_ms: u64,
86 pub lb_strategy: String,
88 pub instance_scheme: String,
90 pub retry_get_head_on_transport_error: bool,
92 pub retry_get_head_on_retryable_status: bool,
94 pub max_attempts: usize,
96}
97impl Default for RestPolicy {
98 fn default() -> Self {
102 Self {
103 request_timeout_ms: 10000,
104 connect_timeout_ms: 2000,
105 lb_strategy: "round_robin".into(),
106 instance_scheme: "http".into(),
107 retry_get_head_on_transport_error: false,
108 retry_get_head_on_retryable_status: false,
109 max_attempts: 1,
110 }
111 }
112}
113impl RestPolicy {
114 pub fn validate(&self) -> Result<()> {
118 if self.request_timeout_ms == 0
119 || self.request_timeout_ms > 86_400_000
120 || self.connect_timeout_ms == 0
121 || self.connect_timeout_ms > self.request_timeout_ms
122 || !matches!(self.lb_strategy.as_str(), "round_robin" | "weighted")
123 || !matches!(self.instance_scheme.as_str(), "http" | "https")
124 || !(1..=16).contains(&self.max_attempts)
125 {
126 return Err(invalid("invalid per-connection REST policy"));
127 }
128 Ok(())
129 }
130}
131
132#[derive(Clone, Deserialize, Serialize)]
134#[serde(default, deny_unknown_fields)]
135pub struct SnapshotPolicy {
136 pub max_age_ms: u64,
138 pub stale_if_error_ms: u64,
140}
141impl Default for SnapshotPolicy {
142 fn default() -> Self {
146 Self {
147 max_age_ms: 15000,
148 stale_if_error_ms: 0,
149 }
150 }
151}
152
153#[derive(Clone, Deserialize, Serialize)]
155#[serde(default, deny_unknown_fields)]
156pub struct RetryPolicy {
157 pub initial_backoff_ms: u64,
159 pub max_backoff_ms: u64,
161}
162impl Default for RetryPolicy {
163 fn default() -> Self {
167 Self {
168 initial_backoff_ms: 1000,
169 max_backoff_ms: 30000,
170 }
171 }
172}
173
174impl Default for ConnectionConfig {
175 fn default() -> Self {
179 Self {
180 provider: String::new(),
181 startup_required: true,
182 request_timeout_ms: 3000,
183 poll_interval_ms: 5000,
184 idle_timeout_ms: 30000,
185 snapshot: SnapshotPolicy::default(),
186 retry: RetryPolicy::default(),
187 nacos: None,
188 consul: None,
189 kubernetes: None,
190 static_config: None,
191 dns: None,
192 rest: None,
193 }
194 }
195}
196
197#[derive(Clone, Deserialize, Serialize)]
199#[serde(deny_unknown_fields)]
200pub struct StaticConfig {
201 #[serde(deserialize_with = "unique_map")]
203 pub services: BTreeMap<String, Vec<Instance>>,
204}
205#[derive(Clone, Deserialize, Serialize)]
207#[serde(deny_unknown_fields)]
208pub struct DnsConfig {
209 #[serde(deserialize_with = "unique_map")]
211 pub services: BTreeMap<String, DnsTarget>,
212}
213#[derive(Clone, Deserialize, Serialize)]
215#[serde(deny_unknown_fields)]
216pub struct DnsTarget {
217 pub host: String,
219 pub port: u16,
221}
222
223#[derive(Clone, Deserialize, Serialize)]
225#[serde(default, deny_unknown_fields)]
226pub struct RegistrationConfig {
227 #[serde(rename = "qualifier", alias = "connection_ref")]
229 pub connection_ref: String,
230 pub service_name: String,
232 pub ip: Option<String>,
234 pub port: Option<u16>,
236 pub listener_ref: Option<String>,
238 pub eligibility_ref: Option<String>,
240 pub metadata: HashMap<String, String>,
242 pub startup_required: bool,
244 pub readiness_critical: bool,
246 pub monitor_interval_ms: u64,
248 pub observation_grace_ms: u64,
250 pub deregister_timeout_ms: u64,
252 pub nacos: Option<NacosRegistration>,
254 #[serde(deserialize_with = "optional_unique_value")]
256 pub consul: Option<serde_json::Value>,
257}
258
259impl Default for RegistrationConfig {
260 fn default() -> Self {
264 Self {
265 connection_ref: "primary".into(),
266 service_name: String::new(),
267 ip: None,
268 port: None,
269 listener_ref: None,
270 eligibility_ref: None,
271 metadata: HashMap::new(),
272 startup_required: true,
273 readiness_critical: false,
274 monitor_interval_ms: 5000,
275 observation_grace_ms: 15000,
276 deregister_timeout_ms: 3000,
277 nacos: None,
278 consul: None,
279 }
280 }
281}
282
283#[derive(Clone, Deserialize, Serialize)]
285#[serde(default, deny_unknown_fields)]
286pub struct NacosRegistration {
287 pub ephemeral: bool,
289 pub cluster_name: String,
291 pub weight: f64,
293}
294impl Default for NacosRegistration {
295 fn default() -> Self {
299 Self {
300 ephemeral: true,
301 cluster_name: "DEFAULT".into(),
302 weight: 1.0,
303 }
304 }
305}
306
307#[derive(Clone)]
309pub struct ListenerEndpoint {
310 pub ip: String,
312 pub port: u16,
314}
315
316impl DiscoveryConfig {
317 pub fn validate(&self) -> Result<()> {
321 self.limits
322 .validate()
323 .map_err(|error| error.at_config_path("discovery.limits"))?;
324 if !self.enabled
325 && (!self.connections.is_empty()
326 || !self.registrations.is_empty()
327 || self.default_connection.is_some()
328 || self.eligibility_probe)
329 {
330 return Err(invalid(
331 "disabled discovery cannot declare connections, registrations or probes",
332 ));
333 }
334 if self.connections.len() > self.limits.max_connections
335 || self.registrations.len() > self.limits.max_registrations
336 {
337 return Err(invalid("discovery configuration exceeds capacity"));
338 }
339 if let Some(name) = &self.default_connection {
340 if !self.connections.contains_key(name) {
341 return Err(invalid("unknown default connection")
342 .at_config_path("discovery.default_connection"));
343 }
344 }
345 for (name, connection) in &self.connections {
346 validate_name(name).map_err(|error| error.at_config_path("discovery.connections"))?;
347 connection
348 .validate()
349 .map_err(|error| error.at_config_path(format!("discovery.connections.{name}")))?;
350 }
351 for (name, registration) in &self.registrations {
352 validate_name(name).map_err(|error| error.at_config_path("discovery.registrations"))?;
353 let connection = self
354 .connections
355 .get(®istration.connection_ref)
356 .ok_or_else(|| {
357 invalid("unknown registration connection")
358 .at_config_path(format!("discovery.registrations.{name}.qualifier"))
359 })?;
360 if !connection.has_service(®istration.service_name) {
361 return Err(invalid("unknown registration service binding")
362 .at_config_path(format!("discovery.registrations.{name}.service_name")));
363 }
364 if registration.service_name.is_empty()
365 || registration.service_name.len() > 512
366 || registration
367 .ip
368 .as_ref()
369 .is_some_and(|ip| ip.is_empty() || ip.trim() != ip)
370 || registration.port == Some(0)
371 || registration.metadata.len() > 128
372 || registration
373 .metadata
374 .iter()
375 .any(|(key, value)| key.len() > 256 || value.len() > 4096)
376 {
377 return Err(invalid(
378 "invalid registration service, endpoint or metadata",
379 ));
380 }
381 registration.policy().validate(&connection.policy())?;
382 match connection.provider.as_str() {
383 "nacos" if registration.consul.is_none() => {}
384 "consul" if registration.nacos.is_none() && registration.consul.is_some() => {}
385 _ => {
386 return Err(invalid(
387 "registration attributes do not match a writable provider",
388 ))
389 }
390 }
391 }
392 Ok(())
393 }
394
395 pub async fn build(&self) -> Result<DiscoveryOwner> {
399 self.validate()?;
400 let mut builder = OwnerBuilder::default().limits(self.limits.clone());
401 if !self.enabled {
402 return builder.build();
403 }
404 if let Some(default) = &self.default_connection {
405 builder = builder.default_connection(default);
406 }
407 for (name, config) in &self.connections {
408 let mut policy = config.policy();
409 policy.startup_required |= self.registrations.values().any(|registration| {
410 registration.connection_ref == *name && registration.startup_required
411 });
412 builder = builder.connection(ConnectionSpec {
413 name: name.clone(),
414 provider: config.build().await.map_err(|error| {
415 error.at_config_path(format!("discovery.connections.{name}"))
416 })?,
417 policy,
418 })?;
419 }
420 builder.build()
421 }
422
423 pub fn resolve(
427 &self,
428 listeners: &BTreeMap<String, ListenerEndpoint>,
429 eligibility: &BTreeMap<String, Eligibility>,
430 ) -> Result<Vec<RegistrationSpec>> {
431 if !self.enabled {
432 return Ok(Vec::new());
433 }
434 self.validate()?;
435 let mut plans = Vec::new();
436 for (name, registration) in &self.registrations {
437 let listener = match ®istration.listener_ref {
438 Some(name) => Some(
439 listeners
440 .get(name)
441 .ok_or_else(|| invalid("unknown registration listener"))?,
442 ),
443 None if listeners.len() == 1 => listeners.values().next(),
444 None => None,
445 };
446 let inferred_listener = registration
447 .listener_ref
448 .as_ref()
449 .or_else(|| (listeners.len() == 1).then(|| listeners.keys().next().unwrap()));
450 let eligibility = match ®istration.eligibility_ref {
451 Some(name) => eligibility.get(name),
452 None if inferred_listener.is_some() => eligibility.get(inferred_listener.unwrap()),
453 None if eligibility.len() == 1 => eligibility.values().next(),
454 None => None,
455 }
456 .cloned()
457 .ok_or_else(|| invalid("registration eligibility source is absent or ambiguous"))?;
458 let ip = registration
459 .ip
460 .as_ref()
461 .or_else(|| listener.map(|listener| &listener.ip))
462 .ok_or_else(|| invalid("registration IP is absent"))?;
463 let port = registration
464 .port
465 .or_else(|| listener.map(|listener| listener.port))
466 .ok_or_else(|| invalid("registration port is absent"))?;
467 let mut instance =
468 Instance::new(ip.clone(), port).with_metadata_map(registration.metadata.clone());
469 if let Some(nacos) = ®istration.nacos {
470 instance = instance
471 .with_cluster(&nacos.cluster_name)
472 .with_ephemeral(nacos.ephemeral)
473 .with_weight(nacos.weight);
474 }
475 #[allow(unused_mut)]
476 let mut plan = RegistrationPlan {
477 id: name.clone(),
478 service: registration.service_name.clone(),
479 instance,
480 attributes: BTreeMap::new(),
481 };
482 if let Some(consul) = ®istration.consul {
483 #[cfg(feature = "consul")]
484 {
485 #[derive(Deserialize)]
486 #[serde(deny_unknown_fields)]
487 struct CheckConfig {
488 check: nadisc_consul::HealthCheck,
489 }
490 let config: CheckConfig = serde_json::from_value(consul.clone())
491 .map_err(|_| invalid("invalid Consul registration configuration"))?;
492 config.check.apply(&mut plan)?;
493 }
494 #[cfg(not(feature = "consul"))]
495 {
496 let _ = consul;
497 return Err(invalid("Consul provider feature is disabled"));
498 }
499 }
500 plans.push(RegistrationSpec {
501 connection_ref: registration.connection_ref.clone(),
502 plan,
503 policy: registration.policy(),
504 eligibility,
505 });
506 }
507 Ok(plans)
508 }
509}
510
511impl RegistrationConfig {
512 pub fn policy(&self) -> RegistrationPolicy {
516 RegistrationPolicy {
517 startup_required: self.startup_required,
518 readiness_critical: self.readiness_critical,
519 monitor_interval_ms: self.monitor_interval_ms,
520 observation_grace_ms: self.observation_grace_ms,
521 deregister_timeout_ms: self.deregister_timeout_ms,
522 }
523 }
524}
525
526impl ConnectionConfig {
527 pub fn has_service(&self, service: &str) -> bool {
531 match self.provider.as_str() {
532 "static" => self
533 .static_config
534 .as_ref()
535 .is_some_and(|config| config.services.contains_key(service)),
536 "dns" => self
537 .dns
538 .as_ref()
539 .is_some_and(|config| config.services.contains_key(service)),
540 "nacos" | "consul" | "kubernetes" => {
541 let value = match self.provider.as_str() {
542 "nacos" => &self.nacos,
543 "consul" => &self.consul,
544 _ => &self.kubernetes,
545 };
546 value
547 .as_ref()
548 .and_then(|value| value.get("services"))
549 .and_then(serde_json::Value::as_object)
550 .is_some_and(|services| services.contains_key(service))
551 }
552 _ => false,
553 }
554 }
555
556 pub fn policy(&self) -> ConnectionPolicy {
560 ConnectionPolicy {
561 startup_required: self.startup_required,
562 request_timeout_ms: self.request_timeout_ms,
563 poll_interval_ms: self.poll_interval_ms,
564 max_age_ms: self.snapshot.max_age_ms,
565 stale_if_error_ms: self.snapshot.stale_if_error_ms,
566 idle_timeout_ms: self.idle_timeout_ms,
567 initial_backoff_ms: self.retry.initial_backoff_ms,
568 max_backoff_ms: self.retry.max_backoff_ms,
569 }
570 }
571
572 pub fn validate(&self) -> Result<()> {
576 self.policy().validate()?;
577 if let Some(rest) = &self.rest {
578 rest.validate()?;
579 }
580 let blocks = [
581 self.nacos.is_some(),
582 self.consul.is_some(),
583 self.kubernetes.is_some(),
584 self.static_config.is_some(),
585 self.dns.is_some(),
586 ];
587 if blocks.into_iter().filter(|present| *present).count() != 1 {
588 return Err(invalid("one provider configuration block is required"));
589 }
590 let supported = match self.provider.as_str() {
591 "static" => self.static_config.is_some(),
592 "dns" => self.dns.is_some(),
593 "nacos" => self.nacos.is_some() && cfg!(feature = "nacos"),
594 "consul" => self.consul.is_some() && cfg!(feature = "consul"),
595 "kubernetes" => self.kubernetes.is_some() && cfg!(feature = "kubernetes"),
596 _ => false,
597 };
598 if !supported {
599 return Err(
600 invalid("provider configuration or compiled feature is unavailable")
601 .at_config_path("provider"),
602 );
603 }
604 if let Some(config) = &self.static_config {
606 LocalProvider::static_services(config.services.clone())?;
607 }
608 if let Some(config) = &self.dns {
609 LocalProvider::dns_services(
610 config
611 .services
612 .iter()
613 .map(|(name, target)| {
614 (
615 name.clone(),
616 DnsService {
617 host: target.host.clone(),
618 port: target.port,
619 },
620 )
621 })
622 .collect(),
623 )?;
624 }
625 #[cfg(feature = "nacos")]
626 if let Some(value) = &self.nacos {
627 let mut config: nanacos::managed::ManagedNacosConfig =
628 serde_json::from_value(value.clone())
629 .map_err(|_| invalid("invalid Nacos provider configuration"))?;
630 config.request_timeout_ms = self.request_timeout_ms;
631 nanacos::managed::ManagedNacos::new(config)?;
632 }
633 #[cfg(feature = "consul")]
634 if let Some(value) = &self.consul {
635 let mut config: nadisc_consul::ConsulConfig = serde_json::from_value(value.clone())
636 .map_err(|_| invalid("invalid Consul provider configuration"))?;
637 config.request_timeout_ms = self.request_timeout_ms;
638 nadisc_consul::ConsulProvider::new(config)?;
639 }
640 #[cfg(feature = "kubernetes")]
641 if let Some(value) = &self.kubernetes {
642 let mut config: nadisc_kubernetes::KubernetesConfig =
643 serde_json::from_value(value.clone())
644 .map_err(|_| invalid("invalid Kubernetes provider configuration"))?;
645 config.request_timeout_ms = self.request_timeout_ms;
646 config.validate()?;
647 }
648 Ok(())
649 }
650
651 async fn build(&self) -> Result<Arc<dyn ManagedProvider>> {
655 self.validate()?;
656 match self.provider.as_str() {
657 "static" => Ok(Arc::new(LocalProvider::static_services(
658 self.static_config.as_ref().unwrap().services.clone(),
659 )?)),
660 "dns" => Ok(Arc::new(LocalProvider::dns_services(
661 self.dns
662 .as_ref()
663 .unwrap()
664 .services
665 .iter()
666 .map(|(name, target)| {
667 (
668 name.clone(),
669 DnsService {
670 host: target.host.clone(),
671 port: target.port,
672 },
673 )
674 })
675 .collect(),
676 )?)),
677 #[cfg(feature = "nacos")]
678 "nacos" => {
679 let mut config: nanacos::managed::ManagedNacosConfig =
680 serde_json::from_value(self.nacos.clone().unwrap())
681 .map_err(|_| invalid("invalid Nacos provider configuration"))?;
682 config.request_timeout_ms = self.request_timeout_ms;
683 Ok(Arc::new(nanacos::managed::ManagedNacos::new(config)?))
684 }
685 #[cfg(feature = "consul")]
686 "consul" => {
687 let mut config: nadisc_consul::ConsulConfig =
688 serde_json::from_value(self.consul.clone().unwrap())
689 .map_err(|_| invalid("invalid Consul provider configuration"))?;
690 config.request_timeout_ms = self.request_timeout_ms;
691 Ok(Arc::new(nadisc_consul::ConsulProvider::new(config)?))
692 }
693 #[cfg(feature = "kubernetes")]
694 "kubernetes" => {
695 let mut config: nadisc_kubernetes::KubernetesConfig =
696 serde_json::from_value(self.kubernetes.clone().unwrap())
697 .map_err(|_| invalid("invalid Kubernetes provider configuration"))?;
698 config.request_timeout_ms = self.request_timeout_ms;
699 Ok(Arc::new(
700 nadisc_kubernetes::KubernetesProvider::new(config).await?,
701 ))
702 }
703 _ => Err(invalid("provider is unavailable")),
704 }
705 }
706}
707
708fn optional_unique_value<'de, D: serde::Deserializer<'de>>(
712 deserializer: D,
713) -> std::result::Result<Option<serde_json::Value>, D::Error> {
714 Option::<UniqueValue>::deserialize(deserializer).map(|value| value.map(|value| value.0))
715}
716
717struct UniqueValue(serde_json::Value);
718impl<'de> Deserialize<'de> for UniqueValue {
719 fn deserialize<D: serde::Deserializer<'de>>(
723 deserializer: D,
724 ) -> std::result::Result<Self, D::Error> {
725 struct Unique;
726 impl<'de> Visitor<'de> for Unique {
727 type Value = UniqueValue;
728 fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
732 formatter.write_str("a configuration value with unique fields")
733 }
734 fn visit_map<A: MapAccess<'de>>(
738 self,
739 mut map: A,
740 ) -> std::result::Result<Self::Value, A::Error> {
741 let mut values = serde_json::Map::new();
742 while let Some((key, value)) = map.next_entry::<String, UniqueValue>()? {
743 if values.insert(key, value.0).is_some() {
744 return Err(A::Error::custom("duplicate discovery field"));
745 }
746 }
747 Ok(UniqueValue(serde_json::Value::Object(values)))
748 }
749 fn visit_seq<A: serde::de::SeqAccess<'de>>(
753 self,
754 mut seq: A,
755 ) -> std::result::Result<Self::Value, A::Error> {
756 let mut values = Vec::new();
757 while let Some(value) = seq.next_element::<UniqueValue>()? {
758 values.push(value.0);
759 }
760 Ok(UniqueValue(serde_json::Value::Array(values)))
761 }
762 fn visit_str<E: Error>(self, value: &str) -> std::result::Result<Self::Value, E> {
766 Ok(UniqueValue(value.into()))
767 }
768 fn visit_bool<E: Error>(self, value: bool) -> std::result::Result<Self::Value, E> {
772 Ok(UniqueValue(value.into()))
773 }
774 fn visit_u64<E: Error>(self, value: u64) -> std::result::Result<Self::Value, E> {
778 Ok(UniqueValue(value.into()))
779 }
780 fn visit_i64<E: Error>(self, value: i64) -> std::result::Result<Self::Value, E> {
784 Ok(UniqueValue(value.into()))
785 }
786 fn visit_f64<E: Error>(self, value: f64) -> std::result::Result<Self::Value, E> {
790 serde_json::Number::from_f64(value)
791 .map(|value| UniqueValue(value.into()))
792 .ok_or_else(|| E::custom("non-finite discovery value"))
793 }
794 fn visit_unit<E: Error>(self) -> std::result::Result<Self::Value, E> {
798 Ok(UniqueValue(serde_json::Value::Null))
799 }
800 }
801 deserializer.deserialize_any(Unique)
802 }
803}
804
805fn unique_map<'de, D, T>(deserializer: D) -> std::result::Result<BTreeMap<String, T>, D::Error>
809where
810 D: serde::Deserializer<'de>,
811 T: Deserialize<'de>,
812{
813 struct Unique<T>(std::marker::PhantomData<T>);
814 impl<'de, T: Deserialize<'de>> Visitor<'de> for Unique<T> {
815 type Value = BTreeMap<String, T>;
816 fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
820 formatter.write_str("a map with unique names")
821 }
822 fn visit_map<A: MapAccess<'de>>(
826 self,
827 mut map: A,
828 ) -> std::result::Result<Self::Value, A::Error> {
829 let mut values = BTreeMap::new();
830 while let Some((key, value)) = map.next_entry::<String, T>()? {
831 if values.insert(key, value).is_some() {
832 return Err(A::Error::custom("duplicate discovery name"));
833 }
834 }
835 Ok(values)
836 }
837 }
838 deserializer.deserialize_map(Unique(std::marker::PhantomData))
839}