1/**
2 * HTTP client support for talking to the Kubernetes API from code running
3 * inside a cluster. The module builds a `K8sHttpClient` service on top of the
4 * shared HTTP client, points requests at the in-cluster API endpoint, and uses
5 * the mounted service-account token when one is available.
6 *
7 * @stability unstable
8 * @since 4.0.0
9 */
10import * as Context from "../Context.ts"
11import * as Effect from "../Effect.ts"
12import * as FileSystem from "../FileSystem.ts"
13import { identity } from "../Function.ts"
14import * as HttpClient from "../http/HttpClient.ts"
15import * as HttpClientError from "../http/HttpClientError.ts"
16import * as HttpClientRequest from "../http/HttpClientRequest.ts"
17import * as HttpClientResponse from "../http/HttpClientResponse.ts"
18import * as Layer from "../Layer.ts"
19import * as Option from "../Option.ts"
20import * as Result from "../Result.ts"
21import * as Schedule from "../Schedule.ts"
22import * as Schema from "../Schema.ts"
23import type { Pod as K8sPod } from "./K8sTypes.ts"
24
25/**
26 * Service tag for the HTTP client used to call the Kubernetes API.
27 *
28 * @stability unstable
29 * @category services
30 * @since 4.0.0
31 */
32export class K8sHttpClient extends Context.Service<
33 K8sHttpClient,
34 HttpClient.HttpClient
35>()("effect/cluster/K8sHttpClient") {}
36
37/**
38 * Layer that configures `K8sHttpClient` for the in-cluster Kubernetes API.
39 *
40 * **Details**
41 *
42 * It targets `https://kubernetes.default.svc/api`, adds the service-account
43 * bearer token when available, requires successful HTTP statuses, and retries
44 * transient failures.
45 *
46 * @stability unstable
47 * @category layers
48 * @since 4.0.0
49 */
50export const layer: Layer.Layer<
51 K8sHttpClient,
52 never,
53 HttpClient.HttpClient | FileSystem.FileSystem
54> = Layer.effect(
55 K8sHttpClient,
56 Effect.gen(function*() {
57 const fs = yield* FileSystem.FileSystem
58 const token = yield* fs.readFileString("/var/run/secrets/kubernetes.io/serviceaccount/token").pipe(
59 Effect.option
60 )
61 return (yield* HttpClient.HttpClient).pipe(
62 HttpClient.mapRequest(HttpClientRequest.prependUrl("https://kubernetes.default.svc/api")),
63 token._tag === "Some" ? HttpClient.mapRequest(HttpClientRequest.bearerToken(token.value.trim())) : identity,
64 HttpClient.filterStatusOk,
65 HttpClient.retryTransient({
66 schedule: Schedule.spaced(5000)
67 })
68 )
69 })
70)
71
72/**
73 * Creates a cached effect that fetches running Kubernetes pods.
74 *
75 * **Details**
76 *
77 * The request can be limited by namespace and label selector, and the result is a
78 * map keyed by pod IP address.
79 *
80 * @stability unstable
81 * @category constructors
82 * @since 4.0.0
83 */
84export const makeGetPods: (
85 options?: {
86 readonly namespace?: string | undefined
87 readonly labelSelector?: string | undefined
88 } | undefined
89) => Effect.Effect<
90 Effect.Effect<Map<string, Pod>, HttpClientError.HttpClientError | Schema.SchemaError, never>,
91 never,
92 K8sHttpClient
93> = Effect.fnUntraced(function*(options?: {
94 readonly namespace?: string | undefined
95 readonly labelSelector?: string | undefined
96}) {
97 const client = yield* K8sHttpClient
98
99 const getPods = HttpClientRequest.get(
100 options?.namespace ? `/v1/namespaces/${options.namespace}/pods` : "/v1/pods"
101 ).pipe(
102 HttpClientRequest.setUrlParam("fieldSelector", "status.phase=Running"),
103 options?.labelSelector ? HttpClientRequest.setUrlParam("labelSelector", options.labelSelector) : identity
104 )
105
106 return yield* client.execute(getPods).pipe(
107 Effect.flatMap(HttpClientResponse.schemaBodyJson(PodList)),
108 Effect.map((list) => {
109 const pods = new Map<string, Pod>()
110 for (let i = 0; i < list.items.length; i++) {
111 const pod = list.items[i]
112 pods.set(pod.status.podIP, pod)
113 }
114 return pods
115 }),
116 Effect.tapCause((cause) => Effect.logWarning("Failed to fetch pods from Kubernetes API", cause)),
117 Effect.cachedWithTTL("10 seconds")
118 )
119})
120
121/**
122 * Creates a scoped function that ensures a Kubernetes pod exists and waits until
123 * it is ready.
124 *
125 * **Details**
126 *
127 * The pod defaults to the `default` namespace and is deleted when the surrounding
128 * scope closes.
129 *
130 * @stability unstable
131 * @category constructors
132 * @since 4.0.0
133 */
134export const makeCreatePod = Effect.gen(function*() {
135 const client = yield* K8sHttpClient
136
137 return Effect.fnUntraced(function*(spec: K8sPod) {
138 spec = {
139 apiVersion: "v1",
140 kind: "Pod",
141 metadata: {
142 namespace: "default",
143 ...spec.metadata
144 },
145 ...spec
146 }
147 const namespace = spec.metadata?.namespace ?? "default"
148 const name = spec.metadata!.name!
149 const readPodRaw = HttpClientRequest.get(`/v1/namespaces/${namespace}/pods/${name}`).pipe(
150 client.execute
151 )
152 const readPod = readPodRaw.pipe(
153 Effect.flatMap(HttpClientResponse.schemaBodyJson(Pod)),
154 Effect.asSome,
155 Effect.retry({
156 while: (e) => e._tag === "SchemaError",
157 schedule: Schedule.spaced("1 seconds")
158 }),
159 Effect.catchFilter(
160 (err) =>
161 HttpClientError.isHttpClientError(err) && err.reason._tag === "StatusCodeError" &&
162 err.reason.response.status === 404
163 ? Result.succeed(err)
164 : Result.fail(err),
165 () => Effect.succeedNone
166 ),
167 Effect.orDie
168 )
169 const isPodFound = readPodRaw.pipe(
170 Effect.as(true),
171 Effect.catchFilter(
172 (err) =>
173 HttpClientError.isHttpClientError(err) && err.reason._tag === "StatusCodeError" &&
174 err.reason.response.status === 404
175 ? Result.succeed(err)
176 : Result.fail(err),
177 () => Effect.succeed(false)
178 )
179 )
180 const createPod = HttpClientRequest.post(`/v1/namespaces/${namespace}/pods`).pipe(
181 HttpClientRequest.bodyJsonUnsafe(spec),
182 client.execute,
183 Effect.catchFilter(
184 (err) =>
185 HttpClientError.isHttpClientError(err) && err.reason._tag === "StatusCodeError" &&
186 err.reason.response.status === 409
187 ? Result.succeed(err)
188 : Result.fail(err),
189 () => readPod
190 ),
191 Effect.tapCause(Effect.logInfo),
192 Effect.orDie
193 )
194 const deletePod = HttpClientRequest.delete(`/v1/namespaces/${namespace}/pods/${name}`).pipe(
195 client.execute,
196 Effect.flatMap((res) => res.json),
197 Effect.catchFilter(
198 (err) =>
199 HttpClientError.isHttpClientError(err) && err.reason._tag === "StatusCodeError" &&
200 err.reason.response.status === 404
201 ? Result.succeed(err)
202 : Result.fail(err),
203 () => Effect.void
204 ),
205 Effect.tapCause(Effect.logInfo),
206 Effect.orDie,
207 Effect.asVoid
208 )
209 yield* Effect.addFinalizer(Effect.fnUntraced(function*() {
210 yield* deletePod
211 yield* isPodFound.pipe(
212 Effect.repeat({
213 until: (found) => !found,
214 schedule: Schedule.spaced("3 seconds")
215 }),
216 Effect.orDie
217 )
218 }))
219
220 let opod = Option.none<Pod>()
221 while (Option.isNone(opod) || !opod.value.isReady) {
222 if (Option.isNone(opod)) {
223 yield* createPod
224 }
225 yield* Effect.sleep("3 seconds")
226 opod = yield* readPod
227 }
228 return opod.value.status
229 }, Effect.withSpan("K8sHttpClient.createPod"))
230})
231
232/**
233 * Schema for the subset of Kubernetes Pod status used by cluster helpers.
234 *
235 * @stability unstable
236 * @category schemas
237 * @since 4.0.0
238 */
239export class PodStatus extends Schema.Class<PodStatus>("@effect/cluster/K8sHttpClient/PodStatus")({
240 phase: Schema.String,
241 conditions: Schema.Array(Schema.Struct({
242 type: Schema.String,
243 status: Schema.String,
244 lastTransitionTime: Schema.NullOr(Schema.String)
245 })),
246 podIP: Schema.String,
247 hostIP: Schema.String
248}) {}
249
250/**
251 * Schema for Kubernetes Pod values used by cluster helpers.
252 *
253 * **Details**
254 *
255 * The model exposes readiness helpers derived from the pod status conditions.
256 *
257 * @stability unstable
258 * @category schemas
259 * @since 4.0.0
260 */
261export class Pod extends Schema.Class<Pod>("@effect/cluster/K8sHttpClient/Pod")({
262 status: PodStatus
263}) {
264 get isReady(): boolean {
265 for (let i = 0; i < this.status.conditions.length; i++) {
266 const condition = this.status.conditions[i]
267 if (condition.type === "Ready") {
268 return condition.status === "True"
269 }
270 }
271 return false
272 }
273
274 get isReadyOrInitializing(): boolean {
275 let initializedAt: string | null | undefined
276 let readyAt: string | null | undefined
277 for (let i = 0; i < this.status.conditions.length; i++) {
278 const condition = this.status.conditions[i]
279 switch (condition.type) {
280 case "Initialized": {
281 if (condition.status !== "True") {
282 return true
283 }
284 initializedAt = condition.lastTransitionTime
285 break
286 }
287 case "Ready": {
288 if (condition.status === "True") {
289 return true
290 }
291 readyAt = condition.lastTransitionTime
292 break
293 }
294 }
295 }
296 // if the pod is still booting up, consider it ready as it would have
297 // already registered itself with RunnerStorage by now
298 return initializedAt === readyAt
299 }
300}
301
302const PodList = Schema.Struct({
303 items: Schema.Array(Pod)
304})