Facility to bypass cli start to enable concurrency with other actors

[casibbald]

I would like to use Loco in my Kubernetes controller to present a UI and aspects that are easier to consume visually rather than kubectl.

I have a concept working where Loco runs and the Kubecontroller runs, however Loco is blocking and the Kubecontroller only starts up when Loco receives its Shutdown signal.

I would like to be able to bypass the cli start that is traditionally setup as follows:

use loco_rs::cli;
use yapp::{
    app::App,
    controllers::{kubecontroller::run, telemetry},
    core::kubecontroller::State,
};

#[tokio::main]
async fn main() -> loco_rs::Result<()> {
    cli::main::<App>().await.expect("TODO: panic message");

    // Initialize Kubernetes controller state
    let state = State::default();
    run(state.clone()).await; // Ensure `run` is awaited
    Ok(())
}

and I was thinking of something like the following which I have compiling but still have issues getting it to work.:

use std::str::FromStr;
use loco_rs::{app::Hooks, boot::StartMode, environment::Environment, prelude::*};
use std::time::Duration;

use tokio::signal;
use yapp::{
    app::App,
    controllers::kubecontroller::run,
    core::{environment::EnvironmentExt, kubecontroller::State},
};

#[warn(dead_code)]
async fn run_kubecontroller() {
    let state = State::default();
    run(state.clone()).await;
}

struct ServeParams {
    binding: String,
    port: i32,
}

#[allow(clippy::redundant_pub_crate)]
#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
    // 1. Environment setup ---------------------------------------------------
    let environment = std::env::var("ENVIRONMENT")
        .map_or(Environment::Development, |env| <Environment as FromStr>::from_str(&env).unwrap_or(Environment::Development));
    println!("🚀 Starting in {environment} environment");

    // 2. Config loading ------------------------------------------------------
    let config = App::load_config(&environment)
        .await
        .map_err(|e| {
            eprintln!("Failed to load config: {e:?}");
            e
        })?;

    // 3. Logger initialization -----------------------------------------------
    if !App::init_logger(&config, &environment).expect("Failed to initialize logger") {
        // Default Loco logger if not overridden
        tracing_subscriber::fmt()
            .with_target(false)
            .with_max_level(environment.log_level())
            .init();
    }

    // 4. Application boot ----------------------------------------------------
    let boot = App::boot(StartMode::ServerAndWorker, &environment)
        .await
        .expect("Failed to boot application");

    // 5. Server parameters ---------------------------------------------------
    let serve_params = ServeParams {
        binding: config.server.host.clone(),
        port: config.server.port,
    };

    // 6. Concurrent tasks setup ----------------------------------------------
    let (_tx, rx) = tokio::sync::oneshot::channel();

    // Server task
    let server_task = tokio::spawn({
        let ctx = boot.app_context.clone();
        let (tx, _rx) = tokio::sync::oneshot::channel();
        let router = boot.router.clone();
        async move {
            tracing::info!(
                "Starting server on {}:{}",
                serve_params.binding,
                serve_params.port
            );
            App::serve(router.expect("REASON"), &ctx)
                .await
                .expect("Server failed");

            tx.send(()).expect("Failed to send shutdown signal");
        }
    });

    // Background task
    let background_task = tokio::spawn(async move {
        tracing::info!("Background task started");
        let mut rx: tokio::sync::oneshot::Receiver<()> = rx;
        tokio::spawn(run_kubecontroller());
        loop {
            tokio::select! {
                _ = &mut rx => {
                    tracing::info!("Background task received shutdown signal");
                    break;
                }
                () = tokio::time::sleep(Duration::from_secs(5)) => {
                    tracing::info!("Background task heartbeat");
                }
            }
        }
    });

    // 7. Graceful shutdown ---------------------------------------------------
    tokio::select! {
        _ = signal::ctrl_c() => {
            tracing::info!("Received CTRL+C, shutting down");
        }
        _ = server_task => {
            tracing::info!("Server task completed");
        }
    }

    // 8. Cleanup -------------------------------------------------------------
    background_task.abort();
    App::on_shutdown(&boot.app_context).await;

    Ok(())
}

Log of currently observed blocking behaviour from the first code snippet:

      Building image
  Beginning garbage collecting Kubernetes objects
  Deleting kubernetes objects:
  → Service/yapp-controller
  → Deployment/yapp-controller
  **2025-02-01T19:15:41.919127Z  INFO loco_rs::app: shutting down...** <<< ------ Loco shuts down, releasing the controller to start working. However They should run concurrently, and this is what I am looking to address.
  2025-02-01T19:15:41.919638Z  INFO yapp::core::telemetry: Global default subscriber already set!
  2025-02-01T19:15:41.919936Z DEBUG tower::buffer::worker: service.ready=true processing request
  2025-02-01T19:15:41.920014Z DEBUG HTTP: kube_client::client::builder: requesting http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.920122Z DEBUG HTTP: hyper_util::client::legacy::connect::http: connecting to 10.96.0.1:443 http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.920332Z DEBUG HTTP: hyper_util::client::legacy::connect::http: connected to 10.96.0.1:443 http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.920368Z DEBUG HTTP: rustls::client::hs: No cached session for IpAddress(V4(Ipv4Addr([10, 96, 0, 1])))     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.920437Z DEBUG HTTP: rustls::client::hs: Not resuming any session     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.921841Z DEBUG HTTP: rustls::client::hs: Using ciphersuite TLS13_AES_128_GCM_SHA256     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.921854Z DEBUG HTTP: rustls::client::tls13: Not resuming     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.921932Z DEBUG HTTP: rustls::client::tls13: TLS1.3 encrypted extensions: []     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.921942Z DEBUG HTTP: rustls::client::hs: ALPN protocol is None     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.921953Z DEBUG HTTP: rustls::client::tls13: Got CertificateRequest CertificateRequestPayloadTls13 { context: , extensions: [Unknown(UnknownExtension { typ: StatusRequest, payload:  }), Unknown(UnknownExtension { typ: SCT, payload:  }), SignatureAlgorithms([RSA_PSS_SHA256, ECDSA_NISTP256_SHA256, ED25519, RSA_PSS_SHA384, RSA_PSS_SHA512, RSA_PKCS1_SHA256, RSA_PKCS1_SHA384, RSA_PKCS1_SHA512, ECDSA_NISTP384_SHA384, ECDSA_NISTP521_SHA512, RSA_PKCS1_SHA1, ECDSA_SHA1_Legacy]), AuthorityNames([DistinguishedName(3015311330110603550403130a6b756265726e65746573), DistinguishedName(3019311730150603550403130e66726f6e742d70726f78792d6361)])] }     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.921965Z DEBUG HTTP: rustls::client::common: Client auth requested but no cert/sigscheme available     http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.925026Z DEBUG HTTP: hyper_util::client::legacy::pool: pooling idle connection for ("https", 10.96.0.1) http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=1 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.925219Z  INFO kube_runtime::controller: press ctrl+c to shut down gracefully
  2025-02-01T19:15:41.925238Z DEBUG kube_runtime::controller: applier runner held until store is ready
  2025-02-01T19:15:41.925380Z DEBUG tower::buffer::worker: service.ready=true processing request
  2025-02-01T19:15:41.925438Z DEBUG HTTP: kube_client::client::builder: requesting http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=500 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.925454Z DEBUG HTTP: hyper_util::client::legacy::pool: reuse idle connection for ("https", 10.96.0.1) http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&limit=500 otel.name="list" otel.kind="client"
  2025-02-01T19:15:41.927285Z DEBUG hyper_util::client::legacy::pool: pooling idle connection for ("https", 10.96.0.1)
  2025-02-01T19:15:41.927671Z DEBUG kube_runtime::controller: store is ready, starting runner
  2025-02-01T19:15:41.927743Z DEBUG tower::buffer::worker: service.ready=true processing request
  2025-02-01T19:15:41.927821Z DEBUG HTTP: kube_client::client::builder: requesting http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&watch=true&timeoutSeconds=290&allowWatchBookmarks=true&resourceVersion=1143 otel.name="watch" otel.kind="client"
  2025-02-01T19:15:41.927861Z DEBUG HTTP: hyper_util::client::legacy::pool: reuse idle connection for ("https", 10.96.0.1) http.method=GET http.url=https://10.96.0.1/apis/kube.rs/v1/documents?&watch=true&timeoutSeconds=290&allowWatchBookmarks=true&resourceVersion=1143 otel.name="watch" otel.kind="client"
  2025-02-01T19:15:41.933259Z  INFO reconciling object:reconcile: yapp::core::kubecontroller: Reconciling Document document_name=samuel namespace=default object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel"
  2025-02-01T19:15:41.933342Z DEBUG reconciling object:reconcile: tower::buffer::worker: service.ready=true processing request object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel"
  2025-02-01T19:15:41.933377Z DEBUG reconciling object:reconcile:HTTP: kube_client::client::builder: requesting object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.933452Z DEBUG reconciling object:reconcile:HTTP: hyper_util::client::legacy::connect::http: connecting to 10.96.0.1:443 object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.934859Z  INFO reconciling object:reconcile: yapp::core::kubecontroller: Reconciling Document document_name=lorem namespace=default object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem"
  2025-02-01T19:15:41.935294Z DEBUG reconciling object:reconcile: tower::buffer::worker: service.ready=true processing request object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem"
  2025-02-01T19:15:41.935313Z DEBUG reconciling object:reconcile:HTTP: kube_client::client::builder: requesting object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.935379Z DEBUG reconciling object:reconcile:HTTP: hyper_util::client::legacy::connect::http: connecting to 10.96.0.1:443 object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937269Z  INFO reconciling object:reconcile: yapp::core::kubecontroller: Reconciling Document document_name=illegal namespace=default object.ref=Document.v1.kube.rs/illegal.default object.reason=object updated document="illegal"
  2025-02-01T19:15:41.937313Z DEBUG reconciling object:reconcile:HTTP: hyper_util::client::legacy::connect::http: connected to 10.96.0.1:443 object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937374Z DEBUG reconciling object:reconcile:HTTP: rustls::client::hs: Resuming session     object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937486Z DEBUG reconciling object:reconcile:HTTP: hyper_util::client::legacy::connect::http: connected to 10.96.0.1:443 object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937490Z DEBUG reconciling object:reconcile:HTTP: rustls::client::hs: No cached session for IpAddress(V4(Ipv4Addr([10, 96, 0, 1])))     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937503Z  WARN reconciling object: yapp::core::kubecontroller: reconcile failed: Any(ApplyFailed(Any(Custom { kind: Other, error: "IllegalDocument" }))) object.ref=Document.v1.kube.rs/illegal.default object.reason=object updated
  2025-02-01T19:15:41.937531Z DEBUG reconciling object:reconcile:HTTP: rustls::client::hs: Not resuming any session     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937832Z DEBUG reconciling object:reconcile:HTTP: rustls::client::hs: Using ciphersuite TLS13_AES_128_GCM_SHA256     object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937848Z DEBUG reconciling object:reconcile:HTTP: rustls::client::tls13: Resuming using PSK     object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937894Z DEBUG reconciling object:reconcile:HTTP: rustls::client::tls13: TLS1.3 encrypted extensions: []     object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.937897Z DEBUG reconciling object:reconcile:HTTP: rustls::client::hs: ALPN protocol is None     object.ref=Document.v1.kube.rs/lorem.default object.reason=object updated document="lorem" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/lorem/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.939045Z DEBUG reconciling object:reconcile:HTTP: rustls::client::hs: Using ciphersuite TLS13_AES_128_GCM_SHA256     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.939061Z DEBUG reconciling object:reconcile:HTTP: rustls::client::tls13: Not resuming     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.939105Z DEBUG reconciling object:reconcile:HTTP: rustls::client::tls13: TLS1.3 encrypted extensions: []     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.939108Z DEBUG reconciling object:reconcile:HTTP: rustls::client::hs: ALPN protocol is None     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.939128Z DEBUG reconciling object:reconcile:HTTP: rustls::client::tls13: Got CertificateRequest CertificateRequestPayloadTls13 { context: , extensions: [Unknown(UnknownExtension { typ: StatusRequest, payload:  }), Unknown(UnknownExtension { typ: SCT, payload:  }), SignatureAlgorithms([RSA_PSS_SHA256, ECDSA_NISTP256_SHA256, ED25519, RSA_PSS_SHA384, RSA_PSS_SHA512, RSA_PKCS1_SHA256, RSA_PKCS1_SHA384, RSA_PKCS1_SHA512, ECDSA_NISTP384_SHA384, ECDSA_NISTP521_SHA512, RSA_PKCS1_SHA1, ECDSA_SHA1_Legacy]), AuthorityNames([DistinguishedName(3015311330110603550403130a6b756265726e65746573), DistinguishedName(3019311730150603550403130e66726f6e742d70726f78792d6361)])] }     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.939140Z DEBUG reconciling object:reconcile:HTTP: rustls::client::common: Client auth requested but no cert/sigscheme available     object.ref=Document.v1.kube.rs/samuel.default object.reason=object updated document="samuel" http.method=PATCH http.url=https://10.96.0.1/apis/kube.rs/v1/namespaces/default/documents/samuel/status?&force=true&fieldManager=cntrlr otel.name="patch_status" otel.kind="client"
  2025-02-01T19:15:41.942417Z DEBUG hyper_util::client::legacy::pool: pooling idle connection for ("https", 10.96.0.1)
  2025-02-01T19:15:41.943350Z DEBUG hyper_util::client::legacy::pool: pooling idle connection for ("https", 10.96.0.1)

[kaplanelad]

You can turn off the cli feature and boot the server or other component by yourself

[casibbald]

Is there an example where someone does this? I have attempted it myself, and it has been a terrible mess!

@kaplanelad I tried to do as per the following boot server directly

[kaplanelad]

you can follow the cli.rs code.