Add option to use kinesis stream to the shipper (#129)

coralogix · Jan 9, 2024 · 4bcd6ce · 4bcd6ce
1 parent a2861f6
commit 4bcd6ce
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 2 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,10 @@
 # Changelog
 
+## v1.0.80
+### 🚀 New components 🚀
+#### **coralogix-aws-shipper**
+- Add option to use Kinesis stream
+
 ## v1.0.79
 ### 🧰 Bug fixes 🧰
 #### **s3-archive**

diff --git a/examples/coralogix-aws-shipper/variables.tf b/examples/coralogix-aws-shipper/variables.tf
@@ -167,3 +167,9 @@ variable "sqs_name" {
   type = string
   default = null
 }
+
+variable "Kinesis_stream_name" {
+  description = "The name of Kinesis stream to subscribe to retrieving messages"
+  type = string
+  default = null
+}
diff --git a/modules/coralogix-aws-shipper/README.md b/modules/coralogix-aws-shipper/README.md
@@ -64,6 +64,11 @@ Coralogix provides a predefined AWS Lambda function to easily forward your logs
 |------|-------------|------|---------|:--------:|
 | <a name="input_sqs_topic_name"></a> [sqs_topic_name](#input\_sqs\_topic\_name) | The SQS name queue to subscribe to retrieving messages| `string` |  n/a | no |
 
+### Integration Kinesis configuration
+| Name | Description | Type | Default | Required | 
+|------|-------------|------|---------|:--------:|
+| <a name="input_Kinesis_stream_name"></a> [Kinesis_stream_name](#input\_Kinesis_\_stream_\_name) | The name of Kinesis stream to subscribe to retrieving messages| `string` |  n/a | no |
+
 ### Integration Generic Config (Optional)
 
 | Name | Description | Type | Default | Required | 

diff --git a/modules/coralogix-aws-shipper/main.tf b/modules/coralogix-aws-shipper/main.tf
@@ -66,6 +66,11 @@ data "aws_sqs_queue" "name" {
   name = var.sqs_name
 }
 
+data "aws_kinesis_stream" "kinesis_stream" {
+  count = var.Kinesis_stream_name != null ? 1 : 0
+  name = var.Kinesis_stream_name
+}
+
 data "aws_iam_policy_document" "topic" {
   count = (local.sns_enable || var.sqs_name != null) && local.is_s3_integration ? 1 : 0
   statement {
@@ -173,6 +178,22 @@ module "lambda" {
     }
     secret_permission = local.secret_access_policy
     destination_on_failure_policy = local.destination_on_failure_policy
+  } : var.Kinesis_stream_name != null ? {
+    Kinesis = {
+      effect    = "Allow"
+      actions   = [
+        "kinesis:GetRecords",
+        "kinesis:GetShardIterator",
+        "kinesis:DescribeStream",
+        "kinesis:ListStreams",
+        "kinesis:ListShards",
+        "kinesis:DescribeStreamSummary",
+        "kinesis:SubscribeToShard"
+        ]
+      resources = [data.aws_kinesis_stream.kinesis_stream[0].arn]
+    }
+    secret_permission = local.secret_access_policy
+    destination_on_failure_policy = local.destination_on_failure_policy
   } : {
     secret_permission = local.secret_access_policy
     destination_on_failure_policy = local.destination_on_failure_policy
@@ -296,6 +317,17 @@ resource "aws_sqs_queue_policy" "sqs_policy" {
 }
 
 
+####################################
+## Kinesis  integration resources ##
+####################################
+
+resource "aws_lambda_event_source_mapping" "example" {
+  count = var.Kinesis_stream_name != null ? 1 : 0
+  event_source_arn  = data.aws_kinesis_stream.kinesis_stream[0].arn
+  function_name     = module.lambda.lambda_function_name
+  starting_position = "LATEST"
+}
+
 ####################################
 ###lambda  integration resources ###
 ####################################

diff --git a/modules/coralogix-aws-shipper/variables.tf b/modules/coralogix-aws-shipper/variables.tf
@@ -111,8 +111,8 @@ variable "integration_type" {
   description = "the aws service that send the data to the s3"
   type        = string
   validation {
-    condition     = contains(["CloudWatch", "CloudTrail", "VpcFlow", "S3", "S3Csv", "Sns", "Sqs"], var.integration_type)
-    error_message = "The integration type must be: [CloudWatch, CloudTrail, VpcFlow, S3, S3Csv, Sns, Sqs]."
+    condition     = contains(["CloudWatch", "CloudTrail", "VpcFlow", "S3", "S3Csv", "Sns", "Sqs", "Kinesis"], var.integration_type)
+    error_message = "The integration type must be: [CloudWatch, CloudTrail, VpcFlow, S3, S3Csv, Sns, Sqs, Kinesis]."
   }
 }
 
@@ -167,3 +167,9 @@ variable "sqs_name" {
   type = string
   default = null
 }
+
+variable "Kinesis_stream_name" {
+  description = "The name of Kinesis stream to subscribe to retrieving messages"
+  type = string
+  default = null
+}