1import * as logging from "@distilled.cloud/gcp/logging_v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { GcpEnvironment } from "../Environment.ts";
11import {
12 alchemyLabelKeys,
13 createInternalLabels,
14 hasAlchemyLabels,
15} from "../Labels.ts";
16import type { Providers } from "../Providers.ts";
17
18const MAX_NAME_LENGTH = 100;
19
20export type SinkExclusion = {
21 /**
22 * Client-assigned exclusion id. Limited to 100 characters (letters,
23 * digits, underscores, hyphens, periods). First character must be
24 * alphanumeric.
25 */
26 name: string;
27 /**
28 * Advanced logs filter matching entries to exclude from the export.
29 */
30 filter: string;
31 /**
32 * Human-readable description of this exclusion.
33 */
34 description?: string;
35 /**
36 * When true, the exclusion exists but does not exclude any entries.
37 * @default false
38 */
39 disabled?: boolean;
40};
41
42export type SinkBigQueryOptions = {
43 /**
44 * Use BigQuery partitioned tables instead of dated tables
45 * (`syslog_20170523`). Only applies when `destination` is a BigQuery
46 * dataset.
47 * @default false
48 */
49 usePartitionedTables?: boolean;
50};
51
52export type SinkProps = {
53 /**
54 * Sink id (the `{sink}` segment of `projects/{project}/sinks/{sink}`).
55 * If omitted, a unique name is generated from the stack, stage, and
56 * logical id. Limited to 100 characters: letters, digits, underscores,
57 * hyphens, periods; first character must be alphanumeric. Immutable —
58 * changing it replaces the sink.
59 */
60 sinkId?: string;
61 /**
62 * Export destination. One of:
63 * `storage.googleapis.com/[GCS_BUCKET]`,
64 * `bigquery.googleapis.com/projects/[PROJECT_ID]/datasets/[DATASET]`,
65 * `pubsub.googleapis.com/projects/[PROJECT_ID]/topics/[TOPIC_ID]`,
66 * `logging.googleapis.com/projects/[PROJECT_ID]`,
67 * `logging.googleapis.com/projects/[PROJECT_ID]/locations/[LOCATION_ID]/buckets/[BUCKET_ID]`.
68 */
69 destination: string;
70 /**
71 * Advanced logs filter. Only matching entries in the sink's parent are
72 * exported. Empty or omitted exports every entry.
73 */
74 filter?: string;
75 /**
76 * Human-readable description (max 8000 characters). Logging sinks have
77 * no labels field, so Alchemy ownership is stored in a `[alchemy …]`
78 * prefix and stripped from attributes.
79 */
80 description?: string;
81 /**
82 * When true, the sink exists but does not export any log entries.
83 * @default false
84 */
85 disabled?: boolean;
86 /**
87 * Exclusion filters. An entry matching `filter` and any exclusion is
88 * not exported.
89 */
90 exclusions?: SinkExclusion[];
91 /**
92 * BigQuery-specific export options. Ignored unless `destination` is a
93 * BigQuery dataset.
94 */
95 bigqueryOptions?: SinkBigQueryOptions;
96 /**
97 * When true, Cloud Logging assigns a unique writer service account as
98 * `writerIdentity`. When false (GCP default for project sinks), the
99 * shared `cloud-logs` identity is used and the destination must be in
100 * the same project. Sinks whose destination is a log bucket in the
101 * same project have no writer identity.
102 */
103 uniqueWriterIdentity?: boolean;
104 /**
105 * Caller-provided writer service account
106 * (`serviceAccount:some@email`). Only valid when routing to a log
107 * bucket in a different project.
108 */
109 customWriterIdentity?: string;
110 /**
111 * Organization/folder sinks only. When true, logs from child projects,
112 * folders, and billing accounts are also eligible for export.
113 * @default false
114 */
115 includeChildren?: boolean;
116 /**
117 * Organization/folder sinks only. When true (requires
118 * `includeChildren`), matching logs are intercepted and omitted from
119 * non-`_Required` child sinks.
120 * @default false
121 */
122 interceptChildren?: boolean;
123};
124
125export type Sink = Resource<
126 "GCP.Logging.Sink",
127 SinkProps,
128 {
129 /** Full resource name `projects/{project}/sinks/{sink}`. */
130 name: string;
131 /** Sink id (last path segment). */
132 sinkId: string;
133 /** Project id. */
134 project: string;
135 /** Export destination. */
136 destination: string;
137 /** Advanced logs filter, if set. */
138 filter: string | undefined;
139 /** User description with the Alchemy ownership prefix stripped. */
140 description: string | undefined;
141 /** Whether the sink is disabled. */
142 disabled: boolean;
143 /** Exclusion filters. */
144 exclusions: SinkExclusion[];
145 /** BigQuery export options, if set. */
146 bigqueryOptions: SinkBigQueryOptions | undefined;
147 /** Whether child resources are included (org/folder sinks). */
148 includeChildren: boolean;
149 /** Whether matching child logs are intercepted (org/folder sinks). */
150 interceptChildren: boolean;
151 /**
152 * IAM identity Cloud Logging uses to write to `destination`. Empty
153 * for same-project log-bucket destinations.
154 */
155 writerIdentity: string | undefined;
156 /** RFC3339 creation timestamp. */
157 createTime: string | undefined;
158 /** RFC3339 last-update timestamp. */
159 updateTime: string | undefined;
160 },
161 never,
162 Providers
163>;
164
165/**
166 * A Cloud Logging sink that routes log entries to a destination.
167 *
168 * Logging sinks have no labels field, so Alchemy stamps ownership into
169 * the description for `list` / nuke. Name is identity — changing `sinkId`
170 * replaces the sink. Filter, destination, description, disabled flag,
171 * exclusions, and BigQuery options update in place.
172 *
173 * ### Creating a Sink
174 * **Example:** Route errors to the `_Default` log bucket
175 * ```typescript
176 * const sink = yield* GCP.Logging.Sink("Errors", {
177 * destination:
178 * "logging.googleapis.com/projects/my-project/locations/global/buckets/_Default",
179 * filter: "severity>=ERROR",
180 * });
181 * ```
182 *
183 * **Example:** Named sink with a description
184 * ```typescript
185 * const sink = yield* GCP.Logging.Sink("Errors", {
186 * sinkId: "app-errors",
187 * destination:
188 * "logging.googleapis.com/projects/my-project/locations/global/buckets/_Default",
189 * filter: "severity>=ERROR",
190 * description: "application errors",
191 * });
192 * ```
193 *
194 * ### Updating a Sink
195 * **Example:** Change the filter and add an exclusion
196 * ```typescript
197 * const sink = yield* GCP.Logging.Sink("Errors", {
198 * destination:
199 * "logging.googleapis.com/projects/my-project/locations/global/buckets/_Default",
200 * filter: "severity>=WARNING",
201 * description: "warnings and errors",
202 * exclusions: [
203 * {
204 * name: "drop-healthchecks",
205 * filter: 'httpRequest.requestUrl="/healthz"',
206 * },
207 * ],
208 * });
209 * ```
210 *
211 * @resource
212 * @category Logging
213 */
214export const Sink = Resource<Sink>("GCP.Logging.Sink");
215
216export class SinkNotResolved extends Data.TaggedError(
217 "GCP.Logging.SinkNotResolved",
218)<{
219 name: string;
220}> {}
221
222const lastSegment = (value: string) => {
223 const trimmed = value.replace(/\/+$/, "");
224 const parts = trimmed.split("/");
225 return parts[parts.length - 1] || trimmed;
226};
227
228const resourceName = (project: string, sinkId: string) =>
229 `projects/${project}/sinks/${sinkId}`;
230
231const sinkIdOf = (sink: logging.LogSink) => {
232 const raw = sink.name ?? sink.resourceName ?? "";
233 return raw.includes("/") ? lastSegment(raw) : raw;
234};
235
236const toId = (id: string, sinkId: string | undefined, existing?: string) =>
237 Effect.gen(function* () {
238 if (sinkId !== undefined) return sinkId;
239 if (existing !== undefined) return existing;
240 const generated = yield* createPhysicalName({
241 id,
242 maxLength: MAX_NAME_LENGTH,
243 lowercase: true,
244 });
245 return /^[a-z]/.test(generated)
246 ? generated
247 : `s${generated}`.slice(0, MAX_NAME_LENGTH);
248 });
249
250const encodeDescription = (
251 labels: Record<string, string>,
252 description: string | undefined,
253): string => {
254 const marker = `[alchemy ${alchemyLabelKeys.stack}=${labels[alchemyLabelKeys.stack]} ${alchemyLabelKeys.stage}=${labels[alchemyLabelKeys.stage]} ${alchemyLabelKeys.id}=${labels[alchemyLabelKeys.id]}]`;
255 return description ? `${marker}\n${description}` : marker;
256};
257
258const parseDescription = (
259 description: string | undefined,
260): {
261 labels: Record<string, string>;
262 description: string | undefined;
263} => {
264 if (!description?.startsWith("[alchemy ")) {
265 return { labels: {}, description };
266 }
267 const end = description.indexOf("]");
268 if (end < 0) return { labels: {}, description };
269 const labels: Record<string, string> = {};
270 for (const part of description.slice("[alchemy ".length, end).split(/\s+/)) {
271 const eq = part.indexOf("=");
272 if (eq > 0) {
273 labels[part.slice(0, eq)] = part.slice(eq + 1);
274 }
275 }
276 const rest = description.slice(end + 1).replace(/^\n/, "");
277 return { labels, description: rest.length > 0 ? rest : undefined };
278};
279
280const hasOwnershipMarker = (description: string | undefined) =>
281 Object.keys(parseDescription(description).labels).some((key) =>
282 key.startsWith("alchemy-"),
283 );
284
285const exclusionsOf = (
286 list: readonly logging.LogExclusion[] | readonly SinkExclusion[] | undefined,
287): SinkExclusion[] =>
288 (list ?? [])
289 .map((exclusion) => ({
290 name: exclusion.name ?? "",
291 filter: exclusion.filter ?? "",
292 description: exclusion.description,
293 disabled: exclusion.disabled === true,
294 }))
295 .sort((left, right) => left.name.localeCompare(right.name));
296
297const sameExclusions = (
298 left: readonly logging.LogExclusion[] | readonly SinkExclusion[] | undefined,
299 right: readonly logging.LogExclusion[] | readonly SinkExclusion[] | undefined,
300) => JSON.stringify(exclusionsOf(left)) === JSON.stringify(exclusionsOf(right));
301
302const toExclusionsBody = (
303 list: readonly SinkExclusion[] | undefined,
304): logging.LogExclusion[] | undefined => {
305 if (list === undefined) return undefined;
306 return list.map((exclusion) => ({
307 name: exclusion.name,
308 filter: exclusion.filter,
309 description: exclusion.description,
310 disabled: exclusion.disabled === true ? true : undefined,
311 }));
312};
313
314const toAttrs = (sink: logging.LogSink, project: string) => {
315 const sinkId = sinkIdOf(sink);
316 const parsed = parseDescription(sink.description);
317 const bq = sink.bigqueryOptions;
318 return {
319 name: sink.resourceName ?? (sinkId ? resourceName(project, sinkId) : ""),
320 sinkId,
321 project,
322 destination: sink.destination ?? "",
323 filter: sink.filter,
324 description: parsed.description,
325 disabled: sink.disabled === true,
326 exclusions: exclusionsOf(sink.exclusions),
327 bigqueryOptions:
328 bq?.usePartitionedTables !== undefined
329 ? { usePartitionedTables: bq.usePartitionedTables }
330 : undefined,
331 includeChildren: sink.includeChildren === true,
332 interceptChildren: sink.interceptChildren === true,
333 writerIdentity: sink.writerIdentity,
334 createTime: sink.createTime,
335 updateTime: sink.updateTime,
336 };
337};
338
339const getByName = (name: string) =>
340 logging
341 .getSinks({ sinkName: name })
342 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
343
344const toCreateBody = (
345 sinkId: string,
346 props: SinkProps,
347 description: string,
348): logging.LogSink => ({
349 name: sinkId,
350 destination: props.destination,
351 filter: props.filter,
352 description,
353 disabled: props.disabled === true ? true : undefined,
354 exclusions: toExclusionsBody(props.exclusions),
355 bigqueryOptions: props.bigqueryOptions,
356 includeChildren: props.includeChildren === true ? true : undefined,
357 interceptChildren: props.interceptChildren === true ? true : undefined,
358});
359
360export const SinkProvider = () =>
361 Provider.succeed(Sink, {
362 stables: ["name", "sinkId", "project", "createTime"],
363
364 diff: Effect.fn(function* ({ news, olds, output }) {
365 if (!isResolved(news)) return undefined;
366 const previous = olds?.sinkId ?? output?.sinkId;
367 if (
368 previous !== undefined &&
369 news.sinkId !== undefined &&
370 news.sinkId !== previous
371 ) {
372 return { action: "replace" as const, deleteFirst: false };
373 }
374 return undefined;
375 }),
376
377 read: Effect.fn(function* ({ id, olds, output }) {
378 const env = yield* GcpEnvironment.current;
379 const sinkId = yield* toId(id, olds?.sinkId, output?.sinkId);
380 const name = output?.name ?? resourceName(env.project, sinkId);
381 const existing = yield* getByName(name);
382 if (existing === undefined) return undefined;
383 const attrs = toAttrs(existing, env.project);
384 const { labels } = parseDescription(existing.description);
385 return (yield* hasAlchemyLabels(id, labels)) ? attrs : Unowned(attrs);
386 }),
387
388 list: () =>
389 Effect.gen(function* () {
390 const env = yield* GcpEnvironment.current;
391 return yield* logging.listSinks
392 .pages({
393 parent: `projects/${env.project}`,
394 pageSize: 1000,
395 })
396 .pipe(
397 Stream.flatMap((page) => Stream.fromIterable(page.sinks ?? [])),
398 Stream.filter((sink) => hasOwnershipMarker(sink.description)),
399 Stream.map((sink) => toAttrs(sink, env.project)),
400 Stream.runCollect,
401 Effect.map((chunk) => Array.from(chunk)),
402 );
403 }),
404
405 reconcile: Effect.fn(function* ({ id, news, output }) {
406 const env = yield* GcpEnvironment.current;
407 const sinkId = yield* toId(id, news.sinkId, output?.sinkId);
408 const name = resourceName(env.project, sinkId);
409 const ownership = yield* createInternalLabels(id);
410 const desiredDescription = encodeDescription(ownership, news.description);
411
412 let current = yield* getByName(output?.name ?? name);
413
414 if (current === undefined) {
415 const created = yield* logging
416 .createSinks({
417 parent: `projects/${env.project}`,
418 uniqueWriterIdentity: news.uniqueWriterIdentity,
419 customWriterIdentity: news.customWriterIdentity,
420 body: toCreateBody(sinkId, news, desiredDescription),
421 })
422 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
423 current = created ?? undefined;
424 }
425
426 if (current === undefined) {
427 return yield* new SinkNotResolved({ name });
428 }
429
430 const desiredDisabled = news.disabled === true;
431 const desiredIncludeChildren = news.includeChildren === true;
432 const desiredInterceptChildren = news.interceptChildren === true;
433 const desiredBq = news.bigqueryOptions?.usePartitionedTables === true;
434 const observedBq = current.bigqueryOptions?.usePartitionedTables === true;
435
436 const destinationChanged =
437 (current.destination ?? "") !== news.destination;
438 const filterChanged = (current.filter ?? "") !== (news.filter ?? "");
439 const descriptionChanged =
440 (current.description ?? "") !== desiredDescription;
441 const disabledChanged = (current.disabled === true) !== desiredDisabled;
442 const exclusionsChanged = !sameExclusions(
443 current.exclusions,
444 news.exclusions,
445 );
446 const bqChanged =
447 news.bigqueryOptions !== undefined && desiredBq !== observedBq;
448 const includeChildrenChanged =
449 news.includeChildren !== undefined &&
450 (current.includeChildren === true) !== desiredIncludeChildren;
451 const interceptChildrenChanged =
452 news.interceptChildren !== undefined &&
453 (current.interceptChildren === true) !== desiredInterceptChildren;
454
455 const updateMask = [
456 destinationChanged ? "destination" : undefined,
457 filterChanged ? "filter" : undefined,
458 descriptionChanged ? "description" : undefined,
459 disabledChanged ? "disabled" : undefined,
460 exclusionsChanged ? "exclusions" : undefined,
461 bqChanged ? "bigqueryOptions.usePartitionedTables" : undefined,
462 includeChildrenChanged ? "includeChildren" : undefined,
463 interceptChildrenChanged ? "interceptChildren" : undefined,
464 ].filter((field): field is string => field !== undefined);
465
466 if (updateMask.length > 0) {
467 const body: logging.LogSink = {
468 name: sinkId,
469 destination: news.destination,
470 filter: news.filter ?? "",
471 description: desiredDescription,
472 disabled: desiredDisabled,
473 };
474 if (exclusionsChanged) {
475 body.exclusions = toExclusionsBody(news.exclusions) ?? [];
476 }
477 if (bqChanged) {
478 body.bigqueryOptions = {
479 usePartitionedTables: desiredBq,
480 };
481 }
482 if (includeChildrenChanged) {
483 body.includeChildren = desiredIncludeChildren;
484 }
485 if (interceptChildrenChanged) {
486 body.interceptChildren = desiredInterceptChildren;
487 }
488 current = yield* logging.updateSinks({
489 sinkName: current.resourceName ?? name,
490 uniqueWriterIdentity: news.uniqueWriterIdentity,
491 customWriterIdentity: news.customWriterIdentity,
492 updateMask: updateMask.join(","),
493 body,
494 });
495 }
496
497 return toAttrs(current, env.project);
498 }),
499
500 delete: Effect.fn(function* ({ output }) {
501 yield* logging
502 .deleteSinks({ sinkName: output.name })
503 .pipe(Effect.catchTag("NotFound", () => Effect.void));
504 }),
505 });