使用 API Gateway 监管客户端与 Apache Kafka 之间的交互

Original URL:https://aws.amazon.com/blogs/big-data/govern-how-your-clients-interact-with-apache-kafka-using-api-gateway/

有时，您可能会自问：

如何为 Amazon Managed Streaming for Apache Kafka (MSK) 实施 IAM 身份验证或授权？
如何根据特定的场景保护我的 Apache Kafka 集群，防止流量峰值的影响，同时又无需设置集群的配额？
如何验证请求是否符合 JSON Schema？
如何确保 URI、查询字符串和标头中包含参数？
Amazon MSK 如何在不需要代理或原生 Apache Kafka 协议的情况下提取消息轻量客户端？

这些需求可使用自定义的代理服务器或网关来实现，但都可能难以实施和管理。而 API Gateway 具有所有这些功能并且是一项完全托管的 AWS 服务。

在本文中，我们将演示 Amazon API Gateway 如何作为 Amazon MSK 集群和客户端之间的一个组件来解决这些问题。

Amazon MSK 是一款适用于 Apache Kafka 的完全托管的服务，借助它只需几次点击即可轻松预置 Kafka 集群，不再需要手动预置服务器、管理存储或者配置 Apache Zookeeper。Apache Kafka 是一个开源平台，可用于构建实时的流式处理数据管道和应用程序。

本文讨论的一些场景包括从不支持原生 Kafka 协议的轻量级 IoT 设备中提取消息，以及将您的流式处理服务与其他后端服务（包括第三方 API）进行编排。

在进行网关架构设计的时候还需要权衡考虑以下几个因素：

要运行和维护其他服务的成本与复杂性。
由于增加额外的处理能力来构建和提出 HTTP 请求导致的性能开销。此外，REST 代理需要解释请求，为生产请求和消费请求执行数据格式转换。

当您在生产环境中实施这种架构时，您应结合您的使用业务使用案例以及 SLA 需求综合考虑。

解决方案概览

要实施此解决方案，请完成下列步骤：

创建 MSK 集群、Kafka 客户端和 Kafka REST 代理
在 Kafka 客户端机上创建 Kafka 主题并配置 REST 代理
通过 API Gateway 创建与 REST 代理集成的 API
通过向Amazon MSK 生产与消费消息的端到端的测试

下图演示了解决方案的架构。

在此架构中，您将创建一个 MSK 集群，并使在一台 Amazon EC2 实例上配置好Kaflka客户端和REST代理。然后，您将通过 Amazon API Gateway 暴露该 REST 代理，还将使用 Postman 向 Amazon MSK 生成消息以测试此解决方案。

对于生产实施，请务必将REST代理部署在负载均衡器后面的 Auto Scaling 组当中。

先决条件

在开始操作前，您必须满足以下先决条件：

拥有一个具有 AWS 服务访问权限的 AWS 账户。
拥有一个具有访问密钥和秘密访问密钥的 IAM 用户，用于配置 AWS CLI。
一个 Amazon EC2 密钥对

创建 MSK 集群、Kafka 客户端和 REST 代理

AWS CloudFormation 将会预置所有需要的资源，包括 VPC、子网、安全组、Amazon MSK 集群、Kafka 客户端和 Kafka REST 代理。要创建这些资源，请完成以下步骤：

在 us-east-1 或 us-west-2 区域启动完成此过程需要大约 15 到 20 分钟。
在 AWS CloudFormation 控制台上，选择 AmzonMSKAPIBlog。
记下输出下的 MSKClusterARN、KafkaClientEC2InstancePublicDNS 和 MSKSecurityGroupID 详细信息。

输入以下代码，获取有关集群的 ZooKeeperConnectionString 和其他信息（须提供您的区域、集群 ARN 和 AWS 指定的配置文件）：

$ aws kafka describe-cluster --region <Replace_With_us-east-1_or_us-west-2> --cluster-arn <Replace_With_Your_cluster-arn> --profile <Replace_With_Your_Profile>

下面的示例代码显示了此命令输出中的一行：

{
….
….
"ZookeeperConnectString": "z-2.XXXXXX.us-east-1.amazonaws.com:2181,z-3.XXXXXX.us-east-1.amazonaws.com:2181,z-1.XXXXXX.us-east-1.amazonaws.com:2181"
}

输入以下代码（须提供您的区域、集群 ARN 和 AWS 指定的配置文件），获取 BootstrapBrokerString：

$ aws kafka get-bootstrap-brokers --region <Replace_With_us-east-1_or_us-west-2> --cluster-arn "<Replace_With_us-east-1_or_us-west-2>" --profile <Replace_With_Your_Profile>

下面的示例代码显示了此命令的输出：

{
"BootstrapBrokerString": "b-2.XXXXXXXXXXXX.us-east-1.amazonaws.com:9092,b-1.XXXXXXXXXXXX.amazonaws.com:9092,b-3.XXXXXXXXXXXX.us-east-1.amazonaws.com:9092"
}

创建 Kafka 主题并配置 Kafka REST 代理

要在 Kafka 客户端机上创建 Kafka 主题并配置 Kafka REST 代理，请完成以下步骤：

通过 SSH 登录您的 Kafka 客户端 Amazon EC2 实例。请参见以下代码：
```
ssh -i <Replace_With_Your_pemfile> ec2-user@<Replace_With_Your_KafkaClientDNS>
```
在客户端机上进入 Apache Kafka 安装的二进制包文件夹 (kafka/kafka_2.12-2.2.1/bin/)。
通过输入以下代码创建主题（须提供上一步中取得的 ZookeeperConnectString 值）：
```
./kafka-topics.sh --create --zookeeper <Replace_With_Your_ZookeeperConnectString> --replication-factor 3 --partitions 1 --topic amazonmskapigwblog
```
如果命令成功执行，您将看到以下消息：已创建主题 amazonmskapigwblog。
要将 Kafka REST 服务器连接到 Amazon MSK 集群，请修改目录 (/home/ec2-user/confluent-5.3.1/etc/kafka-rest/) 中的 kafka-rest.properties，以指向您的 Amazon MSK 的 ZookeeperConnectString 和 BootstrapserversConnectString 信息。请参见以下代码：
```
sudo vi /home/ec2-user/confluent-5.3.1/etc/kafka-rest/kafka-rest.properties

	zookeeper.connect=<Replace_With_Your_ZookeeperConnectString>
bootstrap.servers=<Replace_With_Your_BootstrapserversConnectString> 
```
此外（此步骤可选），您还可以创建一个 SSL 以保护 REST 客户端与 REST 代理之间的通信 (HTTPS)。如果不需要 SSL，您可以跳过第 5 步和第 6 步。
生成服务器和客户端证书。有关更多信息，请参阅 Confluent 网站上的创建 SLL 密钥和证书。

将必要的属性配置添加到 kafka-rest.properties 配置文件。请参见以下代码示例：

listeners=http://0.0.0.0:8082,https://0.0.0.0:8085
ssl.truststore.location=<Replace_With_Your_tuststore.jks>
ssl.truststore.password=<Replace_With_Your_tuststorepassword>
ssl.keystore.location=<Replace_With_Your_keystore.jks>
ssl.keystore.password=<Replace_With_Your_keystorepassword>
ssl.key.password=<Replace_With_Your_sslkeypassword>

有关更多详细说明，请参阅 Confluent 网站的使用 SSL 加密和验证身份。

您现在可以创建 Kafka 主题并配置 Kafka REST 代理，以连接到您的 Amazon MSK 集群。

创建集成了 Kafka REST 代理的 API

要通过 API Gateway 创建集成了 Kafka REST 代理的 API，请完成以下步骤：

打开 API Gateway 控制台，然后选择创建 API。
对于 API 类型，选择 REST API。
选择构建。
选择新 API。
对于 API 名称，输入一个名称（例如 amazonmsk-restapi）。
（可选步骤）对于描述，输入简要描述。
选择创建 API。下一步将是创建子资源。
在资源下，选择父资源项目。
在操作下，选择创建资源。这时将显示新子资源窗格。
选择配置为代理资源。
对于资源名称，输入 proxy。
对于资源路径，输入 /{proxy+}。
选择启用 API Gateway CORS。
选择创建资源。创建好资源后，将会显示创建方法。
对于集成类型，选择 HTTP 代理。
对于终端节点 URL，输入一个 HTTP 后端资源 URL（Kafka 客户端 Amazont EC2 实例的 PublicDNS；例如 http://KafkaClientEC2InstancePublicDNS:8082/{proxy} 或rhttps://KafkaClientEC2InstancePublicDNS:8085/{proxy}）。
其他字段使用默认设置。
选择保存。
对于 SSL，终端节点 URL 将使用 HTTPS 终端节点。在您刚刚创建的 API 中，API 的代理资源路径 {proxy+} 将成为 http://YourKafkaClientPublicIP:8082/ 下任何后端终端节点的占位符。
选择您刚刚创建的 API。
从操作下选择部署 API。
对于部署阶段，选择新阶段。
对于阶段名称，输入阶段的名称（例如 dev、test 或 prod）。
选择部署。
在部署好 API 后记下调用 URL。

您的外部 Kafka REST 代理通过 API Gateway 暴露，类似于 https://YourAPIGWInvoleURL/dev/topics/amazonmskapigwblog。您将在下一步中使用此 URL。

端到端的过程测试

通过生成并向 Amazon MSK 发送消息进行端到端的过程测试。请执行以下步骤：

通过 SSH 登录 Kafka 客户端 Amazon EC2 实例。请参见以下代码：
```
ssh -i "xxxxx.pem" ec2-user@KafkaClientEC2InstancePublicDNS
```

进入 confluent-5.3.1/bin 目录并启动 kafka-rest 服务。请参见以下代码：

./kafka-rest-start /home/ec2-user/confluent-5.3.1/etc/kafka-rest/kafka-rest.properties

如果服务已经启动，您可以使用下面的代码停止服务：

./kafka-rest-stop /home/ec2-user/confluent-5.3.1/etc/kafka-rest/kafka-rest.properties

打开新的终端窗口。
在 kafka/kafka_2.12-2.2.1/bin 目录中，启动 Kafka 控制台用户。请参见以下代码：
```
./kafka-console-consumer.sh --bootstrap-server "BootstrapserversConnectString" --topic amazonmskapigwblog --from-beginning 
```
您现在可以使用 Postman 生成消息。Postman 是一种用于测试 Web 服务的 HTTP 客户端。

务必在Kafka客户端安全组入站规则中允许源自 Postman 请求的IP与 TCP 端口。
在标头下，键选择 Content-Type，值选择 application/vnd.kafka.json.v2+json.
在正文下，选择原始。
选择 JSON。本文输入了以下代码：
```
{"records":[{"value":{"deviceid": "AppleWatch4","heartrate": "72","timestamp":"2019-10-07 12:46:13"}}]} 
```
以下屏幕截图显示了从 API Gateway Kafka REST 终端节点发送到 Kafka 消费者的消息。