1/**
2 * Internal Kubernetes API client: transport-agnostic server-side apply and
3 * kind discovery for arbitrary (CRD) manifests. Powers
4 * `Kubernetes.Manifest`, `Kubernetes.Deployment`, `Kubernetes.Job`,
5 * `Kubernetes.HelmChart`, and the `AWS.EKS.Cluster` kubernetes-object
6 * binding channel. Not exported from the Kubernetes index.
7 *
8 * Authentication is delegated to the connection's {@link ClusterAdapter}:
9 * every request mints headers through the resolved
10 * {@link ClusterTransport}, so short-lived tokens (EKS SigV4 presigns,
11 * exec-plugin credentials) stay fresh.
12 */
13import * as Data from "effect/Data";
14import * as Effect from "effect/Effect";
15import * as Schedule from "effect/Schedule";
16import * as https from "node:https";
17import {
18 findClusterAdapter,
19 type ClusterTransport,
20} from "../ClusterAdapter.ts";
21import type { Connection } from "../Connection.ts";
22import {
23 buildKubernetesObjectPathWithSpec,
24 chunkByApplyRank,
25 DEFAULT_APPLY_RANK,
26 kubernetesObjectKey,
27 lookupKubernetesKindSpec,
28 sortRefsForDelete,
29 toKubernetesObjectRef,
30 type KubernetesObjectDefinition,
31 type KubernetesObjectKindSpec,
32 type KubernetesObjectRef,
33} from "./objects.ts";
34
35export class KubernetesApiError extends Data.TaggedError("KubernetesApiError")<{
36 method: string;
37 path: string;
38 statusCode: number;
39 body: string;
40}> {
41 override get message(): string {
42 return `${this.method} ${this.path} responded ${this.statusCode}: ${
43 this.body.length > 0 ? this.body.slice(0, 1000) : "(empty body)"
44 }`;
45 }
46}
47
48const fieldManager = "alchemy";
49
50/**
51 * Resolve the {@link ClusterTransport} for a connection through its
52 * registered adapter.
53 */
54export const connectCluster = (connection: Connection) =>
55 findClusterAdapter(connection.auth.kind).pipe(
56 Effect.flatMap((adapter) => adapter.connect(connection)),
57 );
58
59const requestJson = Effect.fn(function* ({
60 transport,
61 method,
62 path,
63 body,
64}: {
65 transport: ClusterTransport;
66 method: string;
67 path: string;
68 body?: Record<string, unknown>;
69}) {
70 const headers = yield* transport.headers;
71 const url = new URL(path, transport.endpoint);
72 const payload = body ? JSON.stringify(body) : undefined;
73
74 return yield* Effect.tryPromise({
75 try: () =>
76 new Promise<unknown>((resolve, reject) => {
77 const request = https.request(
78 {
79 protocol: url.protocol,
80 hostname: url.hostname,
81 port: url.port || 443,
82 path: `${url.pathname}${url.search}`,
83 method,
84 headers: {
85 ...headers,
86 Accept: "application/json",
87 ...(payload
88 ? {
89 "Content-Type": "application/apply-patch+yaml",
90 "Content-Length": Buffer.byteLength(payload),
91 }
92 : {}),
93 },
94 ...(transport.certificateAuthorityData
95 ? {
96 ca: Buffer.from(
97 transport.certificateAuthorityData,
98 "base64",
99 ).toString("utf8"),
100 }
101 : {}),
102 ...(transport.clientCert
103 ? {
104 cert: transport.clientCert.certificate,
105 key: transport.clientCert.key,
106 }
107 : {}),
108 ...(transport.insecureSkipTlsVerify
109 ? { rejectUnauthorized: false }
110 : {}),
111 },
112 (response) => {
113 const chunks: Buffer[] = [];
114 response.on("data", (chunk) => {
115 chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
116 });
117 response.on("end", () => {
118 const responseBody = Buffer.concat(chunks).toString("utf8");
119 const statusCode = response.statusCode ?? 500;
120
121 if (statusCode < 200 || statusCode >= 300) {
122 reject(
123 new KubernetesApiError({
124 method,
125 path,
126 statusCode,
127 body: responseBody,
128 }),
129 );
130 return;
131 }
132
133 if (!responseBody.trim()) {
134 resolve(undefined);
135 return;
136 }
137
138 try {
139 resolve(JSON.parse(responseBody));
140 } catch {
141 resolve(responseBody);
142 }
143 });
144 },
145 );
146
147 request.on("error", reject);
148 if (payload) {
149 request.write(payload);
150 }
151 request.end();
152 }),
153 catch: (error) =>
154 error instanceof KubernetesApiError
155 ? error
156 : new Error(
157 `Failed Kubernetes ${method} ${path}: ${error instanceof Error ? error.message : String(error)}`,
158 ),
159 }).pipe(
160 // Transport-level failures (ECONNREFUSED/ECONNRESET/ETIMEDOUT/DNS)
161 // are transient — a fresh managed endpoint's load balancer can refuse
162 // connections for a short window after the cluster reports ready.
163 // Every request here is idempotent (GET / SSA PATCH / DELETE), so
164 // retry them; HTTP errors (KubernetesApiError) are handled by the
165 // callers.
166 Effect.retry({
167 while: (e): boolean => !(e instanceof KubernetesApiError),
168 schedule: Schedule.max([
169 Schedule.spaced("5 seconds"),
170 Schedule.recurs(8),
171 ]),
172 }),
173 );
174});
175
176// ─────────────────────────────────────────────────────── kind discovery ──
177
178interface ApiResourceList {
179 resources?: {
180 name?: string;
181 kind?: string;
182 namespaced?: boolean;
183 }[];
184}
185
186// One resolution per (endpoint, apiVersion, kind) per process. Plain cached
187// values (no finalizers), so module scope is safe.
188const discoveredKinds = new Map<string, KubernetesObjectKindSpec>();
189
190/**
191 * Resolve the REST mapping (plural + scope) for an arbitrary kind: static
192 * table fast path, then the Kubernetes discovery API (`/apis/{g}/{v}` or
193 * `/api/v1`). This is what lets `Kubernetes.Manifest` apply any CRD.
194 */
195export const resolveKindSpec = Effect.fn(function* ({
196 transport,
197 input,
198}: {
199 transport: ClusterTransport;
200 input: Pick<KubernetesObjectRef, "apiVersion" | "kind">;
201}) {
202 const staticSpec = lookupKubernetesKindSpec(input);
203 if (staticSpec) return staticSpec;
204
205 const cacheKey = `${transport.endpoint}|${input.apiVersion}|${input.kind}`;
206 const cached = discoveredKinds.get(cacheKey);
207 if (cached) return cached;
208
209 const discoveryPath = input.apiVersion.includes("/")
210 ? `/apis/${input.apiVersion}`
211 : `/api/${input.apiVersion}`;
212
213 const listed = (yield* requestJson({
214 transport,
215 method: "GET",
216 path: discoveryPath,
217 })) as ApiResourceList;
218
219 const resource = listed.resources?.find(
220 (candidate) =>
221 candidate.kind === input.kind &&
222 typeof candidate.name === "string" &&
223 !candidate.name.includes("/"),
224 );
225
226 if (!resource?.name) {
227 return yield* Effect.fail(
228 new KubernetesApiError({
229 method: "GET",
230 path: discoveryPath,
231 statusCode: 404,
232 body: `Kind '${input.kind}' not found in API group '${input.apiVersion}'`,
233 }),
234 );
235 }
236
237 const spec: KubernetesObjectKindSpec = {
238 plural: resource.name,
239 scope: resource.namespaced ? "Namespaced" : "Cluster",
240 applyRank: DEFAULT_APPLY_RANK,
241 };
242 discoveredKinds.set(cacheKey, spec);
243 return spec;
244});
245
246const buildPath = Effect.fn(function* ({
247 transport,
248 object,
249}: {
250 transport: ClusterTransport;
251 object: KubernetesObjectRef;
252}) {
253 const spec = yield* resolveKindSpec({ transport, input: object });
254 return yield* Effect.try({
255 try: () => buildKubernetesObjectPathWithSpec(object, spec),
256 catch: (error) =>
257 error instanceof Error ? error : new Error(String(error)),
258 });
259});
260
261// ─────────────────────────────────────────────────────────── object ops ──
262
263export const readObject = Effect.fn(function* ({
264 transport,
265 object,
266}: {
267 transport: ClusterTransport;
268 object: KubernetesObjectRef;
269}) {
270 return yield* requestJson({
271 transport,
272 method: "GET",
273 path: yield* buildPath({ transport, object }),
274 });
275});
276
277export const applyObject = Effect.fn(function* ({
278 transport,
279 object,
280}: {
281 transport: ClusterTransport;
282 object: KubernetesObjectDefinition;
283}) {
284 const basePath = yield* buildPath({
285 transport,
286 object: toKubernetesObjectRef(object),
287 });
288 const path = `${basePath}?fieldManager=${fieldManager}&force=true`;
289
290 return yield* requestJson({
291 transport,
292 method: "PATCH",
293 path,
294 body: object,
295 }).pipe(
296 // A freshly provisioned cluster's API server briefly 5xxes while
297 // warming up, and the creator's bootstrap access can propagate
298 // asynchronously (401/403 in the first minute) — retry transient
299 // failures for ~1 min.
300 Effect.retry({
301 while: (e): boolean =>
302 e instanceof KubernetesApiError &&
303 (e.statusCode >= 500 ||
304 e.statusCode === 429 ||
305 e.statusCode === 401 ||
306 e.statusCode === 403),
307 schedule: Schedule.max([
308 Schedule.spaced("6 seconds"),
309 Schedule.recurs(10),
310 ]),
311 }),
312 );
313});
314
315export const deleteObject = Effect.fn(function* ({
316 transport,
317 object,
318}: {
319 transport: ClusterTransport;
320 object: KubernetesObjectRef;
321}) {
322 yield* buildPath({ transport, object }).pipe(
323 Effect.flatMap((path) =>
324 requestJson({
325 transport,
326 method: "DELETE",
327 // batch/v1 Jobs orphan their pods on DELETE unless a propagation
328 // policy is set; Background matches kubectl's default.
329 path: `${path}?propagationPolicy=Background`,
330 }),
331 ),
332 Effect.catchIf(
333 (error): error is KubernetesApiError =>
334 error instanceof KubernetesApiError,
335 (error) => (error.statusCode === 404 ? Effect.void : Effect.fail(error)),
336 ),
337 );
338});
339
340export const reconcileObjects = Effect.fn(function* ({
341 transport,
342 previousObjects,
343 desiredObjects,
344}: {
345 transport: ClusterTransport;
346 previousObjects: ReadonlyArray<KubernetesObjectRef>;
347 desiredObjects: ReadonlyArray<KubernetesObjectDefinition>;
348}) {
349 const desiredRefs = desiredObjects.map(toKubernetesObjectRef);
350 const desiredKeys = new Set(desiredRefs.map(kubernetesObjectKey));
351
352 const removedObjects = previousObjects.filter(
353 (object) => !desiredKeys.has(kubernetesObjectKey(object)),
354 );
355
356 for (const object of sortRefsForDelete(removedObjects)) {
357 yield* deleteObject({
358 transport,
359 object,
360 });
361 }
362
363 for (const chunk of chunkByApplyRank(desiredObjects)) {
364 yield* Effect.forEach(
365 chunk,
366 (object) =>
367 applyObject({
368 transport,
369 object,
370 }),
371 {
372 concurrency: "unbounded",
373 },
374 );
375 }
376
377 return desiredRefs;
378});
379
380export const deleteObjects = Effect.fn(function* ({
381 transport,
382 objects,
383}: {
384 transport: ClusterTransport;
385 objects: ReadonlyArray<KubernetesObjectRef>;
386}) {
387 for (const object of sortRefsForDelete(objects)) {
388 yield* deleteObject({
389 transport,
390 object,
391 });
392 }
393});