xianyun_cluster/driver/
vcluster.rs

1//! Port of `internal/cluster/vcluster/driver.go` and
2//! `internal/cluster/vcluster/errors.go`: a [`super::Driver`] implementation
3//! that runs a vcluster (virtual cluster) inside a host cluster's namespace.
4//!
5//! **Executor injection.** As with `driver::external` (see that module's
6//! doc comment for the full rationale), this driver takes its
7//! `Arc<dyn CommandRunner + Send + Sync>` once, at construction, rather
8//! than pulling it off an ambient context on every call.
9
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12
13use super::{DomainSource, Driver as DriverTrait, DriverError, State, Status};
14use crate::clusterdomain::{COREDNS_COREFILE, KUBE_DNS_DOMAIN_ARG, RESOLV_CONF_SEARCH};
15use crate::kubeconfig;
16use xianyun_core::config::types::ClusterConfig;
17use xianyun_executor::exec::{CommandRunner, Options};
18
19pub const DRIVER_NAME: &str = "vcluster";
20
21/// Marks the host namespace caproni provisions into. The example
22/// multi-tenant Kyverno policy keys its ownership rules off this label.
23const MANAGED_LABEL: &str = "caproni.gitlab.io/managed";
24
25/// Port of `vcluster/errors.go`.
26#[derive(Debug, thiserror::Error, PartialEq, Eq)]
27pub enum VClusterError {
28    #[error("vcluster LoadBalancer has no address")]
29    LoadBalancerPending,
30    #[error("namespace RBAC not granted")]
31    NamespaceRbacPending,
32}
33
34/// Port of `vcluster.Driver`.
35pub struct Driver {
36    pub cfg: ClusterConfig,
37    pub profile_name: String,
38    /// Port of `Driver.pollInterval`, overridable in tests to avoid real sleeps.
39    pub poll_interval: Duration,
40    /// Port of `Driver.pollTimeout`.
41    pub poll_timeout: Duration,
42    runner: Arc<dyn CommandRunner + Send + Sync>,
43}
44
45impl Driver {
46    /// Port of `vcluster.NewDriver`. Takes the injected `CommandRunner`
47    /// explicitly; see the module doc comment for why.
48    #[must_use]
49    pub fn new(
50        cfg: ClusterConfig,
51        profile_name: String,
52        runner: Arc<dyn CommandRunner + Send + Sync>,
53    ) -> Self {
54        Self {
55            cfg,
56            profile_name,
57            poll_interval: Duration::from_secs(5),
58            poll_timeout: Duration::from_secs(5 * 60),
59            runner,
60        }
61    }
62
63    /// Port of `(*vcluster.Driver).ensureBinaries`.
64    fn ensure_binaries(&self) -> Result<(), DriverError> {
65        for bin in ["vcluster", "kubectl"] {
66            self.runner
67                .look_path(bin)
68                .map_err(|_| DriverError::DriverNotFound(bin.to_string()))?;
69        }
70
71        Ok(())
72    }
73
74    /// Port of `(*vcluster.Driver).hostKubeconfig`.
75    fn host_kubeconfig(&self) -> Result<kubeconfig::Config, DriverError> {
76        let mut args = Vec::new();
77        let host_ctx = host_context(self);
78        if !host_ctx.is_empty() {
79            args.push("--context".to_string());
80            args.push(host_ctx);
81        }
82        args.extend(
83            ["config", "view", "--raw", "--minify", "--flatten"]
84                .iter()
85                .map(std::string::ToString::to_string),
86        );
87
88        let out = self
89            .runner
90            .read_stdout("kubectl", &args)
91            .map_err(|e| DriverError::Other(format!("extracting host kubeconfig: {e}")))?;
92
93        kubeconfig::Config::create(&out)
94            .map_err(|e| DriverError::Other(format!("writing host kubeconfig: {e}")))
95    }
96
97    /// Port of `(*vcluster.Driver).runWithEnv`.
98    fn run_with_env(
99        &self,
100        env: &[String],
101        name: &str,
102        args: &[String],
103    ) -> Result<String, DriverError> {
104        let opts = Options {
105            env: Some(env.to_vec()),
106            log_level: Some(tracing::Level::DEBUG),
107            stderr_log_level: Some(tracing::Level::DEBUG),
108            ..Default::default()
109        };
110
111        let result = self.runner.run(name, args, Some(opts));
112        if let Some(err) = result.error {
113            return Err(DriverError::Other(format!(
114                "running {name} {}: {err}",
115                args.join(" ")
116            )));
117        }
118
119        Ok(result.stdout.join("\n"))
120    }
121
122    /// Port of `(*vcluster.Driver).vcluster`.
123    fn vcluster(&self, env: &[String], args: &[&str]) -> Result<String, DriverError> {
124        self.run_with_env(env, "vcluster", &owned(args))
125    }
126
127    /// Port of `(*vcluster.Driver).kubectlOut`.
128    fn kubectl_out(&self, env: &[String], args: &[&str]) -> Result<String, DriverError> {
129        self.run_with_env(env, "kubectl", &owned(args))
130    }
131
132    /// Port of `(*vcluster.Driver).find`.
133    fn find(&self, env: &[String]) -> Result<Option<VClusterListEntry>, DriverError> {
134        let ns = namespace(self);
135        let out = self.vcluster(env, &["list", "-n", &ns, "--output", "json"])?;
136
137        let entries: Vec<VClusterListEntry> =
138            serde_json::from_str::<Option<Vec<VClusterListEntry>>>(&out)
139                .map_err(|e| DriverError::Other(format!("parsing vcluster list JSON: {e}")))?
140                .unwrap_or_default();
141
142        Ok(entries
143            .into_iter()
144            .find(|e| e.name == self.profile_name && e.namespace == ns))
145    }
146
147    /// Port of `(*vcluster.Driver).create`.
148    fn create(&self, env: &[String]) -> Result<(), DriverError> {
149        if manage_namespace(self) {
150            self.ensure_namespace(env)?;
151            self.wait_for_namespace_rbac(env)?;
152        }
153
154        let ns = namespace(self);
155        let mut args: Vec<String> = vec![
156            "create".into(),
157            self.profile_name.clone(),
158            "-n".into(),
159            ns,
160            "--driver".into(),
161            "helm".into(),
162            "--expose".into(),
163            "--connect=false".into(),
164        ];
165
166        if let Some(vc) = &self.cfg.vcluster {
167            if !vc.chart_version.is_empty() {
168                args.push("--chart-version".into());
169                args.push(vc.chart_version.clone());
170            }
171            for f in &vc.values {
172                args.push("-f".into());
173                args.push(f.clone());
174            }
175        }
176
177        if let Some(net) = &self.cfg.network
178            && !net.cluster_domain.is_empty()
179        {
180            args.push("--set".into());
181            args.push(format!(
182                "networking.advanced.clusterDomain={}",
183                net.cluster_domain
184            ));
185        }
186
187        self.run_with_env(env, "vcluster", &args)?;
188
189        Ok(())
190    }
191
192    /// Port of `(*vcluster.Driver).ensureNamespace`.
193    fn ensure_namespace(&self, env: &[String]) -> Result<(), DriverError> {
194        let manifest = namespace_manifest(&namespace(self));
195
196        let opts = Options {
197            env: Some(env.to_vec()),
198            stdin: Some(Box::new(std::io::Cursor::new(manifest.into_bytes()))),
199            log_level: Some(tracing::Level::DEBUG),
200            stderr_log_level: Some(tracing::Level::DEBUG),
201            ..Default::default()
202        };
203
204        let result = self.runner.run(
205            "kubectl",
206            &["apply".to_string(), "-f".to_string(), "-".to_string()],
207            Some(opts),
208        );
209
210        if let Some(err) = result.error {
211            return Err(DriverError::Other(format!(
212                "creating host namespace {}: {err}",
213                namespace(self)
214            )));
215        }
216
217        Ok(())
218    }
219
220    /// Port of `(*vcluster.Driver).waitForNamespaceRBAC`.
221    fn wait_for_namespace_rbac(&self, env: &[String]) -> Result<(), DriverError> {
222        let ns = namespace(self);
223        let deadline = Instant::now() + self.poll_timeout;
224
225        loop {
226            let opts = Options {
227                env: Some(env.to_vec()),
228                log_level: Some(tracing::Level::DEBUG),
229                stderr_log_level: Some(tracing::Level::DEBUG),
230                failure_log_level: Some(tracing::Level::DEBUG),
231                ..Default::default()
232            };
233
234            let result = self.runner.run(
235                "kubectl",
236                &owned(&["auth", "can-i", "list", "secrets", "-n", &ns]),
237                Some(opts),
238            );
239
240            if result.error.is_none() && result.exit_code == 0 {
241                return Ok(());
242            }
243
244            if Instant::now() >= deadline {
245                return Err(DriverError::Other(format!(
246                    "{}: {ns} not usable within {:?}",
247                    VClusterError::NamespaceRbacPending,
248                    self.poll_timeout
249                )));
250            }
251
252            std::thread::sleep(self.poll_interval);
253        }
254    }
255
256    /// Port of `(*vcluster.Driver).resume`.
257    fn resume(&self, env: &[String]) -> Result<(), DriverError> {
258        let ns = namespace(self);
259        self.vcluster(
260            env,
261            &["resume", &self.profile_name, "-n", &ns, "--driver", "helm"],
262        )?;
263        Ok(())
264    }
265
266    /// Port of `(*vcluster.Driver).loadBalancerIP`.
267    fn load_balancer_ip(&self, env: &[String]) -> Result<String, DriverError> {
268        let ns = namespace(self);
269        let out = self.kubectl_out(
270            env,
271            &[
272                "get",
273                "svc",
274                &self.profile_name,
275                "-n",
276                &ns,
277                "-o",
278                "jsonpath={.status.loadBalancer.ingress[0].ip}",
279            ],
280        )?;
281
282        Ok(out.trim().to_string())
283    }
284}
285
286fn owned(args: &[&str]) -> Vec<String> {
287    args.iter().map(std::string::ToString::to_string).collect()
288}
289
290impl DriverTrait for Driver {
291    fn name(&self) -> &str {
292        DRIVER_NAME
293    }
294
295    fn start(&self) -> Result<(), DriverError> {
296        if self.cfg.registry_mirror.is_some() {
297            tracing::warn!(
298                "registry_mirror is currently unsupported for the vcluster driver and will be ignored"
299            );
300        }
301
302        self.ensure_binaries()?;
303
304        let kc = self.host_kubeconfig()?;
305        let env = env_with_kubeconfig(&kc);
306
307        let entry = self.find(&env)?;
308        match entry {
309            None => self.create(&env)?,
310            Some(ref e) if e.status == "Paused" => self.resume(&env)?,
311            Some(_) => {
312                tracing::info!(
313                    name = self.profile_name,
314                    namespace = namespace(self),
315                    "vcluster already running"
316                );
317            }
318        }
319
320        let _ = kc.close();
321
322        wait_for_load_balancer_ip_with_env(self, &env)
323            .map_err(|e| DriverError::Other(e.to_string()))
324    }
325
326    fn stop(&self) -> Result<(), DriverError> {
327        self.ensure_binaries()?;
328
329        let kc = self.host_kubeconfig()?;
330        let env = env_with_kubeconfig(&kc);
331
332        let ns = namespace(self);
333        self.vcluster(
334            &env,
335            &["pause", &self.profile_name, "-n", &ns, "--driver", "helm"],
336        )?;
337
338        let _ = kc.close();
339
340        Ok(())
341    }
342
343    fn destroy(&self) -> Result<(), DriverError> {
344        self.ensure_binaries()?;
345
346        let kc = self.host_kubeconfig()?;
347        let env = env_with_kubeconfig(&kc);
348
349        let entry = self.find(&env)?;
350        if entry.is_none() {
351            tracing::info!(
352                name = self.profile_name,
353                namespace = namespace(self),
354                "vcluster already absent"
355            );
356            let _ = kc.close();
357            return Ok(());
358        }
359
360        let ns = namespace(self);
361        self.vcluster(
362            &env,
363            &[
364                "delete",
365                &self.profile_name,
366                "-n",
367                &ns,
368                "--driver",
369                "helm",
370                "--delete-namespace",
371            ],
372        )?;
373
374        let _ = kc.close();
375
376        Ok(())
377    }
378
379    fn status(&self) -> Result<Status, DriverError> {
380        self.ensure_binaries()?;
381
382        let mut status = Status {
383            driver: DRIVER_NAME.to_string(),
384            state: Some(State::Stopped),
385            kube_context: kube_context_name(self),
386            ..Default::default()
387        };
388
389        let kc = self.host_kubeconfig()?;
390        let env = env_with_kubeconfig(&kc);
391
392        let entry = self.find(&env)?;
393        let _ = kc.close();
394
395        let Some(entry) = entry else {
396            return Ok(status);
397        };
398
399        status.exists = true;
400        status.kubernetes_version = entry.version;
401        status.state = Some(map_state(&entry.status));
402
403        Ok(status)
404    }
405
406    fn get_kubeconfig(&self) -> Result<String, DriverError> {
407        self.ensure_binaries()?;
408
409        let kc = self.host_kubeconfig()?;
410        let env = env_with_kubeconfig(&kc);
411
412        let ip = self.load_balancer_ip(&env)?;
413
414        if ip.is_empty() {
415            let _ = kc.close();
416            return Err(DriverError::Other(format!(
417                "{}: {}/{}",
418                VClusterError::LoadBalancerPending,
419                namespace(self),
420                self.profile_name
421            )));
422        }
423
424        let ns = namespace(self);
425        let result = self.vcluster(
426            &env,
427            &[
428                "connect",
429                &self.profile_name,
430                "-n",
431                &ns,
432                "--print",
433                "--background-proxy=false",
434                "--server",
435                &format!("https://{ip}:443"),
436            ],
437        );
438
439        let _ = kc.close();
440
441        result
442    }
443
444    fn domain_sources(&self) -> Option<Vec<DomainSource>> {
445        // vcluster forwards nodes/proxy to the host, so kubelet configz
446        // would report the host kubelet's domain (#260) — deliberately
447        // excluded from this driver's chain.
448        Some(vec![
449            RESOLV_CONF_SEARCH,
450            COREDNS_COREFILE,
451            KUBE_DNS_DOMAIN_ARG,
452        ])
453    }
454}
455
456/// Appends a `kubeconfig::Config`'s `Environ()` on top of the current
457/// process environment, mirroring `append(os.Environ(), kc.Environ()...)`.
458fn env_with_kubeconfig(kc: &kubeconfig::Config) -> Vec<String> {
459    let mut env: Vec<String> = std::env::vars().map(|(k, v)| format!("{k}={v}")).collect();
460    env.extend(kc.environ());
461    env
462}
463
464/// Port of `(*vcluster.Driver).namespace`.
465pub fn namespace(driver: &Driver) -> String {
466    if let Some(vc) = &driver.cfg.vcluster
467        && !vc.namespace.is_empty()
468    {
469        return vc.namespace.clone();
470    }
471
472    format!("vcluster-{}", driver.profile_name)
473}
474
475/// Port of `(*vcluster.Driver).hostContext`.
476pub fn host_context(driver: &Driver) -> String {
477    driver
478        .cfg
479        .vcluster
480        .as_ref()
481        .map(|vc| vc.host_context.clone())
482        .unwrap_or_default()
483}
484
485/// Port of `vcluster.vclusterListEntry`, the JSON shape of `vcluster list -o json`.
486#[derive(Debug, Clone, Default, serde::Deserialize)]
487pub struct VClusterListEntry {
488    #[serde(default, rename = "Name")]
489    pub name: String,
490    #[serde(default, rename = "Namespace")]
491    pub namespace: String,
492    #[serde(default, rename = "Version")]
493    pub version: String,
494    #[serde(default, rename = "Status")]
495    pub status: String,
496}
497
498/// Port of `vcluster.mapState`.
499pub fn map_state(status: &str) -> State {
500    match status {
501        "Running" => State::Ready,
502        "Paused" => State::Stopped,
503        _ => State::Starting,
504    }
505}
506
507/// Port of `(*vcluster.Driver).manageNamespace`.
508pub fn manage_namespace(driver: &Driver) -> bool {
509    driver.cfg.vcluster.is_none()
510        || driver
511            .cfg
512            .vcluster
513            .as_ref()
514            .is_some_and(|vc| vc.manage_namespace.is_none_or(|v| v))
515}
516
517/// Port of `(*vcluster.Driver).kubeContextName`.
518pub fn kube_context_name(driver: &Driver) -> String {
519    format!(
520        "vcluster_{}_{}_{}",
521        driver.profile_name,
522        namespace(driver),
523        host_context(driver)
524    )
525}
526
527/// Port of `(*vcluster.Driver).waitForLoadBalancerIP`, called with an empty
528/// environment (mirroring the Go test suite calling it directly with a
529/// `nil` env).
530pub fn wait_for_load_balancer_ip(driver: &Driver) -> Result<(), VClusterError> {
531    wait_for_load_balancer_ip_with_env(driver, &[])
532}
533
534fn wait_for_load_balancer_ip_with_env(
535    driver: &Driver,
536    env: &[String],
537) -> Result<(), VClusterError> {
538    let deadline = Instant::now() + driver.poll_timeout;
539
540    loop {
541        if let Ok(ip) = driver.load_balancer_ip(env)
542            && !ip.is_empty()
543        {
544            tracing::info!(
545                name = driver.profile_name,
546                ip,
547                "vcluster API server exposed"
548            );
549            return Ok(());
550        }
551
552        if Instant::now() >= deadline {
553            return Err(VClusterError::LoadBalancerPending);
554        }
555
556        std::thread::sleep(driver.poll_interval);
557    }
558}
559
560/// Port of the namespace `Namespace` manifest applied before `vcluster
561/// create` when `manageNamespace()` is true. Extracted as a standalone,
562/// directly-testable function since the shared `MockCommandRunner` does not
563/// expose the stdin content passed to `run`, unlike the Go test suite's
564/// `RunAndReturn` capture (see `PORT_NOTES.md`).
565#[must_use]
566pub fn namespace_manifest(namespace: &str) -> String {
567    format!(
568        "apiVersion: v1\nkind: Namespace\nmetadata:\n  name: {namespace}\n  labels:\n    {MANAGED_LABEL}: \"true\"\n"
569    )
570}
xianyun_cluster/driver/vcluster.rs

xianyun_cluster/driver/
vcluster.rs
