diff --git a/cluster/pulumi/common-sv/src/bigQuery.ts b/cluster/pulumi/common-sv/src/bigQuery.ts index b45c294d12..92f73ff906 100644 --- a/cluster/pulumi/common-sv/src/bigQuery.ts +++ b/cluster/pulumi/common-sv/src/bigQuery.ts @@ -355,7 +355,9 @@ function installBigqueryStagingDataset(scanBigQuery: ScanBigQueryConfig): gcp.bi friendlyName: `${scanBigQuery.dataset} Staging Dataset`, location: cloudsdkComputeRegion(), deleteContentsOnDestroy: true, - defaultTableExpirationMs: THREE_DAYS_MS, + // ISSUE#6814: Do not rely on ingestion timestamps for retention in staging. + // GCP calculates expiration from the table creation date, which will delete + // staging tables at the 3-day mark even if production sync is incomplete. labels: { cluster: CLUSTER_BASENAME, datastream_id: 'stag_prod', @@ -368,27 +370,28 @@ function installBigqueryProdDataset(scanBigQuery: ScanBigQueryConfig): gcp.bigqu datasetId: `${scanBigQuery.dataset}_prod`, friendlyName: `${scanBigQuery.dataset} Production Dataset`, location: cloudsdkComputeRegion(), - deleteContentsOnDestroy: true, + deleteContentsOnDestroy: false, labels: { cluster: CLUSTER_BASENAME, }, }); } - // ============================================================================ -// HOURLY DEDUPLICATION & SCHEDULED QUERIES +// IAM PERMISSIONS for SCHEDULED QUERIES // ============================================================================ +interface ScheduledQueryContext { + projectId: pulumi.Output; + transferServiceAgentPermission: gcp.projects.IAMMember; +} -const rawSqlTemplate = fs.readFileSync(path.join(__dirname, 'hourly_append.sql'), 'utf8'); - -function installHourlyScheduledQueries( - namespace: ExactNamespace, - stagingDataset: gcp.bigquery.Dataset, - prodDataset: gcp.bigquery.Dataset -) { +function installBqScheduledQueryContext(): ScheduledQueryContext { const currentProject = gcp.organizations.getProjectOutput({}); - const projectId = currentProject.apply(p => p.projectId!); - const schemaName = scanAppDatabaseName(namespace); + const projectId = currentProject.apply(p => { + if (!p.projectId) { + throw new Error('Current GCP project output is missing a projectId.'); + } + return p.projectId; + }); const transferServiceAgentPermission = new gcp.projects.IAMMember('bq-transfer-token-creator', { project: projectId, @@ -398,6 +401,23 @@ function installHourlyScheduledQueries( ), }); + return { projectId, transferServiceAgentPermission }; +} + +// ============================================================================ +// HOURLY DEDUPLICATION & SCHEDULED QUERIES +// ============================================================================ + +const rawSqlTemplate = fs.readFileSync(path.join(__dirname, 'hourly_append.sql'), 'utf8'); + +function installHourlyScheduledQueries( + namespace: ExactNamespace, + stagingDataset: gcp.bigquery.Dataset, + prodDataset: gcp.bigquery.Dataset, + context: ScheduledQueryContext +) { + const { projectId, transferServiceAgentPermission } = context; + const schemaName = scanAppDatabaseName(namespace); Object.entries(replicatedTables).forEach(([tableName, tableConfig]) => { const primaryKeyExpr = tableConfig.primaryKey; const colName = tableConfig.datePartitionColumn; @@ -463,7 +483,67 @@ function installHourlyScheduledQueries( ); }); } +// ============================================================================ +// Purging data older than 7 days from the staging table +// ============================================================================ +function installDailyPurgeScheduledQueries( + namespace: ExactNamespace, + stagingDataset: gcp.bigquery.Dataset, + context: ScheduledQueryContext, + retentionPeriodSeconds: number +) { + const { projectId, transferServiceAgentPermission } = context; + const schemaName = scanAppDatabaseName(namespace); + const retentionDays = retentionPeriodSeconds / 86400; + Object.entries(replicatedTables).forEach(([tableName, tableConfig]) => { + const timeExpression = + tableConfig.timeType === 'datastream_metadata' + ? `TIMESTAMP_MILLIS(datastream_metadata.${tableConfig.datePartitionColumn})` + : `TIMESTAMP_MICROS(${tableConfig.datePartitionColumn})`; + + const procedureBody = pulumi + .all([projectId, stagingDataset.datasetId]) + .apply(([proj, stagingDs]) => { + const stagingTable = `\`${proj}.${stagingDs}.${schemaName}_${tableName}\``; + + return ` + DELETE FROM ${stagingTable} + WHERE ${timeExpression} < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL ${retentionPeriodSeconds} SECOND); + `; + }); + + const routineId = `sp_purge_old_records_${tableName}`; + + // Create the cleanup Stored Procedure + const purgeRoutine = new gcp.bigquery.Routine(`${tableName}-purge-routine`, { + datasetId: stagingDataset.datasetId, + routineId: routineId, + routineType: 'PROCEDURE', + language: 'SQL', + definitionBody: procedureBody, + }); + + // Schedule the Stored Procedure to run daily + new gcp.bigquery.DataTransferConfig( + `${CLUSTER_BASENAME}_${tableName}-daily-purge`, + { + displayName: `${CLUSTER_BASENAME}_${tableName} Daily Retention Purge`, + location: cloudsdkComputeRegion(), + serviceAccountName: pulumi.interpolate`bigquery@${projectId}.iam.gserviceaccount.com`, + dataSourceId: 'scheduled_query', + schedule: 'every day 05:21', // Runs daily at 05:21 AM + + params: { + query: pulumi.interpolate`CALL \`${projectId}.${stagingDataset.datasetId}.${routineId}\`();`, + }, + }, + { + dependsOn: [transferServiceAgentPermission, purgeRoutine], + } + ); + }); +} // ============================================================================ // CONNECTION PROFILES & NETWORKING // ============================================================================ @@ -775,6 +855,7 @@ export async function configureScanBigQuery({ enableStagProdDatastream, legacyDesiredState, stagProdDesiredState, + retentionPeriodSeconds, } = bigQueryConfig; if (!enableLegacyDatastream && !enableStagProdDatastream) { @@ -856,8 +937,14 @@ export async function configureScanBigQuery({ slots.slot2, stagProdDesiredState ); - - installHourlyScheduledQueries(namespace, stagingDataset, prodDataset); + const scheduledQueryContext = installBqScheduledQueryContext(); + installHourlyScheduledQueries(namespace, stagingDataset, prodDataset, scheduledQueryContext); + installDailyPurgeScheduledQueries( + namespace, + stagingDataset, + scheduledQueryContext, + retentionPeriodSeconds + ); } // TODO (DACH-NY/canton-network-internal#6451) not sure if this function needs to return anything, // but we need to return something to satisfy the ScanBigQuery type. diff --git a/cluster/pulumi/common-sv/src/singleSvConfig.ts b/cluster/pulumi/common-sv/src/singleSvConfig.ts index fc4d720b57..bbc8f07cd6 100644 --- a/cluster/pulumi/common-sv/src/singleSvConfig.ts +++ b/cluster/pulumi/common-sv/src/singleSvConfig.ts @@ -98,6 +98,7 @@ export type BulkStorageConfig = z.infer; // 1. Extract ScanBigQueryConfigSchema to validate all Datastream settings. // All new fields are optional to ensure existing deployments do not fail parsing. +const SECONDS_PER_DAY = 24 * 3600; export const ScanBigQueryConfigSchema = z .object({ dataset: z.string(), @@ -107,6 +108,15 @@ export const ScanBigQueryConfigSchema = z enableStagProdDatastream: z.boolean().default(false), legacyDesiredState: z.enum(['RUNNING', 'PAUSED']).default('RUNNING'), stagProdDesiredState: z.enum(['RUNNING', 'PAUSED']).default('RUNNING'), + retentionPeriodSeconds: z + .number() + .min(3 * SECONDS_PER_DAY, { + message: 'Value must be at least 3 days (259,200 seconds)', + }) + .refine(v => v % SECONDS_PER_DAY === 0, { + message: 'Value must be an exact number of days, expressed in seconds', + }) + .default(7 * SECONDS_PER_DAY), }) .strict(); // Keeps strict mode safe now that all known fields are explicitly defined diff --git a/cluster/pulumi/infra/src/cloudArmor.ts b/cluster/pulumi/infra/src/cloudArmor.ts index 82fda8ef65..de26856a32 100644 --- a/cluster/pulumi/infra/src/cloudArmor.ts +++ b/cluster/pulumi/infra/src/cloudArmor.ts @@ -76,9 +76,8 @@ export function configureCloudArmorPolicy( // Step 2: Add predefined WAF rules if (cac.predefinedWafRules && cac.predefinedWafRules.length > 0) { - addPredefinedWafRules( - /*securityPolicy, args.predefinedWafRules, cac.allRulesPreviewOnly, ruleOpts*/ - ); + addPredefinedWafRules(); + /*securityPolicy, args.predefinedWafRules, cac.allRulesPreviewOnly, ruleOpts*/ } // Step 3: Add IP whitelisting rules