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