Skip to contentSkip to navigationSkip to topbar
Page tools
Useful for sharing or LLM
Accelerate development with AI
On this page
Looking for more inspiration?Visit the

Kinesis scripts

Use the following Bash scripts to automate the creation of a Kinesis Stream and retrieve a test message. Copy the code below and save it to your computer, for example as create_kinesis_stream.sh or retrieve_test_message.sh.

Create a Kinesis stream

create-a-kinesis-stream page anchor

Run chmod +x create_kinesis_stream.sh to make it executable.

You also need jq, a command line JSON processor the script depends on. For installation instructions for your OS, see the jq download page(link takes you to an external page).

The script also depends upon the AWS CLI, which you will need to install and configure(link takes you to an external page) before executing the script.

The script takes two arguments: your chosen AWS Kinesis Stream name and a shard count.

1
#!/bin/bash

2


3
JQ_CHECK=$(which jq)

4
if [ -z "$JQ_CHECK" ]; then

5
  echo

6
  echo "This script requires the jq JSON processor. Please install for your OS from https://jqlang.org/download/"

7
  echo

8
  exit 1

9
fi

10


11
if [ $# -ne 2 ]; then

12
  echo

13
  echo "usage: $0 <stream_name> <shard_count>"

14
  echo

15
  exit 1

16
fi

17


18
# Set the stream name

19
STREAM_NAME=${1:-twilio-events}

20
SHARD_COUNT=${2:-1}

21


22
# Create the initial stream

23
aws kinesis create-stream --stream-name $STREAM_NAME --shard-count $SHARD_COUNT

24
if [ $? -ne 0 ]; then

25
  echo "Kinesis create failed"

26
  exit 1

27
fi

28


29
# Get the ARN for the Kinesis Stream

30
KINESIS_ARN=$(aws kinesis describe-stream --stream-name $STREAM_NAME | jq -r .StreamDescription.StreamARN)

31


32
# Create the policy for the Kinesis Stream

33
POLICY_ARN=$(aws iam create-policy --policy-name twilio-events-kinesis-write --policy-document '{

34
   "Version": "2012-10-17",

35
   "Statement": [

36
       {

37
           "Sid": "Quickstart0",

38
           "Effect": "Allow",

39
           "Action": [

40
               "kinesis:PutRecord",

41
               "kinesis:PutRecords"

42
           ],

43
           "Resource": "'$KINESIS_ARN'"

44
       },

45
       {

46
           "Sid": "Quickstart1",

47
           "Effect": "Allow",

48
           "Action": [

49
               "kinesis:ListShards",

50
               "kinesis:DescribeLimits"

51
           ],

52
           "Resource": "*"

53
       }

54
   ]

55
}' | jq -r .Policy.Arn)

56


57
if [ -z "$POLICY_ARN" ]; then

58
  echo "Failed to create IAM policy"

59
  exit 1

60
fi

61


62
# Generate a random external ID

63
EXTERNAL_ID=$(openssl rand -hex 40)

64
if [ -z "$EXTERNAL_ID" ]; then

65
  echo "Failed to generate external ID"

66
  exit 1

67
fi

68


69
# This is Twilio's intermediary role, which needs permission to assume the

70
# role created below (Twilio assumes your role via a two-hop chain, so the

71
# trust policy must name this specific role, not just Twilio's account)

72
TWILIO_ASSUME_ROLE_ACCOUNT=${TWILIO_ASSUME_ROLE_ACCOUNT:-arn:aws:iam::177261743968:role/egress-domain-prod-isolated}

73


74
# Add the random external ID to the role ARN

75
# More information can be found here: https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html

76
ROLE_ARN=$(aws iam create-role --role-name twilio-events-kinesis-write --assume-role-policy-document '{

77
 "Version": "2012-10-17",

78
 "Statement": [

79
   {

80
     "Effect": "Allow",

81
     "Principal": {

82
       "AWS": "'$TWILIO_ASSUME_ROLE_ACCOUNT'"

83
     },

84
     "Action": "sts:AssumeRole",

85
     "Condition": {

86
       "StringEquals": {

87
         "sts:ExternalId": "'$EXTERNAL_ID'"

88
       }

89
     }

90
   }

91
 ]

92
}' | jq -r .Role.Arn)

93


94
if [ -z "$ROLE_ARN" ]; then

95
  echo "Failed to create IAM role"

96
  exit 1

97
fi

98


99
# Finally attach the policy and the role

100
aws iam attach-role-policy --role-name twilio-events-kinesis-write --policy-arn $POLICY_ARN

101


102
if [ $? -ne 0 ]; then

103
  echo "Attaching policy to role failed"

104
  exit 1

105
fi

106


107
# Print out the values needed for creating the sink in nice JSON

108
echo "{"

109
echo '"arn":"'$KINESIS_ARN'",'

110
echo '"role_arn":"'$ROLE_ARN'",'

111
echo '"external_id":"'$EXTERNAL_ID'"'

112
echo "}"

Retreive a test message

retreive-a-test-message page anchor

You can use the following Bash script to automate the retreival of a test message. Copy the code below and save it to your computer, for example as retrieve_test_message.sh.

1
#!/bin/bash

2


3
JQ_CHECK=$(which jq)

4
if [ -z "$JQ_CHECK" ]; then

5
  echo

6
  echo "This script requires the jq JSON processor. Please install for your OS from https://jqlang.org/download/"

7
  echo

8
  exit 1

9
fi

10


11
if [ $# -ne 1 ]; then

12
  echo

13
  echo "usage: $0 <stream_name>"

14
  echo

15
  exit 1

16
fi

17


18
# Set the stream name

19
STREAM_NAME=$1

20


21
# Choose the iterator type:

22
# TRIM HORIZON is for starting at the beginning of the Kinesis Stream.

23
# This can take a while if you have a lot of records.

24
# To use TRIM HORIZON, uncomment the following line:

25
# TYPE=TRIM_HORIZON

26


27
# AT_TIMESTAMP allows you to go back to a point in time. This is set for going back one hour

28
# To use AT_TIMESTAMP, uncomment the following two lines:

29
# TIMESTAMP=$(($(date +%s) - 3600)).000

30
# TYPE="AT_TIMESTAMP --timestamp $TIMESTAMP"

31


32
# LATEST means start at the most current point in the stream and read forward

33
TYPE=LATEST

34


35
# Get a list of shards

36
SHARDS=$(aws kinesis list-shards --stream-name $STREAM_NAME | jq -r .Shards[].ShardId)

37


38
# Get all the starting points

39
SHARD_ITERATOR=()

40
i=0

41
for shard in $SHARDS ; do

42
  SHARD_ITERATOR[$i]=$(aws kinesis get-shard-iterator --shard-id $shard --shard-iterator-type $TYPE --stream-name $STREAM_NAME --query 'ShardIterator')

43
  i=$((i+1))

44
done

45


46
# Start getting events from all shards and display them

47
while [ 1 ] ; do

48
  len=${#SHARD_ITERATOR[@]}

49
  for (( i=0; i < $len; i++ )); do

50
    DATA=$(aws kinesis get-records --limit 50 --shard-iterator ${SHARD_ITERATOR[$i]})

51
    SHARD_ITERATOR[$i]=$(echo $DATA | jq -r .NextShardIterator)

52
    ROWS=$(echo $DATA | jq -r .Records[].Data?)

53
    for row in $ROWS; do

54
      echo $row | base64 -d | jq .

55
    done

56
  done

57
done