I'm trying to stream Punchh webhook JSON data through kinesis delivery stream to multiple S3 buckets but it is giving me this error. However, when I streamed Iterable webhook data, the Iterable S3 buckets are successfully filled with the data from the deliverable stream sent from the Iterable wehbook.

Here is the error:

An error occurred (ValidationException) when calling the PutRecord operation: 1 validation error detected: Value 'data/landing/vendor/punchh/punchh_to_dd/ex_points_reminder' at 'deliveryStreamName' failed to satisfy constraint: Member must satisfy regular expression pattern: [a-zA-Z0-9_.-]

Note: The S3 prefix value -in bold- that the error mentioned above is similar to the Iterable prefixes I have successfully tested .

Here is the Punchh S3 prefix for one of the streams causing the error:

data/landing/vendor/punchh/punchh_to_dd/ex-points-reminder/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/dynamic

Here is the Punch S3 stream destination (note X is used as a place holder to conceal the real address which is irrelative):

arn:aws:firehose:us-west-2:XXXXXXXXXXXX:deliverystream/lou-punchh-ex-points-reminder-events

Here is the Python Code for the proxy (backend) Lambda (Similar code for Punchh as Iterable) . I have several dynamic S3 buckets paired with their respective delivery streams:

import boto3
import json
import base64
import os

client = boto3.client('firehose', region_name= 'us-west-2')
delivery_stream_lou = {}
def lambda_handler(event, context):
    try:
        print(event)
        print('<< EVENT BODY  >>',event['event_name'])
        try:
           
            print('<< EVENT NAME >> ',event['event_name'])
            eventName = event['event_name']
            eventType =  event['event_type']
            if 'users' == eventName.lower():
                #ADD firehose KINESIS for USERS
                print('USERS FOUND')
                delivery_stream_lou = os.environ['DSTREAMUSERS']
               
            elif 'loyalty' in eventType.lower():
                #ADD firehose KINESIS for LOYALTY
                print('LOYALTY FOUND')
                delivery_stream_lou = os.environ['DSTREAMLOYALTY']
                
            elif 'gift' == eventType.lower():
                #ADD firehose KINESIS for LOYALTY
                print('GIFT FOUND')
                delivery_stream_lou = os.environ['DSTREAMGIFT']
                
            elif 'redemptions' == eventName.lower():
                #ADD firehose KINESIS for REDEMPTIONS
                print('REDEMPTIONS FOUND')
                delivery_stream_lou = os.environ['DSTREAMREDEMPTIONS']
                
            elif 'rewards' == eventName.lower():
                #ADD firehose KINESIS for REWARDS
                print('REWARDS FOUND')
                delivery_stream_lou = os.environ['DSTREAMREWARDS']
                
            elif 'redeemables' == eventName.lower():
                #ADD firehose KINESIS for REDEEMABLES
                print('REDEEMABLES FOUND')
                delivery_stream_lou = os.environ['DSTREAMREDEEMABLES']
            
            elif 'points_expiry_reminder' == eventType.lower():
                #ADD firehose KINESIS for EXPIRY PT REMINDER
                print('EXPIRY PT REMINDER FOUND')
                delivery_stream_lou = os.environ['DSTREAMEXPOINTSTREMINDER']  
                print('<< DELIVERY STREAM LOU >> ', delivery_stream_lou)
            
            elif 'points_expiry' == eventType.lower():
                #ADD firehose KINESIS for EXPIRY PT REMINDER
                print('EXPIRY PT FOUND')
                delivery_stream_lou = os.environ['DSTREAMEXPOINTS']
                
            elif 'signup_campaign' == eventType.lower():
                #ADD firehose KINESIS for SIGNUP CAMPAIGN
                print('SIGNUP FOUND')
                delivery_stream_lou = os.environ['DSTREAMSIGNUP']  
                
            else:
                print('<< ERROR - NO VALID EVENT NAME FOUND !!  >>')
                
            if delivery_stream_lou !='' and delivery_stream_lou is not None and delivery_stream_lou != 'null':
                jsonData = json.dumps(event)   '\n' 
                print('<< JSON DATA >> ',jsonData)  
                response = client.put_record(DeliveryStreamName=delivery_stream_lou, Record={'Data': jsonData})  
                print(response)
                return {
                    'statusCode': 200,
                    'headers': {'Content-Type': 'application/json'},
                    'body': json.dumps({"message":"success"})
                }
        except Exception as e: 
            print(str(e))
            return {
                'statusCode': 200,
                'headers': {'Content-Type': 'application/json'},
                'body': json.dumps({"message":"process failed"})
            }
    except Exception as e:
        print(str(e))
        return {
            'statusCode': 200,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({"message":"process failed 2"})
        }

Note: The code is almost identical for this Punchh lambda as the Iterable Lambda except for the different os.environment variables that identify the different buck prefixes.

Thanks for your help.

CodePudding user response：

As per the error message python code is passing data/landing/vendor/punchh/punchh_to_dd/ex_points_reminder value to deliveryStreamName parameter in PutRecord request. I dont think thats the name of your delivery stream. Check values for environments variable that are populating deliveryStreamName parameter