firehose cloudwatch logs

Author: Karthikeyannhs

infrastructure/modules/kinesis_firehose/kinesis_firehose_delivery_stream.tf

@@ -9,6 +9,12 @@ resource "aws_kinesis_firehose_delivery_stream" "eligibility_audit_firehose_deli
     buffering_size     = 1
     buffering_interval = 60
     compression_format = "UNCOMPRESSED"
+
+    cloudwatch_logging_options {
+      enabled         = true
+      log_group_name  = var.kinesis_cloud_watch_log_group_name
+      log_stream_name = "to-s3"
+    }
   }
 
   tags = var.tags

infrastructure/modules/kinesis_firehose/variables.tf

@@ -13,3 +13,9 @@ variable "s3_audit_bucket_arn" {
   type        = string
 }
 
+variable "kinesis_cloud_watch_log_group_name" {
+  description = "kinesis cloud watch log group name"
+  type        = string
+}
+
+

infrastructure/stacks/api-layer/cloudwatch.tf

@@ -9,3 +9,13 @@ resource "aws_cloudwatch_log_group" "lambda_logs" {
     Stack = local.stack_name
   }
 }
+
+resource "aws_cloudwatch_log_group" "firehose_audit" {
+  name              = "/aws/kinesisfirehose/${var.project_name}-${var.environment}-audit"
+  retention_in_days = 365
+
+  tags = {
+    Name  = "kinesis-firehose-logs"
+    Stack = local.stack_name
+  }
+}

infrastructure/stacks/api-layer/iam_policies.tf

@@ -63,6 +63,29 @@ resource "aws_iam_role_policy" "kinesis_firehose_s3_write_policy" {
   policy = data.aws_iam_policy_document.s3_audit_bucket_policy.json
 }
 
+# Policy doc for firehose logging
+resource "aws_iam_role_policy" "kinesis_firehose_logs_policy" {
+  name   = "CloudWatchLogsAccess"
+  role   = aws_iam_role.eligibility_audit_firehose_role.id
+
+  policy = jsonencode({
+    Version = "2012-10-17",
+    Statement = [
+      {
+        Effect = "Allow",
+        Action = [
+          "logs:CreateLogStream",
+          "logs:PutLogEvents",
+          "logs:DescribeLogGroups",
+          "logs:DescribeLogStreams"
+        ],
+        Resource = "*"
+      }
+    ]
+  })
+}
+
+
 # Attach AWSLambdaVPCAccessExecutionRole to Lambda
 resource "aws_iam_role_policy_attachment" "AWSLambdaVPCAccessExecutionRole" {
   role       = aws_iam_role.eligibility_lambda_role.id

infrastructure/stacks/api-layer/kinesis_firehose.tf

@@ -7,4 +7,5 @@ module "eligibility_audit_firehose_delivery_stream" {
   stack_name                          = local.stack_name
   workspace                           = local.workspace
   tags                                = local.tags
+  kinesis_cloud_watch_log_group_name  = aws_cloudwatch_log_group.firehose_audit.name
 }

src/eligibility_signposting_api/config/config.py

@@ -15,6 +15,7 @@
 AwsRegion = NewType("AwsRegion", str)
 AwsAccessKey = NewType("AwsAccessKey", str)
 AwsSecretAccessKey = NewType("AwsSecretAccessKey", str)
+AwsKinesisFirehoseStreamName = NewType("AwsKinesisFirehoseStreamName", str)
 
 
 @cache
@@ -23,7 +24,7 @@ def config() -> dict[str, Any]:
     rules_bucket_name = BucketName(os.getenv("RULES_BUCKET_NAME", "test-rules-bucket"))
     audit_bucket_name = BucketName(os.getenv("AUDIT_BUCKET_NAME", "test-audit-bucket"))
     aws_default_region = AwsRegion(os.getenv("AWS_DEFAULT_REGION", "eu-west-1"))
-    kinesis_audit_stream_to_s3 = AwsRegion(os.getenv("KINESIS_AUDIT_STREAM_TO_S3", "test_kinesis_audit_stream_to_s3"))
+    kinesis_audit_stream_to_s3 = AwsKinesisFirehoseStreamName(os.getenv("KINESIS_AUDIT_STREAM_TO_S3", "test_kinesis_audit_stream_to_s3"))
     log_level = LOG_LEVEL
 
     if os.getenv("ENV"):

src/eligibility_signposting_api/services/audit_service.py

@@ -6,13 +6,15 @@
 from wireup import Inject, service
 from yarl import URL
 
+from eligibility_signposting_api.config.config import AwsKinesisFirehoseStreamName
+
 logger = logging.getLogger(__name__)
 
 
 @service
 class AuditService:
 
-    def __init__(self, firehose: Annotated[BaseClient, Inject(qualifier="firehose")], audit_delivery_stream: Annotated[URL, Inject(param="kinesis_audit_stream_to_s3")] ) -> None:
+    def __init__(self, firehose: Annotated[BaseClient, Inject(qualifier="firehose")], audit_delivery_stream: Annotated[AwsKinesisFirehoseStreamName, Inject(param="kinesis_audit_stream_to_s3")] ) -> None:
         super().__init__()
         self.firehose = firehose
         self.audit_delivery_stream = audit_delivery_stream