scalo 2.10.12

Self-regulating runtime for Rust data-plane services. Backpressure, load shedding and adaptive scaling are on by default.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
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
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
// Project:   scalo
// File:      src/cli/app.rs
// Purpose:   ServiceApp trait and standard lifecycle runner
// Language:  Rust
//
// License:   Apache-2.0
// Copyright: (c) 2026 HYPERI PTY LIMITED

//! Application trait and lifecycle runner for data-plane services.
//!
//! Provides the standard startup sequence: parse → log → config → dispatch.
//!
//! ## Example
//!
//! ```rust,ignore
//! use scalo::cli::{CommonArgs, ServiceApp, CliError, VersionInfo, run_app};
//!
//! struct MyApp { common: CommonArgs }
//!
//! impl ServiceApp for MyApp {
//!     type Config = MyConfig;
//!
//!     fn name(&self) -> &str { "my-service" }
//!     fn env_prefix(&self) -> &str { "MY_SERVICE" }
//!     fn version_info(&self) -> VersionInfo {
//!         VersionInfo::new("my-service", env!("CARGO_PKG_VERSION"))
//!     }
//!     fn common_args(&self) -> &CommonArgs { &self.common }
//!     fn load_config(&self, path: Option<&str>) -> Result<MyConfig, CliError> { todo!() }
//!     async fn run_service(&self, config: MyConfig) -> Result<(), CliError> { todo!() }
//! }
//! ```

use std::fmt::Debug;

use serde::de::DeserializeOwned;

use super::error::CliError;
use super::version::VersionInfo;
use super::{CommonArgs, StandardCommand, output};

/// Trait for data-plane service applications.
///
/// Implement this trait to get the standard CLI lifecycle for free.
/// The 80% common behaviour (logging, config, metrics, version) is handled
/// by `run_app()`. Your app provides the 20% (config type, service logic).
pub trait ServiceApp: Sized {
    /// Application-specific configuration type.
    type Config: DeserializeOwned + Debug + Send + Sync;

    /// Service name (e.g. "dfe-loader").
    fn name(&self) -> &str;

    /// Environment variable prefix for config cascade (e.g. "DFE_LOADER").
    fn env_prefix(&self) -> &str;

    /// Version information for this service.
    fn version_info(&self) -> VersionInfo;

    /// Access the common CLI arguments.
    fn common_args(&self) -> &CommonArgs;

    /// Resolve the active subcommand.
    ///
    /// Returns `None` to default to `StandardCommand::Run`.
    fn command(&self) -> Option<&StandardCommand> {
        None
    }

    /// Load application configuration from the given path (or defaults).
    ///
    /// # Errors
    ///
    /// Returns `CliError` if configuration cannot be loaded or parsed.
    fn load_config(&self, path: Option<&str>) -> Result<Self::Config, CliError>;

    /// Run the main service loop.
    ///
    /// Called after logging, config, and [`ServiceRuntime`](super::ServiceRuntime)
    /// are initialised. The runtime contains all common infrastructure (metrics,
    /// memory guard, shutdown token, worker pool, scaling pressure). Apps just
    /// use it -- no boilerplate needed.
    ///
    /// # Errors
    ///
    /// Returns `CliError` if the service encounters a fatal error.
    fn run_service(
        &self,
        config: Self::Config,
        runtime: super::ServiceRuntime,
    ) -> impl std::future::Future<Output = Result<(), CliError>> + Send;

    /// Provide scaling pressure components for KEDA autoscaling.
    ///
    /// Override to register app-specific scaling signals (buffer depth,
    /// consumer lag, error rate, etc.). The default returns an empty vec.
    #[cfg(feature = "scaling")]
    fn scaling_components(&self, _config: &Self::Config) -> Vec<crate::ScalingComponent> {
        vec![]
    }

    /// Register all metrics for this service.
    ///
    /// Called by `metrics-manifest` and `generate-artefacts` subcommands to
    /// capture the full metric catalogue without starting the service.
    /// The default implementation is a no-op. Override to register
    /// `ServiceMetrics`, metric groups, and app-specific metrics.
    #[cfg(any(feature = "metrics", feature = "otel-metrics"))]
    fn register_metrics(&self, _manager: &crate::metrics::MetricsManager) {}

    /// Build the deployment contract for this service.
    ///
    /// Called by `generate-artefacts` to produce container specs, health
    /// endpoints, KEDA config, and metrics manifest. The default returns
    /// `None`. Override to provide a contract.
    #[cfg(feature = "deployment")]
    fn deployment_contract(&self) -> Option<crate::deployment::DeploymentContract> {
        None
    }
}

/// Drive the standard data-plane service lifecycle.
///
/// Handles subcommand dispatch:
/// - `run` (default): init logger → load config → run service
/// - `version`: print version info and exit
/// - `config-check`: load config, validate, print summary
///
/// # Errors
///
/// Returns `CliError` if any lifecycle step fails.
pub async fn run_app<A: ServiceApp>(app: A) -> Result<(), CliError> {
    let command = app.command().cloned().unwrap_or(StandardCommand::Run);
    let args = app.common_args();

    match command {
        StandardCommand::Version => {
            let info = app.version_info();
            println!("{info}");
            Ok(())
        }

        StandardCommand::ConfigCheck => {
            // Same order as `run`: the level this command REPORTS is resolved
            // partly from config, so a logger built first would report the
            // default rather than what the cascade actually yields.
            let config_path = args.config.as_deref();
            let loaded = app.load_config(config_path);
            init_logger(args)?;

            match loaded {
                Ok(config) => {
                    output::print_success("configuration is valid");
                    if !args.quiet {
                        eprintln!();
                        output::print_kv("service", &app.name());
                        output::print_kv("config", &config_path.unwrap_or("(defaults)"));
                        output::print_kv("log_level", &args.effective_log_level());
                        output::print_kv("log_format", &args.effective_log_format());
                        output::print_kv("metrics_addr", &args.metrics_addr);
                        eprintln!();
                        // Mask the Debug dump before printing: configs hold
                        // ENV-sourced secrets in plain `String` fields, so
                        // `{config:#?}` would print them in clear text.
                        // Without the `logger` feature the consumer has opted
                        // out of every sensitive-field defence anyway, so an
                        // unmasked print here is consistent.
                        let raw = format!("{config:#?}");
                        #[cfg(feature = "logger")]
                        let masked = {
                            let default_fields = crate::logger::default_sensitive_fields();
                            let patterns: Vec<&str> =
                                default_fields.iter().map(String::as_str).collect();
                            crate::logger::mask_sensitive_string(&raw, &patterns)
                        };
                        #[cfg(not(feature = "logger"))]
                        let masked = raw;
                        eprintln!("  config: {masked}");
                    }
                    Ok(())
                }
                Err(e) => {
                    output::print_error(&format!("configuration invalid: {e}"));
                    Err(e)
                }
            }
        }

        #[cfg(any(feature = "metrics", feature = "otel-metrics"))]
        StandardCommand::MetricsManifest => {
            let mgr = crate::metrics::MetricsManager::with_config(
                crate::metrics::MetricsConfig::offline(app.name()),
            );
            app.register_metrics(&mgr);
            let manifest = mgr.registry().manifest();
            println!(
                "{}",
                serde_json::to_string_pretty(&manifest)
                    .map_err(|e| CliError::Service(format!("JSON serialisation failed: {e}")))?
            );
            Ok(())
        }
        #[cfg(not(any(feature = "metrics", feature = "otel-metrics")))]
        StandardCommand::MetricsManifest => {
            output::print_error("metrics feature not enabled -- no manifest available");
            Err(CliError::Service("metrics feature not enabled".into()))
        }

        StandardCommand::GenerateArtefacts(ref artefact_args) => {
            generate_artefacts(&app, artefact_args)?;
            Ok(())
        }

        StandardCommand::ConfigSchema(ref cfg_args) => {
            #[cfg(feature = "deployment")]
            {
                emit_config_schema(&app, &cfg_args.dir)?;
                Ok(())
            }
            #[cfg(not(feature = "deployment"))]
            {
                let _ = cfg_args;
                output::print_error(
                    "deployment feature not enabled -- no config artefacts available",
                );
                Err(CliError::Service("deployment feature not enabled".into()))
            }
        }

        StandardCommand::Run => {
            let version_info = app.version_info();
            let config_path = args.config.as_deref();

            // Config loads before the logger: logger setup composes the OTLP
            // span exporter, whose settings live in the cascade. The cascade
            // has no subscriber while it loads, so its own log lines are lost.
            let loaded = app.load_config(config_path);
            init_logger_for_service(args, app.name(), &version_info.version)?;
            let config = loaded?;

            tracing::info!(
                service = app.name(),
                version = version_info.version,
                config = config_path.unwrap_or("(defaults)"),
                "starting service"
            );

            tracing::debug!(?config, "configuration loaded");

            // Build ServiceRuntime -- all common infrastructure for free
            let commit = option_env!("GIT_COMMIT").unwrap_or("unknown");
            let runtime = super::ServiceRuntime::build(
                app.name(),
                app.env_prefix(),
                &args.metrics_addr,
                &version_info.version,
                commit,
                #[cfg(feature = "scaling")]
                app.scaling_components(&config),
            )
            .await?;

            let result = app.run_service(config, runtime).await;

            // Flush what is queued before the process goes away. Both calls
            // are bounded and safe when nothing was ever wired up.
            #[cfg(feature = "otel-metrics")]
            crate::metrics::shutdown_otel_export();
            #[cfg(feature = "otel-tracing")]
            crate::otel_tracing::shutdown();

            result
        }

        #[cfg(feature = "top")]
        StandardCommand::Top(ref top_args) => {
            let top_config = crate::top::TopConfig::from_args(top_args);
            crate::top::run_top(&top_config).map_err(|e| CliError::Service(e.to_string()))
        }
    }
}

/// Initialise the logger from CLI arguments.
#[cfg(feature = "logger")]
fn init_logger(args: &CommonArgs) -> Result<(), CliError> {
    let opts = args.to_logger_options()?;
    crate::logger::setup(opts)?;
    Ok(())
}

/// Initialise the logger with service name and version injected into JSON output.
#[cfg(feature = "logger")]
fn init_logger_for_service(
    args: &CommonArgs,
    service_name: &str,
    service_version: &str,
) -> Result<(), CliError> {
    let opts = args.to_logger_options()?;
    crate::logger::setup(crate::logger::LoggerOptions {
        service_name: Some(service_name.to_string()),
        service_version: Some(service_version.to_string()),
        ..opts
    })?;
    Ok(())
}

/// Initialise the logger from CLI arguments (no-op without logger feature).
#[cfg(not(feature = "logger"))]
fn init_logger(_args: &CommonArgs) -> Result<(), CliError> {
    Ok(())
}

/// Initialise the logger with service name and version (no-op without logger feature).
#[cfg(not(feature = "logger"))]
fn init_logger_for_service(
    _args: &CommonArgs,
    _service_name: &str,
    _service_version: &str,
) -> Result<(), CliError> {
    Ok(())
}

/// Generate all CI artefacts for this service.
///
/// Produces metrics manifest, deployment contract, and container spec
/// in the output directory. Files are deterministic -- running twice produces
/// identical output (no timestamps that change between runs).
fn generate_artefacts<A: ServiceApp>(
    app: &A,
    args: &super::commands::GenerateArtefactsArgs,
) -> Result<(), CliError> {
    let output_dir = std::path::Path::new(&args.output_dir);
    std::fs::create_dir_all(output_dir)
        .map_err(|e| CliError::Service(format!("failed to create output dir: {e}")))?;

    let mut generated: Vec<String> = Vec::new();

    // Metrics manifest
    #[cfg(any(feature = "metrics", feature = "otel-metrics"))]
    {
        let mgr = crate::metrics::MetricsManager::with_config(
            crate::metrics::MetricsConfig::offline(app.name()),
        );
        app.register_metrics(&mgr);
        let manifest = mgr.registry().manifest();
        let path = output_dir.join("metrics-manifest.json");
        let json = serde_json::to_string_pretty(&manifest)
            .map_err(|e| CliError::Service(format!("metrics manifest JSON failed: {e}")))?;
        std::fs::write(&path, &json)
            .map_err(|e| CliError::Service(format!("failed to write {}: {e}", path.display())))?;
        generated.push(format!(
            "metrics-manifest.json ({} metrics)",
            manifest.metrics.len()
        ));
    }

    // Deployment contract + container manifest
    #[cfg(feature = "deployment")]
    let deployment_contract = app.deployment_contract();
    #[cfg(feature = "deployment")]
    if deployment_contract.is_none() {
        output::print_warn(&format!(
            "ServiceApp::deployment_contract() returned None for `{}` -- \
             only metrics-manifest.json will be generated. \
             Implement the trait hook to emit deployment-contract.json, \
             container-manifest.json, and Dockerfile.runtime.",
            app.name()
        ));
    }
    #[cfg(feature = "deployment")]
    if let Some(contract) = deployment_contract {
        // Full deployment contract (secrets, KEDA, Helm, everything)
        let path = output_dir.join("deployment-contract.json");
        let json = serde_json::to_string_pretty(&contract)
            .map_err(|e| CliError::Service(format!("deployment contract JSON failed: {e}")))?;
        std::fs::write(&path, &json)
            .map_err(|e| CliError::Service(format!("failed to write {}: {e}", path.display())))?;
        generated.push("deployment-contract.json".to_string());

        // Container manifest (minimal subset for CI image builds)
        let cm_path = output_dir.join("container-manifest.json");
        let cm_json = crate::deployment::generate::generate_container_manifest(&contract)
            .map_err(|e| CliError::Service(format!("container manifest failed: {e}")))?;
        std::fs::write(&cm_path, &cm_json).map_err(|e| {
            CliError::Service(format!("failed to write {}: {e}", cm_path.display()))
        })?;
        generated.push("container-manifest.json".to_string());

        // Runtime stage Dockerfile fragment (for CI composition)
        let rt_path = output_dir.join("Dockerfile.runtime");
        let rt_content = crate::deployment::generate::generate_runtime_stage(&contract);
        std::fs::write(&rt_path, &rt_content).map_err(|e| {
            CliError::Service(format!("failed to write {}: {e}", rt_path.display()))
        })?;
        generated.push("Dockerfile.runtime".to_string());

        // ArgoCD Application CR (default generation -- ArgoCD is the
        // standard CD tool across the fleet).
        let argo_path = output_dir.join("argocd-application.yaml");
        let argo_cfg = crate::deployment::ArgocdConfig {
            repo_url: crate::deployment::argocd_repo_url_from_cascade(&contract.app_name),
            ..Default::default()
        };
        let argo_content =
            crate::deployment::generate::generate_argocd_application(&contract, &argo_cfg, None);
        std::fs::write(&argo_path, &argo_content).map_err(|e| {
            CliError::Service(format!("failed to write {}: {e}", argo_path.display()))
        })?;
        generated.push("argocd-application.yaml".to_string());

        // Reflectable config artefacts (config-schema.{json,yaml} +
        // capability-catalog.{json,yaml}). Emitted only when the contract carries a
        // config_schema and/or capabilities (scalo-rs#6). Same output as the
        // standalone `config-schema` subcommand, so the drift test is stable
        // whichever produced the committed copy.
        let cfg_written = crate::deployment::emit_config_artifacts(&contract, output_dir)
            .map_err(|e| CliError::Service(format!("config artefacts failed: {e}")))?;
        for p in &cfg_written {
            let name = p
                .file_name()
                .and_then(std::ffi::OsStr::to_str)
                .unwrap_or("config-artefact");
            generated.push(name.to_string());
        }
    }

    if generated.is_empty() {
        output::print_warn("no artefacts generated (no metrics or deployment features enabled)");
    } else {
        output::print_success(&format!(
            "generated {} artefact(s) in {}",
            generated.len(),
            output_dir.display()
        ));
        for name in &generated {
            output::print_kv("  wrote", name);
        }
    }

    Ok(())
}

/// Emit just the reflectable config artefacts (`config-schema.*`,
/// `capability-catalog.*`) for the `config-schema` subcommand.
#[cfg(feature = "deployment")]
fn emit_config_schema<A: ServiceApp>(app: &A, dir: &str) -> Result<(), CliError> {
    let Some(contract) = app.deployment_contract() else {
        output::print_warn(&format!(
            "ServiceApp::deployment_contract() returned None for `{}` -- no config artefacts",
            app.name()
        ));
        return Ok(());
    };
    let written = crate::deployment::emit_config_artifacts(&contract, dir)
        .map_err(|e| CliError::Service(format!("config artefacts failed: {e}")))?;
    if written.is_empty() {
        output::print_warn(&format!(
            "contract for `{}` carries no config_schema or capabilities -- nothing emitted. \
             Populate `config_schema` + `capabilities` in the app's deployment_contract().",
            app.name()
        ));
    } else {
        output::print_success(&format!(
            "wrote {} config artefact(s) to {dir}",
            written.len()
        ));
        for p in &written {
            output::print_kv("  wrote", &p.display().to_string());
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_standard_command_default_is_run() {
        // When command() returns None, run_app defaults to Run
        let cmd = StandardCommand::Run;
        assert!(matches!(cmd, StandardCommand::Run));
    }
}