Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions core/crates/ployz-build/src/execution.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ pub enum Stage {
Building,
Output,
Cleanup,
/// Pushing the built image out of the host, after the Build itself (`ployz build`).
Push,
}

/// Bounded output and work observations from the execution host.
Expand Down
5 changes: 4 additions & 1 deletion core/crates/ployz-sdk/index.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,10 @@ export type PreparationEvent =
| { Selected: { machine: import("./generated/payloads").Machine; reason: BuilderReason; rejections: string[] } }
| { Build: { Stage: string } | { Output: number[] } | { Step: BuildStep } | { StepOutput: { step: string; stderr: boolean; text: string } } | { Timing: unknown } | { Target: { name: string; outcome: unknown } } }
| "Transfer"
| { Delivered: { image: string; machine_id: MachineId } };
/** One Service's image starts going to these Machines, by name. */
| { Sending: { service: string; machines: string[] } }
/** One Machine received a Service's image. */
| { Delivered: { image: string; service: string; machine_id: MachineId } };

export type RunningPreparation = AsyncIterable<PreparationEvent> & {
abort(): void;
Expand Down
33 changes: 23 additions & 10 deletions core/crates/ployz/src/deploy/pipeline.rs
Original file line number Diff line number Diff line change
Expand Up @@ -248,12 +248,6 @@ fn invalid_argument(message: String) -> RpcError {
}
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PushOutcome {
pub pushed: Vec<PushedImage>,
pub failures: Vec<String>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct PushedImage {
pub image: String,
Expand All @@ -274,8 +268,8 @@ pub(crate) async fn push_project_images(
machines: &[MachineObservation],
preview: &DeployPlan,
cancellation: &CancellationToken,
) -> Result<PushOutcome, crate::sdk::prepare::PreparationError> {
let mut pushed = Vec::new();
progress: &impl Fn(crate::sdk::prepare::Progress),
) -> Result<Vec<String>, crate::sdk::prepare::PreparationError> {
let mut failures = Vec::new();
// Check every actual destination before any image or application changes.
let deliveries = builds.iter().map(|service| {
Expand All @@ -297,15 +291,34 @@ pub(crate) async fn push_project_images(
if targets.is_empty() {
continue;
}
// The Build Machine already holds its image; only the others receive it.
let receivers = machines
.iter()
.filter(|machine| machine.machine.id != service.machine_id)
.filter(|machine| targets.contains(&machine.machine.id.to_string()))
.map(|machine| machine.machine.name.clone())
.collect::<Vec<_>>();
if !receivers.is_empty() {
progress(crate::sdk::prepare::Progress::Sending {
service: service.name.clone(),
machines: receivers,
});
}
match push_image(client, service, machines, &targets, cancellation).await {
Ok((images, service_failures)) => {
pushed.extend(images);
for image in images {
progress(crate::sdk::prepare::Progress::Delivered {
image: image.image,
service: service.name.clone(),
machine_id: image.machine_id,
});
}
failures.extend(service_failures);
}
Err(error) => failures.push(format!("{}: {error}", service.image)),
}
}
Ok(PushOutcome { pushed, failures })
Ok(failures)
}

pub(crate) async fn plan_project(
Expand Down
102 changes: 78 additions & 24 deletions core/crates/ployz/src/handlers/build.rs
Original file line number Diff line number Diff line change
Expand Up @@ -82,26 +82,25 @@ pub(super) fn build(root: &ArgMatches) -> Result<(), Error> {
}
.into());
}
let events = matches
.get_one::<String>("events")
.map(std::fs::File::create)
.transpose()?
.map(std::sync::Mutex::new);
let events = std::sync::Arc::new(Events(
matches
.get_one::<String>("events")
.map(std::fs::File::create)
.transpose()?
.map(std::sync::Mutex::new),
));
runtime()?.block_on(async {
let cancellation = crate::cancellation::on_ctrl_c();
let build = captured.build;
let export = cancellation.clone();
let building = events.clone();
let (build, built) = tokio::task::spawn_blocking(move || {
let built = build.execute_local(&cancellation, &|event| {
if let Some(file) = &events {
write_event(file, event);
}
});
let built = build.execute_local(&cancellation, &|event| building.write(event));
(build, built)
})
.await
.map_err(std::io::Error::other)?;
push(&grant, &built.map_err(BuildCommandError::from)?).await?;
push(&grant, &built.map_err(BuildCommandError::from)?, &events).await?;
// The pushed image and its printed result are final; uploading cache only speeds up
// the next Build, so its failure is a warning and it never reaches the events file.
let exported = tokio::task::spawn_blocking(move || build.export_local_cache(&export))
Expand All @@ -116,16 +115,28 @@ pub(super) fn build(root: &ArgMatches) -> Result<(), Error> {
})
}

/// One `{"at": <unix ms>, "event": <SDK preparation event>}` line, the shape Cloud's build
/// log reads. Progress is best effort: a failed write never fails the build.
fn write_event(file: &std::sync::Mutex<std::fs::File>, event: &ployz_build::Progress) {
use std::io::Write as _;
let at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |elapsed| elapsed.as_millis());
let line = json!({"at": at, "event": {"Build": event}});
if let Ok(mut file) = file.lock() {
let _ = writeln!(file, "{line}");
/// The `--events` file: one `{"at": <unix ms>, "event": {"Build": <progress>}}` line per
/// event, the SDK preparation event shape Cloud's build log reads. Progress is best effort:
/// a failed write never fails the build.
struct Events(Option<std::sync::Mutex<std::fs::File>>);

impl Events {
fn write(&self, event: &ployz_build::Progress) {
use std::io::Write as _;
let Some(file) = &self.0 else { return };
let at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |elapsed| elapsed.as_millis());
if let Ok(mut file) = file.lock() {
let _ = writeln!(file, "{}", json!({"at": at, "event": {"Build": event}}));
}
}

/// One line of the push: output of the open `Push` stage.
fn push_line(&self, line: &str) {
self.write(&ployz_build::Progress::Output(
format!("{line}\n").into_bytes(),
));
}
}

Expand Down Expand Up @@ -179,24 +190,27 @@ fn git(source: &Path, args: &[&str]) -> Result<String, Error> {
}

/// Push through the grant with `docker push`, which sends only the layers the Machine
/// lacks, and print what the Machine verified it received.
async fn push(grant: &BuildGrant, built: &LocalImage) -> Result<(), Error> {
/// lacks, and print what the Machine verified it received. Each layer's progress line
/// goes to the events file as the push's output.
async fn push(grant: &BuildGrant, built: &LocalImage, events: &Events) -> Result<(), Error> {
let digest = &built.image.reference;
let hex = digest
.strip_prefix("sha256:")
.ok_or(BuildCommandError::NotDigestIdentified)?;
let tag = format!("{}:{RETAINED_DIGEST_TAG_PREFIX}{hex}", built.repository);
events.write(&ployz_build::Progress::Stage(ployz_build::Stage::Push));
let registry = open_grant_registry(grant, &ManagementRelay::default()).await?;
let local = format!("{}/{tag}", registry.address());
docker(&["tag", digest, &local]).await?;
let pushed = docker(&["push", "--quiet", &local]).await;
let pushed = docker_push(&local, events).await;
// The loopback name is meaningless once this command exits.
let _ = docker(&["image", "rm", &local]).await;
if let Err(error) = pushed {
return Err(registry.refusal().map_or(error, |refusal| {
BuildCommandError::GrantRefused(refusal).into()
}));
}
events.push_line(&format!("Pushed {digest}"));
// The Machine stores a tagged manifest only when its bytes hash to the tag's
// digest, so a completed push is the Machine's confirmation of `digest`.
println!(
Expand All @@ -206,6 +220,46 @@ async fn push(grant: &BuildGrant, built: &LocalImage) -> Result<(), Error> {
Ok(())
}

/// `docker push`, its per-layer lines streamed to the events file as they come.
async fn docker_push(reference: &str, events: &Events) -> Result<(), Error> {
use tokio::io::AsyncBufReadExt as _;
let mut child = tokio::process::Command::new("docker")
.args(["push", reference])
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()?;
let stderr = child.stderr.take().map(|stderr| {
tokio::spawn(async move {
let mut text = String::new();
let _ = tokio::io::AsyncReadExt::read_to_string(
&mut tokio::io::BufReader::new(stderr),
&mut text,
)
.await;
text
})
});
if let Some(stdout) = child.stdout.take() {
let mut lines = tokio::io::BufReader::new(stdout).lines();
while let Ok(Some(line)) = lines.next_line().await {
events.push_line(&line);
}
}
let status = child.wait().await?;
let stderr = match stderr {
Some(task) => task.await.unwrap_or_default(),
None => String::new(),
};
if status.success() {
return Ok(());
}
Err(BuildCommandError::Docker {
command: "push".to_owned(),
stderr: stderr.trim().to_owned(),
}
.into())
}

async fn docker(args: &[&str]) -> Result<(), Error> {
let output = tokio::process::Command::new("docker")
.args(args)
Expand Down
20 changes: 11 additions & 9 deletions core/crates/ployz/src/sdk/prepare.rs
Original file line number Diff line number Diff line change
Expand Up @@ -220,8 +220,15 @@ pub enum Progress {
Selected(Box<SelectedBuilder>),
Build(ployz_build::Progress),
Transfer,
/// One Service's image starts going to these Machines, by name.
Sending {
service: ployz_core::ServiceName,
machines: Vec<ployz_core::MachineName>,
},
/// One Machine received a Service's image.
Delivered {
image: String,
service: ployz_core::ServiceName,
machine_id: ployz_core::MachineId,
},
}
Expand Down Expand Up @@ -272,27 +279,22 @@ pub async fn prepare(
if !builds.is_empty() {
progress(Progress::Transfer);
}
let outcome = crate::deploy::pipeline::push_project_images(
let failures = crate::deploy::pipeline::push_project_images(
client,
&builds,
&machines,
&plan,
cancellation,
&progress,
)
.await?;
for image in outcome.pushed {
progress(Progress::Delivered {
image: image.image,
machine_id: image.machine_id,
});
}
if cancellation.is_cancelled() {
return Err(PreparationError::Cancelled);
}
if !outcome.failures.is_empty() {
if !failures.is_empty() {
return Err(PreparationError::Delivery(format!(
"image push failed: {}",
outcome.failures.join("; ")
failures.join("; ")
)));
}
Ok(Prepared { plan, builds })
Expand Down
1 change: 1 addition & 0 deletions core/crates/ployz/src/sdk/running.rs
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ impl OutputBudget {
| Progress::Platforms(_)
| Progress::Selected(_)
| Progress::Transfer
| Progress::Sending { .. }
| Progress::Delivered { .. } => 0,
};
buffered.fetch_add(size, Ordering::Relaxed);
Expand Down
2 changes: 1 addition & 1 deletion dashboard/CONTEXT.md
Original file line number Diff line number Diff line change
Expand Up @@ -248,7 +248,7 @@ How a Service's image is described for building: a Dockerfile or Railpack.
_Avoid_: Builder, builder type

**Build Step**:
One unit of an Image Build as the Engine reports it: a BuildKit step (a Dockerfile instruction, image resolution, or context transfer) or a Ployz-owned phase such as source upload or image delivery. A Build Step is keyed stably within its attempt, changes state until it completes, and owns the output attributed to it. Build Steps are retained with the attempt, separately from lifecycle history.
One unit of an Image Build: a BuildKit step (a Dockerfile instruction, image resolution, or context transfer) or a Ployz-owned phase such as source upload, waiting for a runner, installing ployz, pushing, or sending the image to a Machine. Each is recorded where it happens, so its timing is honest: Cloud times the wait for a runner, the runner its install and push, the Engine the rest. A Build Step is keyed stably within one Builder's go at its Image Build, changes state until it completes, and owns the output attributed to it. Each go is a section of the Image Build's one log; a later one opens with why the build moved there. Build Steps are retained with the attempt, separately from lifecycle history.
_Avoid_: Build log line, vertex, build stage (a Cloud Deployment Stage is not a Build Step)

Git repository identity and access are separate. Cloud can read a public GitHub repository anonymously or use an Organization member's connected GitHub App installation. Public access never falls back to installation credentials. Both paths pin a commit per Cloud Deployment Attempt and materialize it through the same source acquisition module. Automatic Git deployment and CI gating require installation access; public sources deploy manually.
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
ALTER TABLE "environment_deployment_build_step" ADD COLUMN "attempt" integer DEFAULT 0 NOT NULL;--> statement-breakpoint
ALTER TABLE "environment_deployment_build_step" DROP CONSTRAINT "environment_deployment_build_step_key_unique";--> statement-breakpoint
ALTER TABLE "environment_deployment_build_step" ADD CONSTRAINT "environment_deployment_build_step_key_unique" UNIQUE NULLS NOT DISTINCT("deployment_id","image","attempt","build","key");
Loading
Loading