DEV Community: Kohei (Max) MATSUSHITA

How to set up cross-account access to Amazon Kinesis Data Streams

Kohei (Max) MATSUSHITA — Thu, 29 Feb 2024 11:24:11 +0000

Overview - Amazon Kinesis Data Streams and separated AWS accounts

Amazon Kinesis Data Streams (hereafter referred to as KDS) is a managed data processing service designed for the real-time collection of high traffic of data and facilitating its transfer to subsequent AWS services. It is particularly suited for handling streaming data, such as logs, where order matters, making it a commonly used service for IoT data collection. For example, it can be specified as the data export destination for Amazon Monitron, which allows for predictive maintenance of industrial equipment through machine learning.

KDS serves as an intermediary between data producers and data consumers, employing this architecture for its operation.

When fully leveraging AWS Cloud, it's not uncommon to operate separate AWS accounts for purposes of distinction and cost management. In such cases, there might be a need to process streaming data collected by KDS in another AWS account.

This blog introduces the procedure for sharing a data stream with another AWS account and referencing it from AWS Lambda (specifying the Lambda function as a trigger) using the resource-based policy of Amazon Kinesis Data Streams (update as of November 2023). A data stream, in this context, is akin to a pipeline through which data flows.

The Architecture and Setup

The architecture is as follows.

Key Points for Setup

Here are the key points for setup:

The execution role of the Lambda function on the data processing side(Account B) requires the AWSLambdaKinesisExecutionRole policy.
For the KDS data stream on the data producer side(Account A), the resource-based policy should specify the IAM role ARN of the execution role of the Lambda function on the data processing side as the Principal, and set the Allow Actions to include kinesis:DescribeStream, kinesis:DescribeStreamSummary, kinesis:GetRecords, kinesis:GetShardIterator and kinesis:ListShards.

Especially regarding the second point on resource-based policies, it's important to note that specifying kinesis:DescribeStream might be missed when using the dialog in the management console. It needs to be manually added using the JSON editor (as of Feb. 2024, reported).

The following official documents might also be helpful:

Steps

The steps involve several cross-accounts: data producer side(Account A) → data processing side(Account B) → data producer side → data processing side. Make sure not to mix up the targets.
Everything must be in the same region; sharing is not possible across different regions (e.g., if the data stream is in us-west-2 and the Lambda function is in ap-northeast-1). If you wish to send data to a different region, consider the architecture using Amazon EventBridge mentioned in the epilogue.

Step 1) On the data producer side - Account A

1-1. Create a data stream in Amazon Kinesis (e.g., `kds-sharing-example1`)

See here for the creation method (Step1: Create Data Stream).

For testing or small data volumes, the "Provisioned" capacity mode with "1" provisioned shard is sufficient.

NOTE: that while this guide assumes the creation of a new data stream, existing data streams can also be repurposed.

Step 2) On the data processing side - Account B

2-1. Create a Lambda function in AWS Lambda (e.g., `kds-reader1`).

See here for the creation method (Create a Lambda function with the console).

The source code is as follows (Python 3). It simply emits the event to Amazon CloudWatch Logs, which is sufficient for operational verification. After modifying the code source as below, click "Deploy" to deploy.

def lambda_handler(event, context):
    print(event)
    return True

NOTE: There's a template for testing the function with Amazon Kinesis Data Streams sample data, which can be used for testing.

2-2. Attach the `AWSLambdaKinesisExecutionRole` policy to the execution role of `kds-reader1`.

After viewing the details of kds-reader1, go to "Cofiguration" > "Permissions" and click on the role name assigned to the execution role to view the role's settings (in the figure below, click on kds-reader1-role-mp67l1v2).

Go to "Add Permission" > "Attach Policy" for the permission policy to display the list of policies to attach.

Select AWSLambdaKinesisExecutionRole from the list of "Other Permission Policies" and then click "Add Permission".

Policy addition is complete.
Check for AWSLambdaKinesisExecutionRole is added to the list of allowed policies as shown below.

NOTE: We assume that the role for executing Lambda functions is an IAM role that is automatically created and appended when a Lambda function is created. Existing IAM roles can also be used.

2-3. Noted the ARN of the IAM role.

This IAM role's ARN (in this step, kds-reader1-role-q6zcv9kq) will be used on next step.

Step 3) Back on the data producer side - Account A

3-1. configure resource-based policy for `kds-sharing-example1` in Amazon Kinesis

After viewing the details of kds-sharing-example1, go to "Data stream sharing" > "Create Policy".

In Policy Details, select the Visual Editor, check "Data stream sharing throughput read Access," enter the ARN of the IAM role you wrote down earlier in "Specify Principal(s)," and click "Create Policy."

NOTE: The principal should be the ARN of the IAM role; specifying an ARN other than the IAM role's ARN (e.g., the ARN of a Lambda function or data stream) will result in an error and the policy cannot be created.

When the resource-based policy appears, click "Edit" to display the JSON editor. Here, add "kinesis:DescribeStream", to the list of Actions as shown below. Finally, click "Save Changes".

NOTE: If the same privileges have already been added at the visual editor stage, the above editing process is not necessary.

Configuration of the resource-based policy is complete.
Check to that the ARN of the IAM role is attached to the Principal and the five permissions are attached to the Action, as shown below.

3-2. Noted the ARN of `kds-sharing-example1`.

kds-sharing-example1 (KDS Stream's) ARN will be used on next step.

Step 4) Back on the data processing side - Account B

4-1. set up a trigger for kds-reader1 in AWS Lambda

After viewing the details of kds-reader1, go to "Configuration" > "Triggers" > "Add trigger".

In the Trigger configuration, select Kinesis for "Select a source". Then set the ARN of kds-sharing-example1 to "Kinesis stream" in the displayed contents. Leave the other items as they are and click "Add".

NOTE1: When the focus moves to the text box, the message "No item" is displayed. No problem, ignore.

NOTE2: If you get an API error when clicking "Add", check the following two things (1) Resource-based policy permissions on the data stream side(Account A). In particular, make sure that kinesis:DescribeStream is included. (2) Permissions for the Lambda function execution role. In particular, make sure that the AWSLambdaKinesisExecutionRole policy is attached.

Trigger configuration is complete.
You can see that Kinesis has been added to the trigger (input source) of kds-reader1 as shown below.

How to check

To check, send data to the data stream (kds-sharing-example1 in this example) on the data generator side(Account A) and check the Amazon CloudWatch Logs output on the data processor side(Account B).

AWS CloudShell on the data generator side(Account A) sends data to the data stream via AWS CLI.

aws --cli-binary-format raw-in-base64-out \
  kinesis put-record --stream-name kds-sharing-example1 \
  --partition-key DUMMY1 \
  --data '{"this_is": "test record"}'

If the result of the command execution is as follows, the data transmission has succeeded.

{
    "ShardId": "shardId-000000000000",
    "SequenceNumber": "49649718468451075013017298672854645152715037125279481858"
}

If the following log is confirmed in the log of kds-reader1 in CloudWatch Logs on the data processing side (Account B), the setting was successful.

{'Records': [{'kinesis': {'kinesisSchemaVersion': '1.0', 'partitionKey': 'DUMMY1', 'sequenceNumber': '49649683864473953433843496278632352517587667908684677122', 'data': 'eyJ0aGlzX2lzIjogInRlc3QgcmVjb3JkIn0=', 'approximateArrivalTimestamp': 1709100911.667}, 'eventSource': 'aws:kinesis', 'eventVersion': '1.0', 'eventID': 'shardId-000000000000:49649683864473953433843496278632352517587667908684677122', 'eventName': 'aws:kinesis:record', 'invokeIdentityArn': 'arn:aws:iam::888800008888:role/service-role/kds-reader1-role-q6zcv9kq', 'awsRegion': 'us-east-1', 'eventSourceARN': 'arn:aws:kinesis:us-east-1:999900009999:stream/kds-sharing-example1'}]}

Epilogue - Architecture with Amazon EventBridge

In this article introduced sharing data streams using Amazon Kinesis Data Streams resource-based policies. This will allow, for example, the Amazon Monitron data introduced at the beginning of this article to be used by other AWS accounts, which will give you more flexibility in the operation of your AWS account.

Other possible architectures for using Amazon Kinesis Data Streams data streams with other AWS accounts include sending them through the Amazon EventBridge event bus.

The advantage of this would be that it can be configured as a no-code and managed service. It can also be used across different regions, and although there is a fee for using Amazon EventBridge, the architecture is well worth it.

Here are some URLs to help you create this architecture.

I personally believe that the ability to create a configuration that matches the skills you possess and the type of operation you wish to achieve is the best part of building blocks.

Not only this configuration, but it would be a good idea to TRY a configuration that is appropriate for the time, along with new features that may come out in the future!

[EoT]

Predictive Maintenance of Industrial machine with Amazon Monitron and What is “ISO 20816” ?

Kohei (Max) MATSUSHITA — Thu, 28 Dec 2023 02:56:36 +0000

In this article, learn about Amazon Monitron, an AWS service that enables predictive maintenance for industrial machinery, and the ISO 20816 standard used in the service.

Predictive Maintenance in Industry: The article highlights the importance of predictive maintenance in the industrial sector, emphasizing the challenges of maintaining factory automation systems and other industrial machines, which are crucial for reliable operations.
Amazon Monitron and ISO 20816: Amazon Monitron, an AWS service, is introduced as a solution for predictive maintenance. It integrates sensors, machine learning models, and follows the ISO 20816 standard for machine vibration, to efficiently detect anomalies in industrial machinery.
Application of ISO 20816 in Amazon Monitron: ISO 20816 is an international standard for evaluating machine vibration, used by Amazon Monitron to classify machines and assess their conditions. This system simplifies setting up predictive maintenance, even for machines without initial sensor systems.

Please see the following a medium article for details;
https://ma2shita.medium.com/predictive-maintenance-of-industrial-machine-with-amazon-monitron-and-what-is-iso-20816-9715f2d597b5

AWS CDK stack for building Amazon EventBridge Pipes (SQS to CloudWatch Logs)

Kohei (Max) MATSUSHITA — Fri, 10 Mar 2023 06:12:12 +0000

I have released an AWS CDK(v2) stack that builds the following Amazon EventBridge Pipes.

This stack provides a quick experience in building Amazon EventBridge Pipes.

ma2shita / cdk-eventbridge-pipes-simplelogger

Example an AWS CDKv2 for building the Amazon EventBridge Pipes that logs messages sent to an SQS queue to CloudWatch Log.

AWS CDKv2 example: SQS to CloudWatch Logs (w/ Enrichment by Lambda) on Amazon EventBridge Pipes

Example an AWS CDKv2 for building the Amazon EventBridge Pipes that logs messages sent to an SQS queue to CloudWatch Log.
It also includes an enrichment(but pass-throught) by a Lambda function.

(ja) SQSキューに送信されたメッセージをCloudWatch Logにログ出力するAmazon EventBridge Pipesを構築する、AWS CDKv2 のサンプル。
Lambda関数による強化(enrichment/内容は"何もしない")も含んでいます。

Created by:

Queue (1)
Log Group (1)
Lambda function (1)
Pipe (1)

Requirements

AWS CDKv2 env. (AWS Cloud9 (It's very easy))
- cdk bootstrap must be done

Build & Deploy

Build:

git clone https://github.com/ma2shita/cdk-eventbridge-pipes-simplelogger.git
cd eventbridge-pipes-simplelogger/
npm install
cdk ls
# => EventBridgePipesSimpleLoggerStack

Note If you have "Newer version" in npm or aws_cdk then run to npm install -g --force npm aws-cdk

Deploy:

cdk deploy

Note The target region is set to us-west-2. This is specified in bin/event_bridge_pipes_simple_logger.ts.

Destroy:

cdk destroy

How it works

Send SQS queue message then will record toeven CloudWatch Log…

View on GitHub

If you are on AWS Cloud9, you can cdk bootstrap immediately because the environment is ready up to the cdk command.

Contains "enrichment" Lambda functions

"Enrichment" can invoke Lambda function and Step Function, reducing the amount of code compared to a stand-alone Lambda implementation.

This CDKv2 stack contains Lambda functions that can be called from "enhancements". It is "do nothing" code, though, so please see it as a framework.
How to customize it is described below.

Customization Points

On the "Enrichment" function

The "enrichment" function allows editing of the payload. For example, you can lookup items from DynamoDB by the input then append the results to the payload.

For specifications of the Lambda function to be specified for enhancement, see here.
AWS Lambda function specifications in "Enrichment" in Amazon EventBridge Pipes

On the "Target" of EventBridge Pipes

Once you have changed the "target", remember to grant write permission to the destination service to the Pipes Execution role.
In particular, events:InvokeApiDestination is required if you specify an "API destination" that can send to an external API.

Conclusion and "CDKv2 for IoT 1-Click"

I'd love to hear your feedback!

And, an AWS CDK(v2) stack for creating AWS IoT 1-Click projects is also available. Button data can be sent to Amazon EventBridge Pipes.

ma2shita / cdk-iot1click-project-to-sqs

Example of an AWS CDKv2 for building the AWS IoT 1-Click project with sending to exists SQS queue Lambda function.

AWS CDKv2 example: IoT 1-Click to exists SQS queue

Example of an AWS CDKv2 for building the AWS IoT 1-Click project with sending to exists SQS queue Lambda function.

(ja) 既存のSQSキューへメッセージ送信をするIoT 1-Clickプロジェクトと、Lambda 関数を構築するAWS CDKv2の例。

Created by:

Lambda function (1)
IoT 1-Click project (1)

NOT creates:

SQS queue
IoT 1-Click placement
IoT 1-Click device registration

Requirements

AWS CDKv2 env. (AWS Cloud9 (It's very easy))
- cdk bootstrap must be done
SQS queue
- queue's ARN
AWS IoT 1-Click device

Build & Deploy

Build:

git clone https://github.com/ma2shita/cdk-iot1click-project-to-sqs.git
cd iot1click-project-to-sqs/
npm install
cdk ls --context destSqsQueueArn=<SQS_QUEUE_ARN>
#=> Iot1ClickProjectStack
# e.g.) cdk ls --context destSqsQueueArn=arn:aws:sqs:REGION:ACCOUNT:sqs-queue-name

Deploy:

cdk deploy --context destSqsQueueArn=<SQS_QUEUE_ARN>

Note The target region is set to us-west-2. This is specified in bin/iot1click_project.ts.

Destroy:

cdk destroy --context destSqsQueueArn=<SQS_QUEUE_ARN>

How it works

Register devices in IoT 1-Click then add the device to the project (created with this CDK).
Pressing the button sends the message…

View on GitHub

So, It would be nice if IoT 1-Click would support sending directly to SQS. :-)

EoT

AWS Lambda function specifications in "Enrichment" in Amazon EventBridge Pipes

Kohei (Max) MATSUSHITA — Wed, 08 Mar 2023 05:40:29 +0000

TL;DR

Input to the "Enrichment" function
The first argument "event" of the Lambda function is passed as an array of payloads from each AWS service.

Output from the "Enrichment" function
return ... is passed to the "target" of the EventBridge Pipes.

Explanation

This section is based on a Lambda function (as will be described later) that receives an event from a source or filter and passes the event to a target without doing anything (pass-through).

def lambda_handler(event, context):
    print(event)
    return event

export const handler = async (event, context) => {
    console.log(event);
    return event;
};

Input to the function of "Enrichment"

The first argument "event" of the Lambda function is passed as a payload from the source or filter. The second argument "context" can also be used.

For example, if you set Amazon SQS as the source and put the data into the SQS queue of MessageBody as {"v":1}, the result of print(event)(in case of python) will look like this.

[
    {
        'messageId': '8c8b72a2-85f0-47be-bf9f-81edff5197e9',
        'receiptHandle': '.....',
        'body': '{"v":1}',
        'attributes': {
            'ApproximateReceiveCount': '1',
            'SentTimestamp': '1672477371472',
            'SenderId': 'AID................KM',
            'ApproximateFirstReceiveTimestamp': '1672477371473'
        },
        'messageAttributes': {},
        'md5OfMessageAttributes': None,
        'md5OfBody': '455c0c10d425837001a72a84239ac362',
        'eventSource': 'aws:sqs',
        'eventSourceARN': 'arn:aws:sqs:REGION:............:SQS_NAME',
        'awsRegion': 'REGION'
    }
]

The first level of "event" is an array. Therefore, fetch within a Lambda function can be made to, for example, messageId as event[0]['messageId'](in case of python).

def lambda_handler(event, context):
    print(event[0]['messageId'])
    return event

"event" is an array. Therefore, I thought multiple payloads would be entered. So far, no multiple payloads have been entered.
For example, even if 3 data were queued in SQS at once using send-message-batch, the Lambda function specified as Enrichment was executed 3 times individually.

Output from the function of "Enrichment"

return ... is passed to the target of the EventBridge Pipes.

The following code is a Lambda function that passes {"foo": "bar"} "target".

def lambda_handler(event, context):
    return {'foo': 'bar'}

If "empty" is returned, the target will not be executed. Empty means {}, [] and "". The following code is a Lambda function(Python3) that does not execute "target".

def lambda_handler(event, context):
    return {}

Implicit body data parsing

One point to note when processing data with Enrichment is Implicit body data parsing.
If Enrichment is not used, Implicit body data parsing is applied and can be referenced in the input transformer on the target side as <$.body.foo>.
If Enrichment is used, Implicit body data parsing is not applied.
Therefore, when using Enrichment, it is better to design the input transformer to be fully formatted in the Enrichment Lambda function before passing it on to the target, to simplify the input transformer.

My strategy for implementing Enrichment Lambda functions

At first, if enrichment is to be used, it would be better to implement without target input transformers. This is to concentrate the formatting implementation.

Let's consider Slack's chat.postMessage API . That requires a channel and text.

The first approach implements formatting to the API in a function. The second approach is to pass the necessary information to target, and the formatting to the API itself is done in the input transformer.

Approach 1: "Rewriting."

def lambda_handler(event, context):
    payload = {
        'channel': 'XXX',
        'text': event[0]['body']
    }
    return payload
# => {
#  'channel': 'XXX',
#  'text': "STRING...."
#}

Input transformar in "target" is nothing.

Approach 2: "Adding."

def lambda_handler(event, context):
    event[0]['_enrichment'] = {
        'slack': {
            'channel': 'XXX',
            'text': event[0]['body']
        }
    }
    return event[0]
# => {
#  (元々のevent),
#  '_enrichment': {
#    'slack': {
#      'channel': 'XXX',
#      'text': 'STRING....'
#     }
#  }
#}

Input transformar

{
  "channel": <$._enrichment.slack.channel>,
  "text": <$._enrichment.slack.text>
}

Conclusion

Amazon EventBridge Pipes allow AWS Lambda functions to be specified in the target. Therefore, some may argue that there is no reason for Enrichment to handle Lambda functions with
It appears that there is no need to use a Lambda function in the Enrichment.

Eliminating the implementation of calling external APIs from within the Lambda function would have the advantage of simplifying the Lambda function.

EoT

AWS IoT EduKit is what? What can we learn from it?

Kohei (Max) MATSUSHITA — Sat, 13 Feb 2021 01:13:21 +0000

AWS IoT EduKit is a new device from AWS that was launched by AWS re:Invent 2020. Introduce what do we learn from the kit.

2 Benefits, 2 Learn points.

Eliminates the need to create client certificates and operate CA for IoT core connectivity
Eliminates the need for measures to prevent leakage of the private key
How to register to IoT Core with the client certificate in ATECC608A Trust&GO
How to use the private key in the ATECC608A Trust&GO when connecting to IoT Core

Please see the following blog for details.

DEV Community: Kohei (Max) MATSUSHITA

How to set up cross-account access to Amazon Kinesis Data Streams

Overview - Amazon Kinesis Data Streams and separated AWS accounts

The Architecture and Setup

Key Points for Setup

Steps

Step 1) On the data producer side - Account A

1-1. Create a data stream in Amazon Kinesis (e.g., kds-sharing-example1)

Step 2) On the data processing side - Account B

2-1. Create a Lambda function in AWS Lambda (e.g., kds-reader1).

2-2. Attach the AWSLambdaKinesisExecutionRole policy to the execution role of kds-reader1.

2-3. Noted the ARN of the IAM role.

Step 3) Back on the data producer side - Account A

3-1. configure resource-based policy for kds-sharing-example1 in Amazon Kinesis

3-2. Noted the ARN of kds-sharing-example1.

Step 4) Back on the data processing side - Account B

How to check

Epilogue - Architecture with Amazon EventBridge

Predictive Maintenance of Industrial machine with Amazon Monitron and What is “ISO 20816” ?

AWS CDK stack for building Amazon EventBridge Pipes (SQS to CloudWatch Logs)

ma2shita / cdk-eventbridge-pipes-simplelogger

Example an AWS CDKv2 for building the Amazon EventBridge Pipes that logs messages sent to an SQS queue to CloudWatch Log.

AWS CDKv2 example: SQS to CloudWatch Logs (w/ Enrichment by Lambda) on Amazon EventBridge Pipes

Requirements

Build & Deploy

How it works

Contains "enrichment" Lambda functions

Customization Points

On the "Enrichment" function

On the "Target" of EventBridge Pipes

Conclusion and "CDKv2 for IoT 1-Click"

ma2shita / cdk-iot1click-project-to-sqs

Example of an AWS CDKv2 for building the AWS IoT 1-Click project with sending to exists SQS queue Lambda function.

AWS CDKv2 example: IoT 1-Click to exists SQS queue

Requirements

Build & Deploy

How it works

AWS Lambda function specifications in "Enrichment" in Amazon EventBridge Pipes

TL;DR

Explanation

Input to the function of "Enrichment"

Output from the function of "Enrichment"

Implicit body data parsing

My strategy for implementing Enrichment Lambda functions

Approach 1: "Rewriting."

Approach 2: "Adding."

Conclusion

AWS IoT EduKit is what? What can we learn from it?

AWS IoT EduKit is what? What can we learn from it? | Medium

Kohei "Max" Matsushita ・ Feb 12, 2021 ・ ma2shita.Medium

1-1. Create a data stream in Amazon Kinesis (e.g., `kds-sharing-example1`)

2-1. Create a Lambda function in AWS Lambda (e.g., `kds-reader1`).

2-2. Attach the `AWSLambdaKinesisExecutionRole` policy to the execution role of `kds-reader1`.

3-1. configure resource-based policy for `kds-sharing-example1` in Amazon Kinesis

3-2. Noted the ARN of `kds-sharing-example1`.

Kohei "Max" Matsushita ・ Feb 12, 2021 ・
ma2shita.Medium