From 200aa72952b0e2bbc37179799078e7bf113ea3e0 Mon Sep 17 00:00:00 2001 From: jasonvinson Date: Wed, 15 Jul 2026 14:45:04 -0400 Subject: [PATCH] Remove deprecated quicksight related code --- terraform/services/quicksight/athena.tf | 111 --- terraform/services/quicksight/firehose-agg.tf | 78 -- terraform/services/quicksight/firehose-api.tf | 57 -- terraform/services/quicksight/glue.tf | 272 ------- terraform/services/quicksight/iam.tf | 680 ------------------ terraform/services/quicksight/lambda.tf | 38 - .../dpc-bfd-cwlog-basic-flatten-json.py | 386 ---------- terraform/services/quicksight/main.tf | 57 -- .../sql_templates/bulk_requests.sql.tfpl | 26 - .../everything_requests.sql.tfpl | 26 - .../sql_templates/group_requests.sql.tfpl | 25 - .../sql_templates/since_requests.sql.tfpl | 26 - .../total_benes_requested.sql.tfpl | 30 - .../sql_templates/uniq_benes_served.sql.tfpl | 30 - terraform/services/quicksight/terraform.tf | 18 - terraform/services/quicksight/variables.tf | 17 - 16 files changed, 1877 deletions(-) delete mode 100644 terraform/services/quicksight/athena.tf delete mode 100644 terraform/services/quicksight/firehose-agg.tf delete mode 100644 terraform/services/quicksight/firehose-api.tf delete mode 100644 terraform/services/quicksight/glue.tf delete mode 100644 terraform/services/quicksight/iam.tf delete mode 100644 terraform/services/quicksight/lambda.tf delete mode 100644 terraform/services/quicksight/lambda_src/dpc-bfd-cwlog-basic-flatten-json.py delete mode 100644 terraform/services/quicksight/main.tf delete mode 100644 terraform/services/quicksight/sql_templates/bulk_requests.sql.tfpl delete mode 100644 terraform/services/quicksight/sql_templates/everything_requests.sql.tfpl delete mode 100644 terraform/services/quicksight/sql_templates/group_requests.sql.tfpl delete mode 100644 terraform/services/quicksight/sql_templates/since_requests.sql.tfpl delete mode 100644 terraform/services/quicksight/sql_templates/total_benes_requested.sql.tfpl delete mode 100644 terraform/services/quicksight/sql_templates/uniq_benes_served.sql.tfpl delete mode 100644 terraform/services/quicksight/terraform.tf delete mode 100644 terraform/services/quicksight/variables.tf diff --git a/terraform/services/quicksight/athena.tf b/terraform/services/quicksight/athena.tf deleted file mode 100644 index 09e2320cf..000000000 --- a/terraform/services/quicksight/athena.tf +++ /dev/null @@ -1,111 +0,0 @@ -resource "aws_athena_workgroup" "quicksight" { - name = local.athena_workgroup_name - - depends_on = [aws_s3_object.folder] - - configuration { - enforce_workgroup_configuration = true - publish_cloudwatch_metrics_enabled = true - - result_configuration { - output_location = "s3://${local.dpc_athena_bucket_id}/${local.dpc_athena_results_folder_key}" - - encryption_configuration { - encryption_option = "SSE_KMS" - kms_key_arn = local.dpc_athena_bucket_key_arn - } - } - } -} - -resource "aws_athena_database" "quicksight" { - name = local.athena_profile - bucket = local.dpc_athena_bucket_id -} - -resource "aws_athena_named_query" "total_benes_req" { - name = "${local.agg_profile}-total-benes" - description = "total unique beneficiaries requested" - workgroup = aws_athena_workgroup.quicksight.id - database = aws_athena_database.quicksight.id - query = templatefile("${path.module}/sql_templates/total_benes_requested.sql.tfpl", - { - env = "'${local.this_env}'", - agg_profile = "\"${aws_glue_catalog_database.agg.name}\".\"${aws_glue_catalog_table.agg_metric_table.name}\"", - days_history = 7 - } - ) -} - -resource "aws_athena_named_query" "unique_benes_served" { - name = "${local.agg_profile}-uniq-benes-served" - description = "unique beneficiaries served" - workgroup = aws_athena_workgroup.quicksight.id - database = aws_athena_database.quicksight.id - query = templatefile("${path.module}/sql_templates/uniq_benes_served.sql.tfpl", - { - env = "'${local.this_env}'", - agg_profile = "\"${aws_glue_catalog_database.agg.name}\".\"${aws_glue_catalog_table.agg_metric_table.name}\"", - days_history = 7 - } - ) -} - -resource "aws_athena_named_query" "group_requests" { - name = "${local.api_profile}-group-requests" - description = "/Group requests made" - workgroup = aws_athena_workgroup.quicksight.id - database = aws_athena_database.quicksight.id - query = templatefile("${path.module}/sql_templates/group_requests.sql.tfpl", - { - env = "'${local.this_env}'", - api_profile = "\"${aws_glue_catalog_database.api.name}\".\"${aws_glue_catalog_table.api_metric_table.name}\"", - days_history = 7 - } - ) -} - -resource "aws_athena_named_query" "bulk_calls_made" { - name = "${local.api_profile}-bulk-data-requests" - description = "bulk data requests made" - workgroup = aws_athena_workgroup.quicksight.id - database = aws_athena_database.quicksight.id - query = templatefile("${path.module}/sql_templates/bulk_requests.sql.tfpl", - { - app = "'dpc-api'", - env = "'${local.this_env}'", - api_profile = "\"${aws_glue_catalog_database.api.name}\".\"${aws_glue_catalog_table.api_metric_table.name}\"", - days_history = 7 - } - ) -} - -resource "aws_athena_named_query" "everything_calls_made" { - name = "${local.api_profile}-everything-data-requests" - description = "data requests made with everything parameter" - workgroup = aws_athena_workgroup.quicksight.id - database = aws_athena_database.quicksight.id - query = templatefile("${path.module}/sql_templates/everything_requests.sql.tfpl", - { - app = "'dpc-api'", - env = "'${local.this_env}'", - api_profile = "\"${aws_glue_catalog_database.api.name}\".\"${aws_glue_catalog_table.api_metric_table.name}\"", - days_history = 7 - } - ) -} - -resource "aws_athena_named_query" "since_calls_made" { - name = "${local.api_profile}-since-data-requests" - description = "data requests made with _since parameter" - workgroup = aws_athena_workgroup.quicksight.id - database = aws_athena_database.quicksight.id - query = templatefile("${path.module}/sql_templates/since_requests.sql.tfpl", - { - app = "'dpc-api'", - env = "'${local.this_env}'", - api_profile = "\"${aws_glue_catalog_database.api.name}\".\"${aws_glue_catalog_table.api_metric_table.name}\"", - days_history = 7 - } - ) -} \ No newline at end of file diff --git a/terraform/services/quicksight/firehose-agg.tf b/terraform/services/quicksight/firehose-agg.tf deleted file mode 100644 index bbb8b8c3e..000000000 --- a/terraform/services/quicksight/firehose-agg.tf +++ /dev/null @@ -1,78 +0,0 @@ -# Firehose Data Stream -resource "aws_kinesis_firehose_delivery_stream" "ingester_agg" { - depends_on = [aws_glue_catalog_table.agg_metric_table] - name = "${local.stack_prefix}-ingester_agg" - destination = "extended_s3" - - extended_s3_configuration { - bucket_arn = local.dpc_glue_bucket_arn - buffering_interval = 300 - buffering_size = 128 - error_output_prefix = "databases/${local.agg_profile}/filter_errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/" - - prefix = "databases/${local.agg_profile}/metric_table/year=!{timestamp:yyyy}/month=!{timestamp:MM}/" - kms_key_arn = local.dpc_glue_bucket_key_arn - - role_arn = aws_iam_role.iam-role-firehose.arn - s3_backup_mode = "Disabled" - compression_format = "UNCOMPRESSED" # Must be UNCOMPRESSED when format_conversion is turned on - - cloudwatch_logging_options { - enabled = false - } - - processing_configuration { - enabled = true - - processors { - type = "Lambda" - - parameters { - parameter_name = "LambdaArn" - parameter_value = "${resource.aws_lambda_function.format_dpc_logs.arn}:$LATEST" - } - } - } - } - - server_side_encryption { - enabled = true - key_type = "AWS_OWNED_CMK" - } -} - -resource "aws_glue_catalog_database" "agg" { - name = "${local.stack_prefix}-db" - description = "DPC Aggregation Insights database" -} - -resource "aws_glue_security_configuration" "main" { - name = "${local.stack_prefix}-db-security" - - encryption_configuration { - cloudwatch_encryption { - cloudwatch_encryption_mode = "DISABLED" - } - - job_bookmarks_encryption { - job_bookmarks_encryption_mode = "DISABLED" - } - - s3_encryption { - kms_key_arn = local.dpc_glue_bucket_key_arn - s3_encryption_mode = "SSE-KMS" - } - } -} - -# CloudWatch Log Subscription -resource "aws_cloudwatch_log_subscription_filter" "quicksight-cloudwatch-agg-log-subscription" { - name = "${local.stack_prefix}-agg-subscription" - # Set the log group name so that if we use an environment ending in "-dev", it will get logs from - # the "real" log group for that environment. So we could make an environment "prod-sbx-dev" that - # we can use for development, and it will read from the "prod-sbx" environment. - log_group_name = "/aws/ecs/fargate/dpc-${local.this_env}-aggregation" - filter_pattern = "" - destination_arn = aws_kinesis_firehose_delivery_stream.ingester_agg.arn - role_arn = aws_iam_role.iam-role-cloudwatch-logs.arn -} diff --git a/terraform/services/quicksight/firehose-api.tf b/terraform/services/quicksight/firehose-api.tf deleted file mode 100644 index 6fa7fbd6c..000000000 --- a/terraform/services/quicksight/firehose-api.tf +++ /dev/null @@ -1,57 +0,0 @@ -# Firehose Data Stream -resource "aws_kinesis_firehose_delivery_stream" "ingester_api" { - depends_on = [aws_glue_catalog_table.api_metric_table] - name = "${local.stack_prefix}-ingester_api" - destination = "extended_s3" - - extended_s3_configuration { - bucket_arn = local.dpc_glue_bucket_arn - buffering_interval = 300 - buffering_size = 128 - error_output_prefix = "databases/${local.api_profile}/filter_errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/" - kms_key_arn = local.dpc_glue_bucket_key_arn - prefix = "databases/${local.api_profile}/metric_table/year=!{timestamp:yyyy}/month=!{timestamp:MM}/" - role_arn = aws_iam_role.iam-role-firehose.arn - s3_backup_mode = "Disabled" - compression_format = "UNCOMPRESSED" # Must be UNCOMPRESSED when format_conversion is turned on - - cloudwatch_logging_options { - enabled = false - } - - processing_configuration { - enabled = true - - processors { - type = "Lambda" - - parameters { - parameter_name = "LambdaArn" - parameter_value = "${resource.aws_lambda_function.format_dpc_logs.arn}:$LATEST" - } - } - } - } - - server_side_encryption { - enabled = true - key_type = "AWS_OWNED_CMK" - } -} - -resource "aws_glue_catalog_database" "api" { - name = "${local.stack_prefix}-db-api" - description = "DPC API Insights database" -} - -# CloudWatch Log Subscription -resource "aws_cloudwatch_log_subscription_filter" "quicksight-cloudwatch-api-log-subscription" { - name = "${local.stack_prefix}-api-subscription" - # Set the log group name so that if we use an environment ending in "-dev", it will get logs from - # the "real" log group for that environment. So we could make an environment "prod-sbx-dev" that - # we can use for development, and it will read from the "prod-sbx" environment. - log_group_name = "/aws/ecs/fargate/dpc-${local.this_env}-api" - filter_pattern = "" - destination_arn = aws_kinesis_firehose_delivery_stream.ingester_api.arn - role_arn = aws_iam_role.iam-role-cloudwatch-logs.arn -} diff --git a/terraform/services/quicksight/glue.tf b/terraform/services/quicksight/glue.tf deleted file mode 100644 index 5640b12a0..000000000 --- a/terraform/services/quicksight/glue.tf +++ /dev/null @@ -1,272 +0,0 @@ -locals { - - serde_format = "parquet" - - table_parameters = { - json = { - EXTERNAL = "TRUE" - }, - parquet = { - classification = "parquet" - EXTERNAL = "TRUE" - "parquet.compression" = "SNAPPY" - } - } - - table_partitions = [ - { - name = "year" - type = "string" - comment = "Year of request" - }, - { - name = "month" - type = "string" - comment = "Month of request" - } - ] - - storage_options = { - json = { - input_format = "org.apache.hadoop.mapred.TextInputFormat" - output_format = "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat" - }, - parquet = { - input_format = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat" - output_format = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat" - } - } - - serde_options = { - json = { - library = "org.apache.hive.hcatalog.data.JsonSerDe" - params = { - "ignore.malformed.json" = true, - "dots.in.keys" = true, - "timestamp.formats" = "yyyy-MM-dd'T'HH:mm:ss.SSSSSS,yyyy-MM-dd'T'HH:mm:ss.SSS,yyyy-MM-dd'T'HH:mm:ss,yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z',yyyy-MM-dd'T'HH:mm:ss.SSS'Z',yyyy-MM-dd'T'HH:mm:ss'Z'" - } - }, - grok = { - library = "com.amazonaws.glue.serde.GrokSerDe" - params = { - "ignore.malformed.json" = true, - "dots.in.keys" = true, - "timestamp.formats" = "yyyy-MM-dd'T'HH:mm:ss.SSSSSS,yyyy-MM-dd'T'HH:mm:ss.SSS,yyyy-MM-dd'T'HH:mm:ss,yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z',yyyy-MM-dd'T'HH:mm:ss.SSS'Z',yyyy-MM-dd'T'HH:mm:ss'Z'" - } - }, - parquet = { - library = "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe" - params = { - "serialization.format" = 1 - } - } - } - - agg_columns = [ - { - "name" = "message", - "type" = "string", - "comment" = "flattened JSON log item" - } - ] - - api_columns = [ - { - "name" = "message", - "type" = "string", - "comment" = "flattened JSON log item" - } - ] - -} - -resource "aws_glue_catalog_table" "agg_metric_table" { - name = local.agg_profile - database_name = aws_glue_catalog_database.agg.name - description = "CW Table for DPC Aggregation" - table_type = "EXTERNAL_TABLE" - owner = "dpc" - - parameters = local.table_parameters["parquet"] - - dynamic "partition_keys" { - for_each = local.table_partitions - - content { - name = partition_keys.value.name - type = partition_keys.value.type - comment = partition_keys.value.comment - } - } - - storage_descriptor { - location = "s3://${local.dpc_glue_s3_name}/databases/${local.agg_profile}/metric_table" - input_format = local.storage_options["json"].input_format - output_format = local.storage_options["parquet"].output_format - compressed = true - - dynamic "columns" { - for_each = local.agg_columns - - content { - name = columns.value.name - type = columns.value.type - comment = columns.value.comment - } - } - - ser_de_info { - name = local.agg_profile - serialization_library = local.serde_options[local.serde_format].library - parameters = local.serde_options[local.serde_format].params - } - } - - # These things get changed by the Crawler (if there is one), and we don't - # need to undo whatever changes the Crawler makes - lifecycle { - ignore_changes = [ - # TODO: Consider removing everything here - parameters - ] - } -} - -# add crawler for metadata inspection -resource "aws_glue_crawler" "agg_metrics" { - classifiers = [] - database_name = aws_glue_catalog_database.agg.name - configuration = jsonencode( - { - CrawlerOutput = { - Partitions = { - AddOrUpdateBehavior = "InheritFromTable" - } - } - Grouping = { - TableGroupingPolicy = "CombineCompatibleSchemas" - } - Version = 1 - } - ) - name = local.agg_profile - role = aws_iam_role.iam-role-glue.arn - - catalog_target { - database_name = aws_glue_catalog_database.agg.name - tables = [ - aws_glue_catalog_table.agg_metric_table.name, - ] - } - - lineage_configuration { - crawler_lineage_settings = "DISABLE" - } - - recrawl_policy { - recrawl_behavior = "CRAWL_EVERYTHING" - } - - schema_change_policy { - delete_behavior = "LOG" - update_behavior = "LOG" - } - - depends_on = [aws_glue_catalog_table.agg_metric_table] -} - -resource "aws_glue_catalog_table" "api_metric_table" { - - name = local.api_profile - database_name = aws_glue_catalog_database.api.name - description = "CW Table for DPC API" - table_type = "EXTERNAL_TABLE" - owner = "dpc" - - parameters = local.table_parameters["parquet"] - - dynamic "partition_keys" { - for_each = local.table_partitions - - content { - name = partition_keys.value.name - type = partition_keys.value.type - comment = partition_keys.value.comment - } - } - - storage_descriptor { - location = "s3://${local.dpc_glue_s3_name}/databases/${local.api_profile}/metric_table" - input_format = local.storage_options["json"].input_format - output_format = local.storage_options["parquet"].output_format - compressed = true - - dynamic "columns" { - for_each = local.api_columns - - content { - name = columns.value.name - type = columns.value.type - comment = columns.value.comment - } - } - - ser_de_info { - name = local.api_profile - serialization_library = local.serde_options[local.serde_format].library - parameters = local.serde_options[local.serde_format].params - } - } - - # These things get changed by the Crawler (if there is one), and we don't - # need to undo whatever changes the Crawler makes - lifecycle { - ignore_changes = [ - # TODO: Consider removing everything here - parameters - ] - } -} - -# add crawler for metadata inspection -resource "aws_glue_crawler" "api_metrics" { - classifiers = [] - database_name = aws_glue_catalog_database.api.name - configuration = jsonencode( - { - CrawlerOutput = { - Partitions = { - AddOrUpdateBehavior = "InheritFromTable" - } - } - Grouping = { - TableGroupingPolicy = "CombineCompatibleSchemas" - } - Version = 1 - } - ) - name = local.api_profile - role = aws_iam_role.iam-role-glue.arn - - catalog_target { - database_name = aws_glue_catalog_database.api.name - tables = [ - aws_glue_catalog_table.api_metric_table.name, - ] - } - - lineage_configuration { - crawler_lineage_settings = "DISABLE" - } - - recrawl_policy { - recrawl_behavior = "CRAWL_EVERYTHING" - } - - schema_change_policy { - delete_behavior = "LOG" - update_behavior = "LOG" - } - - depends_on = [aws_glue_catalog_table.api_metric_table] -} diff --git a/terraform/services/quicksight/iam.tf b/terraform/services/quicksight/iam.tf deleted file mode 100644 index 47eb70695..000000000 --- a/terraform/services/quicksight/iam.tf +++ /dev/null @@ -1,680 +0,0 @@ -## quicksight-related policies and roles -resource "aws_iam_group" "main" { - name = local.stack_prefix - path = "/delegatedadmin/developer/" -} - -resource "aws_iam_policy" "full" { - name = "dpc-insights-full-${var.env}" - path = "/delegatedadmin/developer/" - description = "Allow full access and use of the ${local.stack_prefix} bucket for this account" - policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Sid = "S3Perms" - Effect = "Allow", - Action = [ - "s3:PutAnalyticsConfiguration", - "s3:GetObjectVersionTagging", - "s3:CreateBucket", - "s3:ReplicateObject", - "s3:GetObjectAcl", - "s3:GetBucketObjectLockConfiguration", - "s3:DeleteBucketWebsite", - "s3:PutLifecycleConfiguration", - "s3:GetObjectVersionAcl", - "s3:PutBucketAcl", - "s3:PutObjectTagging", - "s3:HeadBucket", - "s3:DeleteObject", - "s3:DeleteObjectTagging", - "s3:GetBucketPolicyStatus", - "s3:PutAccountPublicAccessBlock", - "s3:GetObjectRetention", - "s3:GetBucketWebsite", - "s3:PutReplicationConfiguration", - "s3:DeleteObjectVersionTagging", - "s3:PutObjectLegalHold", - "s3:GetObjectLegalHold", - "s3:GetBucketNotification", - "s3:PutBucketCORS", - "s3:DeleteBucketPolicy", - "s3:GetReplicationConfiguration", - "s3:ListMultipartUploadParts", - "s3:PutObject", - "s3:GetObject", - "s3:PutBucketNotification", - "s3:PutBucketLogging", - "s3:PutObjectVersionAcl", - "s3:GetAnalyticsConfiguration", - "s3:PutBucketObjectLockConfiguration", - "s3:GetObjectVersionForReplication", - "s3:GetLifecycleConfiguration", - "s3:GetInventoryConfiguration", - "s3:GetBucketTagging", - "s3:PutAccelerateConfiguration", - "s3:DeleteObjectVersion", - "s3:GetBucketLogging", - "s3:ListBucketVersions", - "s3:ReplicateTags", - "s3:RestoreObject", - "s3:ListBucket", - "s3:GetAccelerateConfiguration", - "s3:GetBucketPolicy", - "s3:PutEncryptionConfiguration", - "s3:GetEncryptionConfiguration", - "s3:GetObjectVersionTorrent", - "s3:AbortMultipartUpload", - "s3:PutBucketTagging", - "s3:GetBucketRequestPayment", - "s3:GetObjectTagging", - "s3:GetMetricsConfiguration", - "s3:DeleteBucket", - "s3:PutBucketVersioning", - "s3:PutObjectAcl", - "s3:GetBucketPublicAccessBlock", - "s3:ListBucketMultipartUploads", - "s3:PutBucketPublicAccessBlock", - "s3:ListAccessPoints", - "s3:PutMetricsConfiguration", - "s3:PutObjectVersionTagging", - "s3:GetBucketVersioning", - "s3:GetBucketAcl", - "s3:BypassGovernanceRetention", - "s3:PutInventoryConfiguration", - "s3:GetObjectTorrent", - "s3:ObjectOwnerOverrideToBucketOwner", - "s3:GetAccountPublicAccessBlock", - "s3:PutBucketWebsite", - "s3:ListAllMyBuckets", - "s3:PutBucketRequestPayment", - "s3:PutObjectRetention", - "s3:GetBucketCORS", - "s3:PutBucketPolicy", - "s3:GetBucketLocation", - "s3:ReplicateDelete", - "s3:GetObjectVersion" - ] - Resource = [ - "${local.dpc_glue_bucket_arn}/*", - "${local.dpc_glue_bucket_arn}" - ] - }, - { - Sid = "CMK" - Effect = "Allow" - Action = [ - "kms:Encrypt", - "kms:Decrypt", - "kms:ReEncrypt*", - "kms:GenerateDataKey*", - "kms:DescribeKey" - ] - Resource = local.dpc_glue_bucket_key_arn - } - ] - }) -} - -resource "aws_iam_group_policy_attachment" "full_attach" { - group = aws_iam_group.main.id - policy_arn = aws_iam_policy.full.arn -} - -# Allows reads of inputs -resource "aws_iam_policy" "athena_query_source" { - name = "dpc-insights-athena-query-src-${var.env}" - path = "/delegatedadmin/developer/" - description = "Rights needed for source athena query access" - policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Sid = "s3QueryPolicy" - Effect = "Allow" - Action = [ - "s3:GetBucketLocation", - "s3:GetObject", - "s3:ListBucket", - "s3:ListBucketMultipartUploads", - "s3:ListMultipartUploadParts", - "s3:AbortMultipartUpload", - "s3:CreateBucket", - "s3:PutObject", - "s3:ListAllMyBuckets", - "athena:ListDataCatalogs", - "athena:GetDataCatalog", - "quicksight:*" - ] - Resource = [ - "arn:aws:s3:::aws-athena-query-results-*", - "${local.dpc_glue_bucket_arn}", - "${local.dpc_glue_bucket_arn}/*" - ] - }, - { - Sid = "CMK" - Effect = "Allow" - Action = [ - "kms:Encrypt", - "kms:Decrypt", - "kms:ReEncrypt*", - "kms:GenerateDataKey*", - "kms:DescribeKey" - ] - Resource = local.dpc_glue_bucket_key_arn - } - ] - }) -} - -resource "aws_iam_policy" "athena_glue_access" { - name = "dpc-insights-athena-glue-access-${var.env}" - path = "/delegatedadmin/developer/" - description = "Permissions needed for Athena to access Glue databases" - policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Sid = "DatabasePermissions" - Effect = "Allow" - Action = [ - "glue:GetDatabase", - "glue:GetDatabases", - "glue:CreateDatabase" - ] - Resource = [ - "arn:aws:glue:us-east-1:${local.account_id}:catalog", - "${aws_glue_catalog_database.agg.arn}", - "${aws_glue_catalog_database.api.arn}" - ] - }, - { - Sid = "TablePermissions" - Effect = "Allow" - Action = [ - "glue:GetDatabase", - "glue:GetTables" - ] - Resource = [ - "${aws_glue_catalog_database.agg.arn}", - "${aws_glue_catalog_database.api.arn}", - "arn:aws:glue::${local.account_id}:table/${aws_glue_catalog_database.agg.name}/*", - "arn:aws:glue::${local.account_id}:table/${aws_glue_catalog_database.api.name}/*" - ] - } - ] - }) -} - -resource "aws_iam_policy" "athena_query_results" { - name = "dpc-insights-athena-query-results-${var.env}" - path = "/delegatedadmin/developer/" - description = "Rights needed for results athena query access" - policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Sid = "s3QueryResultPolicy" - Effect = "Allow" - Action = [ - "s3:GetBucketLocation", - "s3:GetObject", - "s3:ListBucket", - "s3:ListBucketMultipartUploads", - "s3:ListMultipartUploadParts", - "s3:AbortMultipartUpload", - "s3:CreateBucket", - "s3:PutObject", - "s3:ListAllMyBuckets", - "athena:ListDataCatalogs", - "athena:GetDataCatalog", - "quicksight:*" - ] - Resource = [ - "arn:aws:s3:::aws-athena-query-results-*", - "${local.dpc_athena_bucket_arn}", - "${local.dpc_athena_bucket_arn}/*" - ] - }, - { - Sid = "CMK" - Effect = "Allow" - Action = [ - "kms:Encrypt", - "kms:Decrypt", - "kms:ReEncrypt*", - "kms:GenerateDataKey*", - "kms:DescribeKey" - ] - Resource = local.dpc_athena_bucket_key_arn - } - ] - }) -} - -resource "aws_iam_group_policy_attachment" "athena_query_attach" { - group = aws_iam_group.main.id - policy_arn = aws_iam_policy.athena_query_source.arn -} - -resource "aws_iam_group_policy_attachment" "athena_catalog_attach" { - group = aws_iam_group.main.id - policy_arn = aws_iam_policy.athena_glue_access.arn -} - -resource "aws_iam_group_policy_attachment" "athena_results_attach" { - group = aws_iam_group.main.id - policy_arn = aws_iam_policy.athena_query_results.arn -} - -resource "aws_iam_group_policy_attachment" "athena_full_attach" { - group = aws_iam_group.main.id - policy_arn = aws_iam_policy.full.arn -} - -# CloudWatch Role -resource "aws_iam_role" "iam-role-cloudwatch-logs" { - name = "${local.stack_prefix}-cloudwatch-logs-role" - description = "Allows access to the DPC Insights Firehose Delivery Stream and Export to S3" - path = "/delegatedadmin/developer/" - - permissions_boundary = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:policy/cms-cloud-admin/developer-boundary-policy" - - assume_role_policy = jsonencode( - { - Statement = [ - { - Action = "sts:AssumeRole" - Effect = "Allow" - Principal = { - Service = "logs.${data.aws_region.current.name}.amazonaws.com" - } - Sid = "" - }, - ] - Version = "2012-10-17" - } - ) -} - -resource "aws_iam_policy" "cwlogs-firehose" { - name = "${local.stack_prefix}-cloudwatch-logs-policy" - path = "/delegatedadmin/developer/" - description = "Rights needed for CW and Firehose" - - policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Action = "firehose:*" - Effect = "Allow" - Resource = [ - aws_kinesis_firehose_delivery_stream.ingester_agg.arn, - aws_kinesis_firehose_delivery_stream.ingester_api.arn - ] - } - ] - }) -} - -resource "aws_iam_role_policy_attachment" "cwlogs-firehose-attach" { - role = aws_iam_role.iam-role-cloudwatch-logs.id - policy_arn = aws_iam_policy.cwlogs-firehose.arn -} - -# Firehose Policy -resource "aws_iam_policy" "iam-policy-firehose" { - description = "Allow firehose delivery to DPC insights S3 bucket" - name = "${local.stack_prefix}-firehose-to-s3-policy" - path = "/delegatedadmin/developer/" - - policy = jsonencode( - { - Statement = [ - { - Action = [ - "glue:GetTable", - "glue:GetTableVersion", - "glue:GetTableVersions", - ] - Effect = "Allow" - Resource = [ - "arn:aws:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:table/${aws_glue_catalog_database.agg.name}/${aws_glue_catalog_table.agg_metric_table.name}", - "arn:aws:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:database/${aws_glue_catalog_database.agg.name}", - "arn:aws:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:table/${aws_glue_catalog_database.api.name}/${aws_glue_catalog_table.api_metric_table.name}", - "arn:aws:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:database/${aws_glue_catalog_database.api.name}", - "arn:aws:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:catalog" - ] - Sid = "GetGlueTable" - }, - { - Action = [ - "s3:AbortMultipartUpload", - "s3:GetBucketLocation", - "s3:GetObject", - "s3:ListBucket", - "s3:ListBucketMultipartUploads", - "s3:PutObject", - ] - Effect = "Allow" - Resource = [ - "${local.dpc_glue_bucket_arn}", - "${local.dpc_glue_bucket_arn}/*", - ] - Sid = "GetS3Bucket" - }, - { - Action = [ - "kms:Encrypt", - "kms:Decrypt", - "kms:ReEncrypt*", - "kms:GenerateDataKey*", - "kms:DescribeKey", - ] - Effect = "Allow" - Resource = local.dpc_glue_bucket_key_arn - Sid = "UseKMSKey" - }, - { - Action = [ - "logs:PutLogEvents", - ] - Effect = "Allow" - Resource = [ - "arn:aws:logs:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:log-group:/aws/kinesisfirehose/${local.agg_profile}-firehose:log-stream:*", - "arn:aws:logs:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:log-group:/aws/kinesisfirehose/${local.api_profile}-firehose:log-stream:*", - ] - Sid = "PutLogEvents" - } - ] - Version = "2012-10-17" - } - ) -} - -resource "aws_iam_role_policy_attachment" "iam-policy-firehose" { - role = aws_iam_role.iam-role-cloudwatch-logs.id - policy_arn = aws_iam_policy.iam-policy-firehose.arn -} - -# Firehose Role -resource "aws_iam_role" "iam-role-firehose" { - name = "${local.stack_prefix}-firehose-role" - description = "allows Firehose access to Lambda transformation" - path = "/delegatedadmin/developer/" - - permissions_boundary = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:policy/cms-cloud-admin/developer-boundary-policy" - - force_detach_policies = false - - max_session_duration = 3600 - assume_role_policy = jsonencode( - { - Statement = [ - { - Action = "sts:AssumeRole" - Effect = "Allow" - Principal = { - Service = "firehose.amazonaws.com" - } - Sid = "FirehoseAssume" - }, - ] - Version = "2012-10-17" - } - ) -} - -resource "aws_iam_role_policy_attachment" "iam-policy-firehose-role" { - role = aws_iam_role.iam-role-firehose.id - policy_arn = aws_iam_policy.iam-policy-firehose.arn -} - -resource "aws_iam_role_policy_attachment" "role-firehose-attach" { - role = aws_iam_role.iam-role-firehose.id - policy_arn = aws_iam_policy.cwlogs-firehose.arn -} - -resource "aws_iam_policy" "iam-policy-lambda-firehose" { - description = "Allow firehose lambda execution" - name = "invoke-${aws_lambda_function.format_dpc_logs.function_name}" - path = "/delegatedadmin/developer/" - - policy = jsonencode( - { - Statement = [ - { - Action = "lambda:InvokeFunction" - Effect = "Allow" - Resource = [ - "arn:aws:lambda:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:function:${aws_lambda_function.format_dpc_logs.function_name}", - "arn:aws:lambda:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:function:${aws_lambda_function.format_dpc_logs.function_name}:$LATEST", - ] - Sid = "InvokeCW2Json" - }, - ] - Version = "2012-10-17" - } - ) -} - -resource "aws_iam_role_policy_attachment" "iam-policy-invoke-lambda-firehose" { - role = aws_iam_role.iam-role-firehose.id - policy_arn = aws_iam_policy.iam-policy-lambda-firehose.arn -} - - -# Lambda Role -resource "aws_iam_role" "iam-role-firehose-lambda" { - name = "${local.stack_prefix}-firehose-lambda-role" - description = "Allow Lambda to create and write to its log group" - path = "/delegatedadmin/developer/" - - permissions_boundary = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:policy/cms-cloud-admin/developer-boundary-policy" - - max_session_duration = 3600 - force_detach_policies = false - assume_role_policy = jsonencode( - { - Statement = [ - { - Action = "sts:AssumeRole" - Effect = "Allow" - Principal = { - Service = "lambda.amazonaws.com" - } - }, - ] - Version = "2012-10-17" - } - ) -} - -resource "aws_iam_policy" "iam-policy-lambda-firehose-logging" { - description = "Allow firehose lambda execution logging" - name = "${local.stack_prefix}-lambda-logging-policy" - path = "/delegatedadmin/developer/" - - policy = jsonencode({ - Version = "2012-10-17" - Statement = [ - { - Effect = "Allow" - Action = "logs:CreateLogGroup" - Resource = "arn:aws:logs:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:*" - }, - { - Effect = "Allow" - Action = [ - "logs:CreateLogStream", - "logs:PutLogEvents" - ] - Resource = [ - "arn:aws:logs:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:log-group:/aws/lambda/${local.agg_profile}-cw-to-flattened-json:*", - "arn:aws:logs:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:log-group:/aws/lambda/${local.api_profile}-cw-to-flattened-json:*" - ] - }, - { - Effect = "Allow" - Action = [ - "firehose:PutRecordBatch", - "firehose:PutRecord" - ] - Resource = [ - "arn:aws:firehose:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:deliverystream/${local.agg_profile}-ingester_agg", - "arn:aws:firehose:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:deliverystream/${local.api_profile}-ingester_api" - ] - } - ] - }) -} - -resource "aws_iam_role_policy_attachment" "iam-policy-invoke-lambda-firehose-logging" { - role = aws_iam_role.iam-role-firehose-lambda.id - policy_arn = aws_iam_policy.iam-policy-lambda-firehose-logging.arn -} - -# Glue role for Crawler -resource "aws_iam_role" "iam-role-glue" { - name = "${local.stack_prefix}-glue-role" - description = "allows Glue access to S3 database" - path = "/delegatedadmin/developer/" - - permissions_boundary = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:policy/cms-cloud-admin/developer-boundary-policy" - - force_detach_policies = false - - max_session_duration = 3600 - assume_role_policy = jsonencode( - { - Statement = [ - - { - Action = "sts:AssumeRole" - Effect = "Allow" - Principal = { - Service = "glue.amazonaws.com" - } - Sid = "GlueAssume" - }, - ] - Version = "2012-10-17" - } - ) -} - -resource "aws_iam_policy" "iam-policy-glue-crawler" { - description = "Allow glue crawler execution" - name = "${local.stack_prefix}-glue-crawler-policy" - path = "/delegatedadmin/developer/" - - policy = jsonencode({ - - Statement = [ - { - Action = [ - "s3:ListBucket", - "s3:HeadBucket", - "s3:GetObject*", - "s3:GetBucketLocation" - ] - Effect = "Allow" - Resource = [ - "arn:aws:s3:::awsglue-datasets/*", - "arn:aws:s3:::awsglue-datasets" - ] - Sid = "GlueList" - }, - { - Action = [ - "s3:ListBucketMultipartUploads", - "s3:ListBucket", - "s3:HeadBucket", - "s3:GetBucketLocation" - ] - Effect = "Allow" - Resource = [ - "${local.dpc_glue_bucket_arn}" - ] - Sid = "s3Buckets" - }, - { - Action = [ - "s3:PutObject*", - "s3:ListMultipartUploadParts", - "s3:GetObject*", - "s3:DeleteObject*", - "s3:AbortMultipartUpload" - ] - Effect = "Allow" - Resource = [ - "${local.dpc_glue_bucket_arn}/*" - ] - Sid = "s3Objects" - }, - { - Action = [ - "kms:ReEncrypt*", - "kms:GenerateDataKey*", - "kms:Encrypt", - "kms:DescribeKey", - "kms:Decrypt" - ] - Effect = "Allow" - Resource = local.dpc_glue_bucket_key_arn - Sid = "CMK" - } - ] - Version = "2012-10-17" - - }) -} - -resource "aws_iam_role_policy_attachment" "iam-policy-glue-crawler" { - role = aws_iam_role.iam-role-glue.id - policy_arn = aws_iam_policy.iam-policy-glue-crawler.arn -} - -data "aws_iam_policy" "aws_glue_service_role" { - arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole" -} - -resource "aws_iam_role_policy_attachment" "iam-policy-glue-service" { - role = aws_iam_role.iam-role-glue.id - policy_arn = data.aws_iam_policy.aws_glue_service_role.arn -} - -data "aws_iam_policy" "aws_athena_full_policy" { - arn = "arn:aws:iam::aws:policy/AmazonAthenaFullAccess" -} - -resource "aws_iam_role_policy_attachment" "iam-policy-athena-service" { - role = aws_iam_role.iam-role-glue.id - policy_arn = data.aws_iam_policy.aws_athena_full_policy.arn -} - -data "aws_kms_alias" "glue_s3_key" { - name = local.dpc_glue_bucket_key_alias -} - -# Grant to QuickSight access to the Glue bucket key -resource "aws_kms_grant" "glue" { - name = "${local.dpc_glue_s3_name}-grant" - key_id = data.aws_kms_alias.glue_s3_key.target_key_id - grantee_principal = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:role/service-role/aws-quicksight-service-role-v0" - operations = ["Encrypt", "Decrypt", "GenerateDataKey", "DescribeKey", "ReEncryptTo", "ReEncryptFrom"] -} - -data "aws_kms_alias" "athena_s3_key" { - name = local.dpc_athena_bucket_key_alias -} - -# Grant to QuickSight access to the Athena bucket key -resource "aws_kms_grant" "athena" { - name = "${local.dpc_athena_s3_name}-grant" - key_id = data.aws_kms_alias.athena_s3_key.target_key_id - grantee_principal = "arn:aws:iam::${data.aws_caller_identity.current.account_id}:role/service-role/aws-quicksight-service-role-v0" - operations = ["Encrypt", "Decrypt", "GenerateDataKey", "DescribeKey", "ReEncryptTo", "ReEncryptFrom"] -} diff --git a/terraform/services/quicksight/lambda.tf b/terraform/services/quicksight/lambda.tf deleted file mode 100644 index 7ad5ab081..000000000 --- a/terraform/services/quicksight/lambda.tf +++ /dev/null @@ -1,38 +0,0 @@ -data "archive_file" "zip-archive-format-dpc-logs" { - type = "zip" - source_file = "${path.module}/lambda_src/dpc-bfd-cwlog-basic-flatten-json.py" - output_path = "${path.module}/lambda_src/${local.this_env}/dpc-bfd-cwlog-basic-flatten-json.zip" -} - -# Lambda Function to process logs from Firehose -resource "aws_lambda_function" "format_dpc_logs" { - architectures = [ - "x86_64", - ] - description = "Extracts and flattens JSON messages from CloudWatch log subscriptions" - function_name = "${local.stack_prefix}-cw-to-flattened-json" - filename = data.archive_file.zip-archive-format-dpc-logs.output_path - handler = "dpc-bfd-cwlog-basic-flatten-json.lambda_handler" - layers = [] - memory_size = 256 - package_type = "Zip" - reserved_concurrent_executions = -1 - role = aws_iam_role.iam-role-firehose-lambda.arn - - runtime = "python3.12" - source_code_hash = data.archive_file.zip-archive-format-dpc-logs.output_base64sha256 - - tags = { "lambda-console:blueprint" = "kinesis-firehose-cloudwatch-logs-processor-python" } - - timeout = 300 - - ephemeral_storage { - size = 512 - } - - timeouts {} - - tracing_config { - mode = "PassThrough" - } -} diff --git a/terraform/services/quicksight/lambda_src/dpc-bfd-cwlog-basic-flatten-json.py b/terraform/services/quicksight/lambda_src/dpc-bfd-cwlog-basic-flatten-json.py deleted file mode 100644 index 9493e9f68..000000000 --- a/terraform/services/quicksight/lambda_src/dpc-bfd-cwlog-basic-flatten-json.py +++ /dev/null @@ -1,386 +0,0 @@ -# Copyright 2014, Amazon.com, Inc. or its affiliates. All Rights Reserved. -# -# Licensed under the Amazon Software License (the "License"). -# You may not use this file except in compliance with the License. -# A copy of the License is located at -# -# http://aws.amazon.com/asl/ -# -# or in the "license" file accompanying this file. This file is distributed -# on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either -# express or implied. See the License for the specific language governing -# permissions and limitations under the License. - -""" -For processing data sent to Firehose by Cloudwatch Logs subscription filters. - -Cloudwatch Logs sends to Firehose records that look like this: - -{ - "messageType": "DATA_MESSAGE", - "owner": "123456789012", - "logGroup": "log_group_name", - "logStream": "log_stream_name", - "subscriptionFilters": [ - "subscription_filter_name" - ], - "logEvents": [ - { - "id": "01234567890123456789012345678901234567890123456789012345", - "timestamp": 1510109208016, - "message": "log message 1" - }, - { - "id": "01234567890123456789012345678901234567890123456789012345", - "timestamp": 1510109208017, - "message": "log message 2" - } - ... - ] -} - -The data is additionally compressed with GZIP. - -The code below will: - -1) Gunzip the data -2) Parse the json -3) Set the result to ProcessingFailed for any record whose messageType is not DATA_MESSAGE, thus - redirecting them to the processing error output. Such records do not contain any log events. You - can modify the code to set the result to Dropped instead to get rid of these records completely. -4) For records whose messageType is DATA_MESSAGE, extract the individual log events from the - logEvents field, and pass each one to the transformLogEvent method. You can modify the - transformLogEvent method to perform custom transformations on the log events. -5) Concatenate the result from (4) together and set the result as the data of the record returned to - Firehose. Note that this step will not add any delimiters. Delimiters should be appended by the - logic within the transformLogEvent method. -6) Any individual record exceeding 6,000,000 bytes in size after decompression, processing and - base64-encoding is marked as Dropped, and the original record is split into two and re-ingested - back into Firehose or Kinesis. The re-ingested records should be about half the size compared to - the original, and should fit within the size limit the second time round. -7) When the total data size (i.e. the sum over multiple records) after decompression, processing and - base64-encoding exceeds 6,000,000 bytes, any additional records are re-ingested back into - Firehose or Kinesis. -8) The retry count for intermittent failures during re-ingestion is set 20 attempts. If you wish to - retry fewer number of times for intermittent failures you can lower this value. - -Except for type hints/annotations and code noted as a 'BFD modification' all code here is original -to the AWS Blueprint and should not be changed without accepting the maintenance burden. -""" - -import base64 -import gzip -import json -from datetime import datetime # BFD modification -from typing import Any, Generator - -import boto3 - - -def format_json(y: dict[str, Any]) -> dict[str, Any]: - """ - BFD modification to add a function to flatten a JSON object (but not pivot out arrays), - convert field names to lower case, and replace "." with "_" in field names. - Code credits: https://towardsdatascience.com/flattening-json-objects-in-python-f5343c794b10 - """ - - out: dict[str, Any] = {} - - def flatten(x: dict[str, Any] | str, name: str = ""): - if isinstance(x, dict): - for a in x: - flatten(x[a], name + a.replace(".", "_") + "_") - else: - out[name.lower()[:-1]] = x - - flatten(y) - return out - - -def transformLogEvent(log_event: dict[str, Any]) -> str | None: - """Transform each log event. - - The default implementation below just extracts the message and appends a newline to it. - - Args: - log_event (dict): The original log event. - Structure is {"id": str, "timestamp": long, "message": str} - - Returns: - str: The transformed log event. - """ - - # BFD modification to the blueprint to format the message json uniformly. - try: - log_event_json = json.loads(log_event["message"]) - except json.JSONDecodeError as exc: - print( - f'Unable to transform log ID {log_event["id"]} due to JSON error "{str(exc)}" decoding' - f" message: {log_event['message']}" - ) - return None - - flattened_log_event_json = format_json(log_event_json) - flattened_log_event_json["cw_id"] = log_event["id"] - # Ignoring that utcfromtimestamp is deprecated. Should probably be replaced with fromtimestamp, - # but without comprehensive tests I don't want to accept the possibility of unexpected changes - flattened_log_event_json["cw_timestamp"] = datetime.utcfromtimestamp( # type: ignore - log_event["timestamp"] / 1000 - ).isoformat() - - stringized_flattened_log_event_json = json.dumps(flattened_log_event_json) - return stringized_flattened_log_event_json + "\n" - - -def processRecords( - records: list[dict[str, Any]], -) -> Generator[dict[str, Any], None, None]: - for r in records: - data = loadJsonGzipBase64(r["data"]) - recId = r["recordId"] - # CONTROL_MESSAGE are sent by CWL to check if the subscription is reachable. - # They do not contain actual data. - if data["messageType"] == "CONTROL_MESSAGE": - yield {"result": "Dropped", "recordId": recId} - elif data["messageType"] == "DATA_MESSAGE": - valid_log_events = [ - x - for x in (transformLogEvent(e) for e in data["logEvents"]) - if x is not None - ] - if valid_log_events: - joinedData = "".join(valid_log_events) - dataBytes = joinedData.encode("utf-8") - encodedData = base64.b64encode(dataBytes).decode("utf-8") - - yield {"data": encodedData, "result": "Ok", "recordId": recId} - else: - print( - f"Record ID {r['recordId']} was empty after attempting to transform its log" - " events. Marking as Dropped" - ) - yield {"result": "Dropped", "recordId": recId} - else: - yield {"result": "ProcessingFailed", "recordId": recId} - - -def splitCWLRecord(cwlRecord: dict[str, Any]) -> list[bytes]: - """ - Splits one CWL record into two, each containing half the log events. - Serializes and compreses the data before returning. That data can then be - re-ingested into the stream, and it'll appear as though they came from CWL - directly. - """ - logEvents = cwlRecord["logEvents"] - mid = len(logEvents) // 2 - rec1 = {k: v for k, v in cwlRecord.items()} - rec1["logEvents"] = logEvents[:mid] - rec2 = {k: v for k, v in cwlRecord.items()} - rec2["logEvents"] = logEvents[mid:] - return [gzip.compress(json.dumps(r).encode("utf-8")) for r in [rec1, rec2]] - - -def putRecordsToFirehoseStream( - streamName: str, - records: list[dict[str, Any]], - client: Any, - attemptsMade: int, - maxAttempts: int, -): - failedRecords: list[dict[str, Any]] = [] - codes: list[str] = [] - errMsg = "" - # if put_record_batch throws for whatever reason, response['xx'] will error out, adding a check - # for a valid response will prevent this - response = None - try: - response = client.put_record_batch( - DeliveryStreamName=streamName, Records=records - ) - except Exception as e: - failedRecords = records - errMsg = str(e) - - # if there are no failedRecords (put_record_batch succeeded), iterate over the response to - # gather results - if not failedRecords and response and response["FailedPutCount"] > 0: - for idx, res in enumerate(response["RequestResponses"]): - # (if the result does not have a key 'ErrorCode' OR if it does and is empty) => we do - # not need to re-ingest - if not res.get("ErrorCode"): - continue - - codes.append(res["ErrorCode"]) - failedRecords.append(records[idx]) - - errMsg = "Individual error codes: " + ",".join(codes) - - if failedRecords: - if attemptsMade + 1 < maxAttempts: - print( - "Some records failed while calling PutRecordBatch to Firehose stream," - " retrying. %s" % (errMsg) - ) - putRecordsToFirehoseStream( - streamName, failedRecords, client, attemptsMade + 1, maxAttempts - ) - else: - raise RuntimeError( - "Could not put records after %s attempts. %s" - % (str(maxAttempts), errMsg) - ) - - -def putRecordsToKinesisStream( - streamName: str, - records: list[dict[str, Any]], - client: Any, - attemptsMade: int, - maxAttempts: int, -) -> None: - failedRecords: list[dict[str, Any]] = [] - codes: list[str] = [] - errMsg = "" - # if put_records throws for whatever reason, response['xx'] will error out, adding a check for a - # valid response will prevent this - response = None - try: - response = client.put_records(StreamName=streamName, Records=records) - except Exception as e: - failedRecords = records - errMsg = str(e) - - # if there are no failedRecords (put_record_batch succeeded), iterate over the response to - # gather results - if not failedRecords and response and response["FailedRecordCount"] > 0: - for idx, res in enumerate(response["Records"]): - # (if the result does not have a key 'ErrorCode' OR if it does and is empty) => we do - # not need to re-ingest - if not res.get("ErrorCode"): - continue - - codes.append(res["ErrorCode"]) - failedRecords.append(records[idx]) - - errMsg = "Individual error codes: " + ",".join(codes) - - if failedRecords: - if attemptsMade + 1 < maxAttempts: - print( - "Some records failed while calling PutRecords to Kinesis stream," - " retrying. %s" % (errMsg) - ) - putRecordsToKinesisStream( - streamName, failedRecords, client, attemptsMade + 1, maxAttempts - ) - else: - raise RuntimeError( - "Could not put records after %s attempts. %s" - % (str(maxAttempts), errMsg) - ) - - -def createReingestionRecord( - isSas: bool, originalRecord: dict[str, Any], data: bytes | None = None -) -> dict[str, Any]: - if data is None: - data = base64.b64decode(originalRecord["data"]) - r = {"Data": data} - if isSas: - r["PartitionKey"] = originalRecord["kinesisRecordMetadata"]["partitionKey"] - return r - - -def loadJsonGzipBase64(base64Data: bytes) -> dict[str, Any]: - return json.loads(gzip.decompress(base64.b64decode(base64Data))) - - -def lambda_handler(event: dict[str, Any], context: dict[str, Any]): - isSas = "sourceKinesisStreamArn" in event - streamARN = event["sourceKinesisStreamArn"] if isSas else event["deliveryStreamArn"] - region = streamARN.split(":")[3] - streamName = streamARN.split("/")[1] - records = list(processRecords(event["records"])) - projectedSize = 0 - recordListsToReingest: list[list[dict[str, Any]]] = [] - - for idx, rec in enumerate(records): - originalRecord = event["records"][idx] - - if rec["result"] != "Ok": - continue - - # If a single record is too large after processing, split the original CWL data into two, - # each containing half the log events, and re-ingest both of them (note that it is the - # original data that is re-ingested, not the processed data). If it's not possible to split - # because there is only one log event, then mark the record as ProcessingFailed, which sends - # it to error output. - if len(rec["data"]) > 6000000: - cwlRecord = loadJsonGzipBase64(originalRecord["data"]) - if len(cwlRecord["logEvents"]) > 1: - rec["result"] = "Dropped" - recordListsToReingest.append( - [ - createReingestionRecord(isSas, originalRecord, data) - for data in splitCWLRecord(cwlRecord) - ] - ) - else: - rec["result"] = "ProcessingFailed" - print( - ( - "Record %s contains only one log event but is still too large" - " after processing (%d bytes), " + "marking it as %s" - ) - % (rec["recordId"], len(rec["data"]), rec["result"]) - ) - del rec["data"] - else: - projectedSize += len(rec["data"]) + len(rec["recordId"]) - # 6000000 instead of 6291456 to leave ample headroom for the stuff we didn't account for - if projectedSize > 6000000: - recordListsToReingest.append( - [createReingestionRecord(isSas, originalRecord)] - ) - del rec["data"] - rec["result"] = "Dropped" - - # call putRecordBatch/putRecords for each group of up to 500 records to be re-ingested - if recordListsToReingest: - recordsReingestedSoFar = 0 - client = boto3.client("kinesis" if isSas else "firehose", region_name=region) # type: ignore - maxBatchSize = 500 - flattenedList = [r for sublist in recordListsToReingest for r in sublist] - for i in range(0, len(flattenedList), maxBatchSize): - recordBatch = flattenedList[i : i + maxBatchSize] - # last argument is maxAttempts - if isSas: - putRecordsToKinesisStream(streamName, recordBatch, client, 0, 20) - else: - putRecordsToFirehoseStream(streamName, recordBatch, client, 0, 20) - recordsReingestedSoFar += len(recordBatch) - print("Reingested %d/%d" % (recordsReingestedSoFar, len(flattenedList))) - - num_processed_ok = len([r for r in records if r["result"] == "Ok"]) - num_processed_failed = len( - [r for r in records if r["result"] == "ProcessingFailed"] - ) - num_split = len([rl for rl in recordListsToReingest if len(rl) > 1]) - num_reingested_asis = len([rl for rl in recordListsToReingest if len(rl) == 1]) - num_actual_dropped = len([r for r in records if r["result"] == "Dropped"]) - ( - num_reingested_asis + num_split - ) - print( - "%d input records, %d returned as Ok, %d returned as ProcessingFailed, %d split and" - " re-ingested, %d re-ingested as-is, %d record(s) dropped" - % ( - len(event["records"]), - num_processed_ok, - num_processed_failed, - num_split, - num_reingested_asis, - num_actual_dropped, - ) - ) - - return {"records": records} diff --git a/terraform/services/quicksight/main.tf b/terraform/services/quicksight/main.tf deleted file mode 100644 index 9550ce29f..000000000 --- a/terraform/services/quicksight/main.tf +++ /dev/null @@ -1,57 +0,0 @@ -locals { - ab2d_env_lbs = { - dev = "ab2d-dev" - test = "ab2d-east-impl" - sbx = "ab2d-sbx-sandbox" - prod = "api-ab2d-east-prod" - } - load_balancers = { - ab2d = "${local.ab2d_env_lbs[var.env]}" - bcda = "bcda-api-${var.env == "sbx" ? "opensbx" : var.env}-01" - dpc = "dpc-${var.env == "sbx" ? "prod-sbx" : var.env}-1" - } - stack_prefix = "${var.app}-${local.this_env}" - this_env = var.env == "sbx" ? "prod-sbx" : var.env - account_id = data.aws_caller_identity.current.account_id - agg_profile = "${local.stack_prefix}-aggregator" - api_profile = "${local.stack_prefix}-api" - - athena_profile = "${var.app}_${local.this_env}_insights_${local.account_id}" - athena_prefix = "${var.app}-${local.this_env}-insights" - - dpc_glue_s3_name = "${local.stack_prefix}-${local.account_id}" - dpc_logging_s3_name = "${local.stack_prefix}-logs-${local.account_id}" - dpc_athena_s3_name = local.athena_prefix - - dpc_glue_bucket_arn = module.dpc_insights_data.arn - dpc_glue_bucket_key_alias = module.dpc_insights_data.key_alias - dpc_glue_bucket_key_arn = module.dpc_insights_data.key_arn - dpc_glue_bucket_id = module.dpc_insights_data.id - dpc_athena_bucket_arn = module.dpc_insights_athena.arn - dpc_athena_bucket_key_alias = module.dpc_insights_athena.key_alias - dpc_athena_bucket_key_arn = module.dpc_insights_athena.key_arn - dpc_athena_bucket_id = module.dpc_insights_data.id - - athena_workgroup_name = local.athena_prefix - dpc_athena_results_folder_key = "workgroups/${local.athena_workgroup_name}/" -} - -data "aws_caller_identity" "current" {} - -data "aws_region" "current" {} - -module "dpc_insights_data" { - source = "../../modules/bucket" - name = local.dpc_glue_s3_name -} - -module "dpc_insights_athena" { - source = "../../modules/bucket" - name = local.dpc_athena_s3_name -} - -resource "aws_s3_object" "folder" { - bucket = module.dpc_insights_athena.id - content_type = "application/x-directory" - key = local.dpc_athena_results_folder_key -} \ No newline at end of file diff --git a/terraform/services/quicksight/sql_templates/bulk_requests.sql.tfpl b/terraform/services/quicksight/sql_templates/bulk_requests.sql.tfpl deleted file mode 100644 index e9795e2a6..000000000 --- a/terraform/services/quicksight/sql_templates/bulk_requests.sql.tfpl +++ /dev/null @@ -1,26 +0,0 @@ --- --- This view is the basis for the DPC Quicksight dashboard Component --- Bulk Calls to API --- View returns one row for each day that DPC processed bulk data requests. --- The content of the returned rows is count of calls and corresponding date --- -create or replace view bulk_data_requests_by_day as -select request_ct, date_format(msgday, '%Y-%m-%d') as msgdate -from ( - SELECT count(message) as request_ct, date_trunc('day', messagets ) as msgday - FROM ( - select message, - cast(json_extract_scalar (message, '$.contentlength') as VARCHAR) as contentlength, - cast(json_extract_scalar (message, '$.message') as VARCHAR) as payloadmessage, - from_iso8601_timestamp(cast(json_extract_scalar(message,'$.timestamp') as VARCHAR)) as messagets - FROM ${api_profile} - ) rowset - WHERE contentlength is null - and payloadmessage is not null - and cast(json_extract_scalar(message,'$.application') as VARCHAR) = ${app} - and cast(json_extract_scalar(message, '$.environment') as VARCHAR) = ${env} - and payloadmessage like 'Exporting data for provider%' - and date_diff('day', current_date, messagets) < ${days_history} - group by date_trunc('day', messagets ) -) summarydata -order by summarydata.msgday diff --git a/terraform/services/quicksight/sql_templates/everything_requests.sql.tfpl b/terraform/services/quicksight/sql_templates/everything_requests.sql.tfpl deleted file mode 100644 index 0704835ba..000000000 --- a/terraform/services/quicksight/sql_templates/everything_requests.sql.tfpl +++ /dev/null @@ -1,26 +0,0 @@ --- --- This view is the basis for the DPC Quicksight dashboard Component --- /Patient/$Everything Requests --- View returns one row for each day that DPC processed /Everything requests for the /Patient endpoint. --- The content of the returned rows is count of calls and corresponding date --- -create or replace view everything_requests_by_day as -select request_ct, date_format(msgday, '%Y-%m-%d') as msgdate -from ( - SELECT count(message) as request_ct, - date_trunc('day', messagets ) as msgday - FROM ( - select message, - cast(json_extract_scalar (message, '$.contentlength') as VARCHAR) as contentlength, - from_iso8601_timestamp(cast(json_extract_scalar(message,'$.timestamp') as VARCHAR)) as messagets - FROM ${api_profile} - ) rowset - where contentlength is not null - and cast(json_extract_scalar(message,'$.method') as VARCHAR) = 'GET' - and cast(json_extract_scalar(message,'$.useragent') as VARCHAR) != 'ELB-HealthChecker/2.0' - and cast(json_extract_scalar(message,'$.uri') as VARCHAR) like '/api/v1/Patient/%everything' - and cast(json_extract_scalar(message,'$.environment') as VARCHAR) = ${env} - and date_diff('day', current_date, messagets) < ${days_history} - group by date_trunc('day', messagets ) -) summarydata -order by summarydata.msgday diff --git a/terraform/services/quicksight/sql_templates/group_requests.sql.tfpl b/terraform/services/quicksight/sql_templates/group_requests.sql.tfpl deleted file mode 100644 index c451bfb12..000000000 --- a/terraform/services/quicksight/sql_templates/group_requests.sql.tfpl +++ /dev/null @@ -1,25 +0,0 @@ --- --- This view is the basis for the DPC Quicksight dashboard Component --- /Group Requests By Day --- View returns one row for each day that DPC processed /Group endpoint requests. --- The content of the returned rows is the count of calls and corresponding date --- -create or replace view group_requests_made_by_day as -select request_ct, date_format(msgday, '%Y-%m-%d') as msgdate -from ( - SELECT count(message) as request_ct, date_trunc('day', messagets ) as msgday - FROM ( - select message, - cast(json_extract_scalar (message, '$.contentlength') as VARCHAR) as contentlength, - from_iso8601_timestamp(cast(json_extract_scalar(message,'$.timestamp') as VARCHAR)) as messagets - FROM ${api_profile} - ) rowset - WHERE contentlength is not null - and cast(json_extract_scalar(message,'$.method') as VARCHAR) = 'GET' - and cast(json_extract_scalar(message,'$.useragent') as VARCHAR) != 'ELB-HealthChecker/2.0' - and cast(json_extract_scalar(message,'$.uri') as VARCHAR) = '/api/v1/Group' - and cast(json_extract_scalar(message,'$.environment') as VARCHAR) = ${env} - and date_diff('day', current_date, messagets) < ${days_history} - group by date_trunc('day', messagets ) -) summarydata -order by summarydata.msgday diff --git a/terraform/services/quicksight/sql_templates/since_requests.sql.tfpl b/terraform/services/quicksight/sql_templates/since_requests.sql.tfpl deleted file mode 100644 index 1c5e8d316..000000000 --- a/terraform/services/quicksight/sql_templates/since_requests.sql.tfpl +++ /dev/null @@ -1,26 +0,0 @@ --- --- This view is the basis for the DPC Quicksight dashboard Component --- "Since" Parameter Calls by Day --- View returns one row for each day that DPC processed calls that leveraged '%_since'. --- The content of the returned rows is the count of calls and corresponding date --- -create or replace view since_requests_by_day as -select since_call_ct, date_format(msgday, '%Y-%m-%d') as msgdate -from ( - SELECT count(message) as since_call_ct, date_trunc('day', messagets ) as msgday - FROM ( - select message, - cast(json_extract_scalar (message, '$.contentlength') as VARCHAR) as contentlength, - cast(json_extract_scalar (message, '$.message') as VARCHAR) as payloadmessage, - from_iso8601_timestamp(cast(json_extract_scalar(message,'$.timestamp') as VARCHAR)) as messagets - FROM ${api_profile} - ) rowset - where contentlength is null - and payloadmessage is not null - and cast(json_extract_scalar(message,'$.application') as VARCHAR) = ${app} - and cast(json_extract_scalar(message, '$.environment') as VARCHAR) = ${env} - and payloadmessage like 'Exporting data for provider%_since%' - and date_diff('day', current_date, messagets) < ${days_history} - group by date_trunc('day', messagets ) -) summarydata -order by summarydata.msgday \ No newline at end of file diff --git a/terraform/services/quicksight/sql_templates/total_benes_requested.sql.tfpl b/terraform/services/quicksight/sql_templates/total_benes_requested.sql.tfpl deleted file mode 100644 index 25b1cedba..000000000 --- a/terraform/services/quicksight/sql_templates/total_benes_requested.sql.tfpl +++ /dev/null @@ -1,30 +0,0 @@ --- --- This view is the basis for the DPC Quicksight dashboard Component --- Beneficiaries Requested by Customers --- View returns one row for each day that DPC processed request for beneficiaries. --- The content of the returned rows is count of calls and corresponding date --- -create or replace view total_benes_by_day as -select sum(summarydata.ct_by_day) as daily_sum, summarydata.msgdate -from ( - select count(distinctdata.ptid) as ct_by_day, distinctdata.ptid, msgdate - from ( - select distinct rawdata.ptid, date_format(rawdata.msgday,'%Y-%m-%d') as msgdate - from ( - SELECT cast(json_extract_scalar(message,'$.mdc_patientid') as VARCHAR) as ptid, - from_iso8601_timestamp(cast(json_extract_scalar(message,'$.timestamp') as VARCHAR)) as msgday - FROM - ( select message, cast(json_extract_scalar (message, '$.dpcmetric') as VARCHAR) as dpcmetric - FROM ${agg_profile} - ) rowset - where - dpcmetric = 'DataExportResult' - and cast(json_extract_scalar (message, '$.environment') as VARCHAR) = ${env} - ) rawdata - where date_diff('day',current_date, rawdata.msgday) < ${days_history} - ) distinctdata - group by distinctdata.ptid, distinctdata.msgdate - order by distinctdata.ptid, distinctdata.msgdate -) summarydata -group by summarydata.msgdate -order by summarydata.msgdate \ No newline at end of file diff --git a/terraform/services/quicksight/sql_templates/uniq_benes_served.sql.tfpl b/terraform/services/quicksight/sql_templates/uniq_benes_served.sql.tfpl deleted file mode 100644 index d533436c5..000000000 --- a/terraform/services/quicksight/sql_templates/uniq_benes_served.sql.tfpl +++ /dev/null @@ -1,30 +0,0 @@ --- --- This view is the basis for the DPC Quicksight dashboard Component --- Unique Beneficiaries Served --- View returns one row for each day that DPC processed a unique beneficiary request. --- The content of the returned rows is count of calls and correponding date. --- -create or replace view uniq_benes_served_by_day as -select sum(summarydata.ct_by_day) as daily_sum, summarydata.msgdate -from ( - select count(distinctdata.ptid) as ct_by_day, distinctdata.ptid, msgdate - from ( - select distinct rawdata.ptid, date_format(rawdata.msgday,'%Y-%m-%d') as msgdate - from ( - SELECT cast(json_extract_scalar(message,'$.mdc_patientid') as VARCHAR) as ptid, - from_iso8601_timestamp(cast(json_extract_scalar(message,'$.timestamp') as VARCHAR)) as msgday - FROM - ( select message, cast(json_extract_scalar (message, '$.dpcmetric') as VARCHAR) as dpcmetric - FROM ${agg_profile} - ) rowset - WHERE dpcmetric = 'DataExportResult' - and cast(json_extract_scalar (message, '$.environment') as VARCHAR) = ${env} - and cast(json_extract_scalar (message, '$.dataretrieved') as VARCHAR) = 'true' - ) rawdata - where date_diff('day',current_date, rawdata.msgday) < ${days_history} - ) distinctdata - group by distinctdata.ptid, distinctdata.msgdate - order by distinctdata.ptid, distinctdata.msgdate -) summarydata -group by summarydata.msgdate -order by summarydata.msgdate \ No newline at end of file diff --git a/terraform/services/quicksight/terraform.tf b/terraform/services/quicksight/terraform.tf deleted file mode 100644 index 5a2402a40..000000000 --- a/terraform/services/quicksight/terraform.tf +++ /dev/null @@ -1,18 +0,0 @@ -provider "aws" { - default_tags { - tags = { - application = var.app - business = "oeda" - code = "https://github.com/CMSgov/cdap/tree/main/terraform/services/quicksight" - component = "quicksight" - environment = var.env - terraform = true - } - } -} - -terraform { - backend "s3" { - key = "quicksight/terraform.tfstate" - } -} diff --git a/terraform/services/quicksight/variables.tf b/terraform/services/quicksight/variables.tf deleted file mode 100644 index 0a3dea4b1..000000000 --- a/terraform/services/quicksight/variables.tf +++ /dev/null @@ -1,17 +0,0 @@ -variable "app" { - description = "The application name (ab2d, bcda, dpc)" - type = string - validation { - condition = contains(["ab2d", "bcda", "dpc"], var.app) - error_message = "Valid value for app is ab2d, bcda, or dpc." - } -} - -variable "env" { - description = "The application environment (dev, test, sbx, prod)" - type = string - validation { - condition = contains(["dev", "test", "sbx", "prod"], var.env) - error_message = "Valid value for env is dev, test, sbx, or prod." - } -}