kinesisfirehose: decompress CloudWatch logs and extract message fields

[Tietew]

Describe the feature

Data Firehose supports following convenient conversions:

• Decompress CloudWatch logs
• Extract message after decompression of CloudWatch Logs
• Add a new line delimiter when delivering data to Amazon S3

Use Case

• To use Firehose Data Format Conversion (Parquet, ORC) or Dynamic partitioning, decompression must be enabled.
• These features enable us to avoid decompressing and extracting log entries explicitly on analyze collected logs.

Proposed Solution

They are implemented as processors.

PoC code:

declare const bucket: s3.Bucket;
declare const func: lambda.Function;
new firehose.DeliveryStream(this, 'DeliveryStream', {
  destination: new firehose.S3Bucket(bucket, {
    processors: [
      new firehose.DecompressionProcessor(),
      new firehose.CloudWatchProcessingProcessor(),
      new firehose.LambdaFunctionProcessor(func),
      new firehose.AppendDelimiterToRecordProcessor(),
    ],
  }),
});

will generate a CloudFormation template like:

{
  "ExtendedS3DestinationDescription": {
    // other configurations
    "ProcessingConfiguration": {
      "Enabled": true,
      "Processors": [
        {
          "Type": "Decompression",
          "Parameters": [
            {
              "ParameterName": "CompressionFormat",
              "ParameterValue": "GZIP"
            }
          ]
        },
        {
          "Type": "CloudWatchLogProcessing",
          "Parameters": [
            {
              "ParameterName": "DataMessageExtraction",
              "ParameterValue": "true"
            }
          ]
        },
        {
          "Type": "Lambda",
          "Parameters": [
            {
              "ParameterName": "LambdaArn",
              "ParameterValue": "arn:aws:lambda:..."
            },
            {
              "ParameterName": "RoleArn",
              "ParameterValue": "arn:aws:iam:..."
            }
          ]
        },
        {
          "Type": "AppendDelimiterToRecord",
          "Parameters": []
        }
      ]
    }
  }
}

Other Information

Related issues and PRs:

• aws-logs-destinations: Missing FirehoseDestination or DataFirehoseDestination #32038
• feat(logs-destinations): support Amazon Data Firehose logs destination #33683
• (aws-kinesisfirehose): Record format conversion with AWS Glue for DeliveryStream S3 destination #15501
• aws-kinesisfirehose-destinations-alpha: Support dynamic partitioning with inline parsing #28740
• Add delimiter and/or generic implementation of IDataProcessor for Kinesis Firehose #20242
• feat(kinesisfirehose): add built-in data processors to decompress CloudWatch logs and extract messages #33749

Acknowledgements

• I may be able to implement this feature request
• This feature might incur a breaking change

CDK version used

2.181.0

Environment details (OS name and version, etc.)

Linux

[pahud]

Thank you. I guess we probably need to add DecompressionProcessor and CloudWatchLogMessageExtractionProcessor implementations and maybe update S3Bucket destination to support accepting multiple processors.

Anyways, we appreciate your pull request and we'll take a look when it's ready.

[Tietew]

@pahud Thank you for your suggestion!

But the reasons why I chose booleans instead of processor classes are:

• AWS management console only shows us checkboxes with no additional parameters.
• The processors should be specified in order: decompression - extraction - format conversion - custom lambda processor - appending delimiter
• The parameters of the processors should have predefined fixed values.

Anyway I will make more investigation about the behavior.

[Tietew]

Specify Decompression only

Worked but all event entries are joined without delimiter.

{"messageType":"DATA_MESSAGE",...,"logEvents":[...]}{"messageType":"DATA_MESSAGE",...

Adding the AppendDelimiterToRecord processor will be needed.

Specify Decompression and CloudWatchLogProcessing

Worked as expected. Delivered file contains only event messages delimited by LF.
The AppendDelimiterToRecord processor will not be needed.

Specify CloudWatchLogProcessing then Decompression

Worked. Log event messages were decompressed then extracted.

Specify CloudWatchLogProcessing without Decompression

CloudFormation rejected:

Resource handler returned message: "CloudWatchLogProcessingProcessor can only be enabled with DecompressionProcessor (Service: Firehose, Status Code: 400, Request ID: xxx)"

[Tietew]

I've changed my position.
I'll add built-in processor classes and update destination to support accepting multiple processors.
The PoC code in the description has been updated.

[Tietew]

Hi @pahud!
I've submitted a PR requesting a clarification how to resolve incompatible changes.
Would you please take a look at the PR?
Thank you.