1import * as amp from "@distilled.cloud/aws/amp";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
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 { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 decodeDefinition,
13 encodeDefinition,
14 syncAmpTags,
15 toTagRecord,
16} from "./internal.ts";
17
18export interface ScraperEksSource {
19 /**
20 * ARN of the Amazon EKS cluster the scraper collects metrics from.
21 */
22 clusterArn: string;
23 /**
24 * IDs of the subnets the scraper's network interfaces are placed in. Must
25 * be subnets of the cluster's VPC.
26 */
27 subnetIds: string[];
28 /**
29 * IDs of the security groups applied to the scraper's network interfaces.
30 * If omitted, the cluster's default security group is used.
31 */
32 securityGroupIds?: string[];
33}
34
35export interface ScraperVpcSource {
36 /**
37 * IDs of the subnets the scraper's network interfaces are placed in.
38 */
39 subnetIds: string[];
40 /**
41 * IDs of the security groups applied to the scraper's network interfaces.
42 */
43 securityGroupIds: string[];
44}
45
46/**
47 * Where the scraper collects metrics from — an Amazon EKS cluster or a
48 * VPC-based source (Amazon MSK, self-managed Kubernetes, or any
49 * Prometheus-compatible endpoint discoverable via DNS in the VPC).
50 */
51export type ScraperSource =
52 | { eksConfiguration: ScraperEksSource }
53 | { vpcConfiguration: ScraperVpcSource };
54
55export interface ScraperRoleConfiguration {
56 /**
57 * ARN of the role used by the scraper to discover and collect metrics on
58 * your behalf (cross-account collection).
59 */
60 sourceRoleArn?: string;
61 /**
62 * ARN of the role used by the scraper to write to the destination
63 * workspace (cross-account collection).
64 */
65 targetRoleArn?: string;
66}
67
68export interface ScraperProps {
69 /**
70 * A human-readable alias for the scraper. Aliases are not unique.
71 * Updating the alias is an in-place update.
72 */
73 alias?: string;
74 /**
75 * The scraper's configuration as Prometheus scrape-configuration YAML
76 * text. It is base64-blob encoded on the wire automatically. Use the
77 * {@link GetDefaultScraperConfiguration} binding (or the
78 * `GetDefaultScraperConfiguration` API) to fetch the AWS-managed default
79 * as a starting point. Updating the configuration is an in-place update.
80 */
81 scrapeConfiguration: string;
82 /**
83 * Where the scraper collects metrics from. The source is immutable —
84 * changing it replaces the scraper.
85 */
86 source: ScraperSource;
87 /**
88 * ARN of the Amazon Managed Service for Prometheus workspace the scraped
89 * metrics are written to. Updating the destination is an in-place update.
90 */
91 destinationWorkspaceArn: string;
92 /**
93 * Cross-account role configuration for the scraper. If omitted, the
94 * scraper collects and writes within the current account using the
95 * service-created role.
96 */
97 roleConfiguration?: ScraperRoleConfiguration;
98 /**
99 * User-defined tags for the scraper.
100 */
101 tags?: Record<string, string>;
102}
103
104export interface Scraper extends Resource<
105 "AWS.AMP.Scraper",
106 ScraperProps,
107 {
108 scraperId: string;
109 scraperArn: string;
110 roleArn: string;
111 alias: string | undefined;
112 status: string;
113 },
114 never,
115 Providers
116> {}
117
118/**
119 * An Amazon Managed Service for Prometheus scraper — a fully-managed,
120 * agentless collector that pulls metrics from an Amazon EKS cluster (or a
121 * VPC-based Prometheus-compatible source) and remote-writes them into an AMP
122 * workspace.
123 *
124 * Scraper provisioning is slow (the service creates network interfaces and
125 * an IAM role; expect several minutes to reach `ACTIVE`).
126 *
127 * ### Creating a Scraper
128 * **Example:** Scrape an EKS Cluster into a Workspace
129 * ```typescript
130 * const workspace = yield* AMP.Workspace("Metrics", {});
131 * const scraper = yield* AMP.Scraper("ClusterScraper", {
132 * alias: "eks-metrics",
133 * scrapeConfiguration: defaultScrapeConfigYaml,
134 * source: {
135 * eksConfiguration: {
136 * clusterArn: cluster.clusterArn,
137 * subnetIds: [subnetA.subnetId, subnetB.subnetId],
138 * },
139 * },
140 * destinationWorkspaceArn: workspace.workspaceArn,
141 * });
142 * ```
143 *
144 * **Example:** Scrape a VPC-Based Source
145 * ```typescript
146 * const scraper = yield* AMP.Scraper("VpcScraper", {
147 * scrapeConfiguration: scrapeConfigYaml,
148 * source: {
149 * vpcConfiguration: {
150 * subnetIds: [subnet.subnetId],
151 * securityGroupIds: [securityGroup.securityGroupId],
152 * },
153 * },
154 * destinationWorkspaceArn: workspace.workspaceArn,
155 * });
156 * ```
157 *
158 * @resource
159 */
160export const Scraper = Resource<Scraper>("AWS.AMP.Scraper");
161
162/** Canonical JSON of a scraper source for observed-vs-desired comparison. */
163const canonicalSource = (source: ScraperSource): string => {
164 const canonical = (config: {
165 clusterArn?: string;
166 subnetIds: string[];
167 securityGroupIds?: string[];
168 }) => ({
169 clusterArn: config.clusterArn ?? null,
170 subnetIds: [...config.subnetIds].sort(),
171 securityGroupIds: [...(config.securityGroupIds ?? [])].sort(),
172 });
173 return "eksConfiguration" in source
174 ? JSON.stringify({ eks: canonical(source.eksConfiguration) })
175 : JSON.stringify({ vpc: canonical(source.vpcConfiguration) });
176};
177
178export const ScraperProvider = () =>
179 Provider.effect(
180 Scraper,
181 Effect.gen(function* () {
182 const toAttrs = (scraper: {
183 scraperId: string;
184 arn: string;
185 roleArn: string;
186 alias?: string;
187 status: amp.ScraperStatus;
188 }) => ({
189 scraperId: scraper.scraperId,
190 scraperArn: scraper.arn,
191 roleArn: scraper.roleArn,
192 alias: scraper.alias,
193 status: scraper.status.statusCode,
194 });
195
196 /** Describe a scraper by id; typed not-found → undefined. */
197 const describe = Effect.fn(function* (scraperId: string) {
198 const response = yield* amp
199 .describeScraper({ scraperId })
200 .pipe(
201 Effect.catchTag("ResourceNotFoundException", () =>
202 Effect.succeed(undefined),
203 ),
204 );
205 return response?.scraper;
206 });
207
208 const toWireSource = (source: ScraperSource): amp.Source =>
209 "eksConfiguration" in source
210 ? {
211 eksConfiguration: {
212 clusterArn: source.eksConfiguration.clusterArn,
213 subnetIds: source.eksConfiguration.subnetIds,
214 securityGroupIds: source.eksConfiguration.securityGroupIds,
215 },
216 }
217 : {
218 vpcConfiguration: {
219 subnetIds: source.vpcConfiguration.subnetIds,
220 securityGroupIds: source.vpcConfiguration.securityGroupIds,
221 },
222 };
223
224 /**
225 * Poll until the scraper leaves CREATING/UPDATING. Scraper
226 * provisioning creates network interfaces in each subnet and can take
227 * many minutes — the wait is bounded at 25 minutes and fails fast on a
228 * *_FAILED terminal status.
229 */
230 const waitActive = Effect.fn(function* (scraperId: string) {
231 const scraper = yield* amp.describeScraper({ scraperId }).pipe(
232 Effect.map((r) => r.scraper),
233 Effect.repeat({
234 schedule: Schedule.max([
235 Schedule.fixed("15 seconds"),
236 Schedule.recurs(100),
237 ]),
238 until: (s): boolean =>
239 s.status.statusCode !== "CREATING" &&
240 s.status.statusCode !== "UPDATING",
241 }),
242 );
243 if (scraper.status.statusCode !== "ACTIVE") {
244 return yield* Effect.fail(
245 new Error(
246 `AMP scraper ${scraperId} did not become ACTIVE (status: ${scraper.status.statusCode}${
247 scraper.statusReason ? `: ${scraper.statusReason}` : ""
248 })`,
249 ),
250 );
251 }
252 return scraper;
253 });
254
255 return {
256 stables: ["scraperId", "scraperArn", "roleArn"],
257
258 diff: Effect.fn(function* ({ olds, news }) {
259 if (!isResolved(news)) return undefined;
260 // The metric source is immutable — a change replaces the scraper.
261 if (
262 olds !== undefined &&
263 canonicalSource(olds.source) !== canonicalSource(news.source)
264 ) {
265 return { action: "replace" } as const;
266 }
267 }),
268
269 read: Effect.fn(function* ({ id, output }) {
270 // Scraper ids are server-assigned; without an output cache there
271 // is no deterministic identity to look up.
272 if (!output?.scraperId) return undefined;
273 const scraper = yield* describe(output.scraperId);
274 if (scraper === undefined) return undefined;
275 const attrs = toAttrs(scraper);
276 const tags = toTagRecord(scraper.tags);
277 return (yield* hasAlchemyTags(id, tags)) ? attrs : Unowned(attrs);
278 }),
279
280 reconcile: Effect.fn(function* ({ id, news, output, session }) {
281 const internalTags = yield* createInternalTags(id);
282 const desiredTags = { ...internalTags, ...news!.tags };
283 const desiredConfig = news!.scrapeConfiguration;
284
285 // 1. Observe — cloud state is authoritative; output is an id cache.
286 let scraper =
287 output?.scraperId !== undefined
288 ? yield* describe(output.scraperId)
289 : undefined;
290
291 // 2. Ensure — create if missing, then wait for ACTIVE.
292 if (scraper === undefined) {
293 const created = yield* amp.createScraper({
294 alias: news!.alias,
295 scrapeConfiguration: {
296 configurationBlob: yield* encodeDefinition(desiredConfig),
297 },
298 source: toWireSource(news!.source),
299 destination: {
300 ampConfiguration: {
301 workspaceArn: news!.destinationWorkspaceArn,
302 },
303 },
304 roleConfiguration: news!.roleConfiguration,
305 tags: desiredTags,
306 });
307 scraper = yield* waitActive(created.scraperId);
308 }
309
310 const scraperId = scraper.scraperId;
311
312 // 3. Sync mutable aspects — diff OBSERVED state against desired
313 // and issue a single update only when something drifted.
314 const observedConfig = yield* decodeDefinition(
315 scraper.scrapeConfiguration.configurationBlob,
316 );
317 const observedDestination =
318 scraper.destination.ampConfiguration?.workspaceArn;
319 const aliasDrifts =
320 (news!.alias ?? undefined) !== (scraper.alias ?? undefined);
321 const configDrifts = observedConfig !== desiredConfig;
322 const destinationDrifts =
323 observedDestination !== news!.destinationWorkspaceArn;
324 const roleDrifts =
325 news!.roleConfiguration !== undefined &&
326 ((news!.roleConfiguration.sourceRoleArn ?? undefined) !==
327 (scraper.roleConfiguration?.sourceRoleArn ?? undefined) ||
328 (news!.roleConfiguration.targetRoleArn ?? undefined) !==
329 (scraper.roleConfiguration?.targetRoleArn ?? undefined));
330
331 if (aliasDrifts || configDrifts || destinationDrifts || roleDrifts) {
332 yield* amp.updateScraper({
333 scraperId,
334 alias: news!.alias,
335 scrapeConfiguration: configDrifts
336 ? {
337 configurationBlob: yield* encodeDefinition(desiredConfig),
338 }
339 : undefined,
340 destination: destinationDrifts
341 ? {
342 ampConfiguration: {
343 workspaceArn: news!.destinationWorkspaceArn,
344 },
345 }
346 : undefined,
347 roleConfiguration: roleDrifts
348 ? news!.roleConfiguration
349 : undefined,
350 });
351 scraper = yield* waitActive(scraperId);
352 }
353
354 // 3b. Sync tags — diff against OBSERVED cloud tags.
355 yield* syncAmpTags(scraper.arn, desiredTags);
356
357 yield* session.note(scraperId);
358 return toAttrs(scraper);
359 }),
360
361 delete: Effect.fn(function* ({ output }) {
362 yield* amp.deleteScraper({ scraperId: output.scraperId }).pipe(
363 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
364 // A scraper mid-transition rejects deletion; retry briefly.
365 Effect.retry({
366 while: (e) => e._tag === "ConflictException",
367 schedule: Schedule.max([
368 Schedule.fixed("10 seconds"),
369 Schedule.recurs(30),
370 ]),
371 }),
372 );
373 }),
374
375 list: () =>
376 amp.listScrapers.pages({}).pipe(
377 Stream.runCollect,
378 Effect.map((chunk) =>
379 Array.from(chunk).flatMap((page) =>
380 page.scrapers.map((summary) => toAttrs(summary)),
381 ),
382 ),
383 ),
384 };
385 }),
386 );