Something went wrong. Try again.
A Forgejo Runner plugin that spawn jobs as Kubernetes pods
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121mod config;mod error;mod execution;mod plugin { #![allow(clippy::all, clippy::pedantic, clippy::nursery, clippy::restriction)] tonic::include_proto!("plugin.v1alpha");}mod server;
use std::future::Future;
use kube::Client;use tokio::net::TcpListener;use tokio_stream::wrappers::TcpListenerStream;use tonic::transport::{Error, Server};use tonic_health::server::health_reporter;use tracing::info;
pub use config::Config;use plugin::backend_plugin_server::BackendPluginServer;use server::PluginServer;
/// Runs Kuranto on an already-bound listener until shutdown is requested.////// The caller owns configuration, Kubernetes authentication, and logging setup./// Shutdown stops accepting requests and waits for active RPCs to complete.////// # Errors/// Returns an error if the gRPC transport fails.pub async fn serve<Shutdown>( client: Client, config: Config, listener: TcpListener, shutdown: Shutdown,) -> Result<(), Error>where Shutdown: Future<Output = ()> + Send + 'static,{ let plugin = PluginServer::new(client, config.clone()); let (health_reporter, health_service) = health_reporter(); health_reporter .set_serving::<BackendPluginServer<PluginServer>>() .await;
info!( address = ?listener.local_addr(), namespace = %config.namespace, default_job_image = %config.default_job_image, "starting Kuranto" );
Server::builder() .add_service(health_service) .add_service(BackendPluginServer::new(plugin)) .serve_with_incoming_shutdown(TcpListenerStream::new(listener), shutdown) .await}
#[cfg(test)]mod tests { use std::time::Duration;
use kube::{Client, Config as KubeConfig}; use tokio::{net::TcpListener, sync::oneshot, time::timeout}; use tonic::transport::Endpoint; use tonic_health::pb::{ HealthCheckRequest, health_check_response::ServingStatus, health_client::HealthClient, };
use super::{Config, serve};
#[tokio::test] async fn serves_health_and_shuts_down_without_kubernetes_requests() { let listener = TcpListener::bind("127.0.0.1:0") .await .expect("bind listener"); let address = listener.local_addr().expect("listener address"); let client = Client::try_from(KubeConfig::new( "http://127.0.0.1:1".parse().expect("Kubernetes URI"), )) .expect("Kubernetes client"); let config = Config { namespace: "test".to_owned(), listen_addr: address, default_job_image: "alpine:latest".to_owned(), job_node_selector: None, job_affinity: None, job_tolerations: None, job_resource_requirements: None, pod_resource_requirements: None, service_resource_requirements: None, job_service_account: None, max_environment_timeout_seconds: 60, }; let (shutdown, receiver) = oneshot::channel(); let task = tokio::spawn(serve(client, config, listener, async { let _ = receiver.await; })); let channel = Endpoint::from_shared(format!("http://{address}")) .expect("health endpoint") .connect() .await .expect("health connection"); let mut health = HealthClient::new(channel); let response = health .check(HealthCheckRequest { service: "plugin.v1alpha.BackendPlugin".to_owned(), }) .await .expect("health response"); assert_eq!(response.into_inner().status, ServingStatus::Serving as i32); drop(health); shutdown.send(()).expect("send shutdown"); timeout(Duration::from_secs(5), task) .await .expect("shutdown deadline") .expect("server task") .expect("server result"); }}