astral/kuber
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement namespace management and improve build process

Browse Source
This commit is contained in:
dmgnr
2026-08-09 13:06:57 +07:00 Unverified
parent 8f331d4cc7
commit 58c26b735a
8 changed files with 202 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/apply.ts
+37
View File
@@ -23,6 +23,7 @@ const ManagedResources = [
] as const;
const ManagedBySelector = `app.kubernetes.io/managed-by=${LABELS["app.kubernetes.io/managed-by"]}`;
const ManagedByLabel = LABELS["app.kubernetes.io/managed-by"];
function getResourceOrder(resource: KubernetesObject): number {
return ResourceOrder[resource.kind as keyof typeof ResourceOrder] ?? 999;
@@ -38,6 +39,42 @@ export function getResourceKey(resource: KubernetesObject): string {
return `${resource.kind}:${resource.metadata?.namespace ?? ""}:${resource.metadata?.name ?? ""}`;
}
export async function getNamespaceManagementStatus(
namespace: string,
): Promise<"missing" | "managed" | "external"> {
try {
const existing = await objectApi.read({
apiVersion: "v1",
kind: "Namespace",
metadata: { name: namespace },
});
return existing.metadata?.labels?.["app.kubernetes.io/managed-by"] === ManagedByLabel
? "managed"
: "external";
} catch (error) {
if (
error &&
typeof error === "object" &&
"code" in error &&
error.code === 404
) {
return "missing";
}
throw error;
}
}
export async function assertManagedNamespace(namespace: string): Promise<void> {
const status = await getNamespaceManagementStatus(namespace);
if (status === "external") {
throw new Error(
`Namespace ${namespace} is not managed by kuber. Refusing to mutate it.`,
);
}
}
export async function applyResource<T extends KubernetesObject>(
resource: T,
): Promise<T> {
lib/build.ts
+113 -33
View File
@@ -1,6 +1,7 @@
import { basename, isAbsolute, relative, resolve } from "node:path";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import type { Writable } from "node:stream";
import type { ComposeSpecification, Service } from "../schema/docker.d";
import { IMAGE_REGISTRY } from "../const";
@@ -17,6 +18,22 @@ type BuildPlan = {
};
type ProgressReporter = (message: string) => void | Promise<void>;
type BuildReporter = {
progress?: ProgressReporter;
stream?: Writable;
};
type SpawnResult = {
exitCode: number;
stdout: string;
stderr: string;
};
function toReadableStream(
stream: number | ReadableStream<Uint8Array> | undefined,
): ReadableStream<Uint8Array> | undefined {
return typeof stream === "number" ? undefined : stream;
}
function shellQuote(value: string): string {
return `'${value.replaceAll("'", `'"'"'`)}'`;
@@ -83,28 +100,87 @@ function resolveBuildPlan(
};
}
async function runWithOutput(
command: ReturnType<typeof Bun.$>,
report?: ProgressReporter,
async function pumpStream(
stream: ReadableStream<Uint8Array> | null | undefined,
onLine: (line: string) => void | Promise<void>,
) {
if (!report) {
await command;
return;
if (!stream) return;
const reader = stream.getReader();
const decoder = new TextDecoder();
let buffer = "";
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
let newline = buffer.indexOf("\n");
while (newline !== -1) {
const line = buffer.slice(0, newline).replace(/\r$/, "");
buffer = buffer.slice(newline + 1);
if (line) await onLine(line);
newline = buffer.indexOf("\n");
}
}
buffer += decoder.decode();
const line = buffer.replace(/\r$/, "");
if (line) await onLine(line);
} finally {
reader.releaseLock();
}
}
async function runWithOutput(
command: Bun.Subprocess,
reporter?: BuildReporter,
) {
const stdoutLines: string[] = [];
const stderrLines: string[] = [];
await Promise.all([
pumpStream(toReadableStream(command.stdout), async (line) => {
stdoutLines.push(line);
if (reporter?.stream) reporter.stream.write(`${line}\n`);
else if (reporter?.progress) await reporter.progress(line);
}),
pumpStream(toReadableStream(command.stderr), async (line) => {
stderrLines.push(line);
if (reporter?.stream) reporter.stream.write(`${line}\n`);
else if (reporter?.progress) await reporter.progress(line);
}),
]);
const exitCode = await command.exited;
if (exitCode !== 0) {
throw new Error(
stderrLines.join("\n") || stdoutLines.join("\n") || `Command failed with exit code ${exitCode}`,
);
}
for await (const line of command.lines()) {
const text = line.trim();
if (text) await report(text);
}
return {
exitCode,
stdout: stdoutLines.join("\n"),
stderr: stderrLines.join("\n"),
} satisfies SpawnResult;
}
function ssh(script: string) {
const remoteCommand = `bash -lc ${shellQuote(script)}`;
return Bun.$`sh -lc ${`ssh ${shellQuote(REMOTE_BUILDER)} ${shellQuote(remoteCommand)} 2>&1`}`;
return Bun.spawn(["ssh", REMOTE_BUILDER, remoteCommand], {
stdout: "pipe",
stderr: "pipe",
});
}
function scp(localPath: string, remotePath: string) {
return Bun.$`sh -lc ${`scp ${shellQuote(localPath)} ${shellQuote(`${REMOTE_BUILDER}:${remotePath}`)} 2>&1`}`;
return Bun.spawn(["scp", localPath, `${REMOTE_BUILDER}:${remotePath}`], {
stdout: "pipe",
stderr: "pipe",
});
}
async function getRepoRoot(cwd: string): Promise<string> {
@@ -120,20 +196,24 @@ async function getTrackedDiff(repoRoot: string): Promise<string> {
}
async function getRemoteHead(remoteRepo: string): Promise<string | undefined> {
const result = await ssh(
`if [ -d ${shellQuote(`${remoteRepo}/.git`)} ]; then git -C ${shellQuote(remoteRepo)} rev-parse HEAD; fi`,
).nothrow();
const result = Bun.spawn(
[
"ssh",
REMOTE_BUILDER,
`bash -lc ${shellQuote(`if [ -d ${shellQuote(`${remoteRepo}/.git`)} ]; then git -C ${shellQuote(remoteRepo)} rev-parse HEAD; fi`)}`,
],
{ stdout: "pipe", stderr: "pipe" },
);
if (result.exitCode !== 0) return;
const text = await result.text();
const head = text.trim();
const output = await runWithOutput(result);
const head = output.stdout.trim();
return head || undefined;
}
async function syncRemoteRepo(
repoRoot: string,
remoteRepo: string,
report?: ProgressReporter,
reporter?: BuildReporter,
) {
const headSha = await getHeadSha(repoRoot);
const remoteHead = await getRemoteHead(remoteRepo);
@@ -141,13 +221,13 @@ async function syncRemoteRepo(
try {
if (remoteHead !== headSha) {
await report?.("Transferring latest git bundle");
await reporter?.progress?.("Transferring latest git bundle");
const bundlePath = `${tempDir}/repo.bundle`;
const remoteBundle = `${REMOTE_BUILD_ROOT}/${basename(repoRoot)}.bundle`;
await Bun.$.cwd(repoRoot)`git bundle create ${bundlePath} HEAD`;
await runWithOutput(ssh(`mkdir -p ${shellQuote(REMOTE_BUILD_ROOT)}`), report);
await runWithOutput(scp(bundlePath, remoteBundle), report);
await runWithOutput(ssh(`mkdir -p ${shellQuote(REMOTE_BUILD_ROOT)}`), reporter);
await runWithOutput(scp(bundlePath, remoteBundle), reporter);
await runWithOutput(ssh(
[
"set -euo pipefail",
@@ -157,7 +237,7 @@ async function syncRemoteRepo(
`git -C ${shellQuote(remoteRepo)} reset --hard FETCH_HEAD`,
`git -C ${shellQuote(remoteRepo)} clean -fd`,
].join("; "),
), report);
), reporter);
} else {
await runWithOutput(ssh(
[
@@ -165,23 +245,23 @@ async function syncRemoteRepo(
`git -C ${shellQuote(remoteRepo)} reset --hard HEAD`,
`git -C ${shellQuote(remoteRepo)} clean -fd`,
].join("; "),
), report);
), reporter);
}
const diff = await getTrackedDiff(repoRoot);
if (diff.trim().length === 0) return;
await report?.("Applying local git diff on remote builder");
await reporter?.progress?.("Applying local git diff on remote builder");
const diffPath = `${tempDir}/repo.diff`;
const remoteDiff = `${REMOTE_BUILD_ROOT}/${basename(repoRoot)}.diff`;
await Bun.write(diffPath, diff);
await runWithOutput(scp(diffPath, remoteDiff), report);
await runWithOutput(scp(diffPath, remoteDiff), reporter);
await runWithOutput(ssh(
[
"set -euo pipefail",
`git -C ${shellQuote(remoteRepo)} apply --allow-binary-replacement "$PWD/${remoteDiff}"`,
].join("; "),
), report);
), reporter);
} finally {
await rm(tempDir, { recursive: true, force: true });
}
@@ -200,8 +280,8 @@ function getBuildPlans(
});
}
async function buildRemote(plan: BuildPlan, report?: ProgressReporter) {
await report?.(`Building ${plan.name}`);
async function buildRemote(plan: BuildPlan, reporter?: BuildReporter) {
await reporter?.progress?.(`Building ${plan.name}`);
const args = [
"docker",
"build",
@@ -216,14 +296,14 @@ async function buildRemote(plan: BuildPlan, report?: ProgressReporter) {
];
const command = args.map(shellQuote).join(" ");
await runWithOutput(ssh(`set -euo pipefail; ${command}`), report);
await runWithOutput(ssh(`set -euo pipefail; ${command}`), reporter);
}
export async function buildServices(
project: string,
compose: ComposeSpecification,
cwd = process.cwd(),
report?: ProgressReporter,
reporter?: BuildReporter,
): Promise<number> {
const repoRoot = await getRepoRoot(cwd);
const remoteRepo = `${REMOTE_BUILD_ROOT}/${basename(repoRoot)}`;
@@ -231,10 +311,10 @@ export async function buildServices(
if (plans.length === 0) return 0;
await syncRemoteRepo(repoRoot, remoteRepo, report);
await syncRemoteRepo(repoRoot, remoteRepo, reporter);
for (const plan of plans) {
await buildRemote(plan, report);
await buildRemote(plan, reporter);
}
return plans.length;
lib/shared.ts
+2 -3
View File
@@ -2,12 +2,10 @@ import { basename } from "node:path";
import type { V1Deployment, V1Pod } from "@kubernetes/client-node";
import { LABELS } from "../const";
import { apps, core, objectApi } from "./k8s";
import { feature } from "bun:bundle";
const ManagedBySelector = `app.kubernetes.io/managed-by=${LABELS["app.kubernetes.io/managed-by"]}`;
export function getProject() {
if (!feature("prod")) return "kube-system";
const cwd = process.cwd();
return basename(cwd);
}
@@ -51,7 +49,8 @@ export async function getPodForDeployment(name: string): Promise<V1Pod> {
export function getPodContainerName(pod: V1Pod): string {
const container = pod.spec?.containers[0]?.name;
if (!container) throw new Error(`No container found in pod ${pod.metadata?.name}`);
if (!container)
throw new Error(`No container found in pod ${pod.metadata?.name}`);
return container;
}