9301161
use anyhow::Context;
use reqwest::StatusCode;
use uuid::Uuid;
use ci_protocol::runner::{
ArtifactCompleteRequest, ArtifactInitRequest, ArtifactInitResponse, ClaimRequest,
ClaimResponse, HeartbeatRequest, HeartbeatResponse, JobFinishRequest, LogChunksRequest,
RegisterRequest, RegisterResponse, StepReport,
};
use config::runner::RunnerConfig;
#[derive(Clone)]
pub struct RunnerClient {
http: reqwest::Client,
base_url: String,
credential: Option<String>,
}
impl RunnerClient {
pub fn new(cfg: &RunnerConfig) -> anyhow::Result<Self> {
let http = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(120))
.build()
.context("building http client")?;
Ok(Self {
http,
base_url: cfg.server_url.trim_end_matches('/').to_string(),
credential: None,
})
}
#[must_use]
pub fn with_credential(mut self, credential: String) -> Self {
self.credential = Some(credential);
self
}
fn url(&self, path: &str) -> String {
format!("{}/api/v1/runner{path}", self.base_url)
}
async fn post<B: serde::Serialize, T: serde::de::DeserializeOwned>(
&self,
path: &str,
body: &B,
) -> anyhow::Result<T> {
let url = self.url(path);
let mut request = self.http.post(&url).json(body);
if let Some(credential) = &self.credential {
request = request.bearer_auth(credential);
}
let span = export::telemetry::http_client_span("POST", &url);
let response = export::telemetry::send(span, request)
.await
.context("sending runner request")?;
let status = response.status();
let text = response.text().await.unwrap_or_default();
if status != StatusCode::OK {
anyhow::bail!("runner request {path} failed: {status} {text}");
}
let envelope: serde_json::Value =
serde_json::from_str(&text).context("decoding response envelope")?;
let data = envelope
.get("data")
.cloned()
.ok_or_else(|| anyhow::anyhow!("response envelope has no data"))?;
serde_json::from_value(data).with_context(|| format!("decoding response data for {path}"))
}
pub async fn register(&self, request: &RegisterRequest) -> anyhow::Result<RegisterResponse> {
self.post("/register", request).await
}
pub async fn heartbeat(&self, request: &HeartbeatRequest) -> anyhow::Result<HeartbeatResponse> {
self.post("/heartbeat", request).await
}
pub async fn claim(&self, request: &ClaimRequest) -> anyhow::Result<ClaimResponse> {
self.post("/jobs/claim", request).await
}
pub async fn start(&self, job_id: &Uuid) -> anyhow::Result<Uuid> {
self.post(&format!("/jobs/{job_id}/start"), &serde_json::json!({}))
.await
}
pub async fn renew(&self, job_id: &Uuid) -> anyhow::Result<Uuid> {
self.post(&format!("/jobs/{job_id}/renew"), &serde_json::json!({}))
.await
}
pub async fn report_step(
&self,
job_id: Uuid,
index: i32,
report: &StepReport,
) -> anyhow::Result<Uuid> {
self.post(&format!("/jobs/{job_id}/steps/{index}"), report)
.await
}
pub async fn upload_logs(
&self,
job_id: Uuid,
request: &LogChunksRequest,
) -> anyhow::Result<Uuid> {
self.post(&format!("/jobs/{job_id}/logs"), request).await
}
pub async fn finish(&self, job_id: &Uuid, request: &JobFinishRequest) -> anyhow::Result<Uuid> {
self.post(&format!("/jobs/{job_id}/finish"), request).await
}
pub async fn event(
&self,
job_id: Uuid,
event: &ci_protocol::actions::EngineEvent,
) -> anyhow::Result<Uuid> {
self.post(&format!("/jobs/{job_id}/events"), event).await
}
pub async fn init_artifact(
&self,
job_id: Uuid,
request: &ArtifactInitRequest,
) -> anyhow::Result<ArtifactInitResponse> {
self.post(&format!("/jobs/{job_id}/artifacts/init"), request)
.await
}
pub async fn complete_artifact(
&self,
job_id: Uuid,
artifact_id: Uuid,
request: &ArtifactCompleteRequest,
) -> anyhow::Result<Uuid> {
self.post(
&format!("/jobs/{job_id}/artifacts/{artifact_id}/complete"),
request,
)
.await
}
}