Event Sources - Kinesis

In this section we’ll walkthrough how to trigger your lambda function in response to Amazon Kinesis streams. This overview is based on the SpartaApplication sample code if you’d rather jump to the end result.

Goal

The goal of this example is to provision a Sparta lambda function that logs Amazon Kinesis events to CloudWatch logs.

Getting Started

We’ll start with an empty lambda function and build up the needed functionality.

import (
	awsLambdaEvents "github.com/aws/aws-lambda-go/events"
)
func echoKinesisEvent(ctx context.Context, kinesisEvent awsLambdaEvents.KinesisEvent) (*awsLambdaEvents.KinesisEvent, error) {
	logger, _ := ctx.Value(sparta.ContextKeyRequestLogger).(*logrus.Entry)
	logger.WithFields(logrus.Fields{
		"Event": kinesisEvent,
	}).Info("Event received")
	return &kinesisEvent, nil
}

For this sample all we’re going to do is transparently unmarshal the Kinesis event to an AWS Lambda event, log it, and return the value.

With the function defined let’s register it with Sparta.

Sparta Integration

First we wrap the go function in a LambdaAWSInfo struct:

lambdaFn := sparta.HandleAWSLambda(sparta.LambdaName(echoKinesisEvent),
	echoKinesisEvent,
	sparta.IAMRoleDefinition{})

Since our lambda function doesn’t access any other AWS Services, we can use an empty IAMRoleDefinition (sparta.IAMRoleDefinition{}).

Event Source Registration

Then last step is to configure our AWS Lambda function with Kinesis as the EventSource

lambdaFn.EventSourceMappings = append(lambdaFn.EventSourceMappings,
  &lambda.CreateEventSourceMappingInput{
    EventSourceArn:   aws.String(kinesisTestStream),
    StartingPosition: aws.String("TRIM_HORIZON"),
    BatchSize:        aws.Int64(100),
    Enabled:          aws.Bool(true),
  })

The kinesisTestStream parameter is the Kinesis stream ARN (eg: arn:aws:kinesis:us-west-2:123412341234:stream/kinesisTestStream) whose events will trigger lambda execution.

Wrapping Up

With the lambdaFn fully defined, we can provide it to sparta.Main() and deploy our service. The workflow below is shared by all Kinesis-triggered lambda functions:

Notes