资讯中心

首页
将 Amazon CloudWatch 日志传送到 Amazon OpenSearch Server

将 Amazon CloudWatch 日志传送到 Amazon OpenSearch Server

将Amazon CloudWatch日志发送到Amazon OpenSearch Serverless

关键要点

本文介绍了如何通过AWS架构将CloudWatch日志实时传送到Amazon OpenSearch Serverless，本文将阐述整个流程及其各个组成部分。

在现代云架构中，Amazon CloudWatch 提供了一个高效的方式来收集和分析来自不同系统的日志数据。它的订阅功能允许用户将这些日志数据实时传输到其他AWS服务，例如 Amazon Kinesis Data Streams、 AWS Lambda，以及 Amazon OpenSearch Service。然而，截止到本文发布时，这些订阅过滤器只支持将日志发送到 Amazon OpenSearch Service 的预置集群。越来越多的客户开始采用 Amazon OpenSearch Serverless，作为一种具备成本效益的解决方案，适用于不频繁、间歇性和不可预测的工作负载。

在本文中，我们将展示如何使用 Amazon OpenSearch Ingestion 将CloudWatch日志几乎实时地交付到OpenSearch Serverless，概述连接Lambda订阅过滤器与OpenSearch Ingestion的机制，并不用为其明确创建单独的订阅过滤器。

解决方案概览

以下图示描绘了该解决方案的架构。

CloudWatch Logs：收集并存储来自各种AWS资源和应用程序的日志，作为本解决方案中的日志数据源。订阅过滤器 CloudWatch Logs 订阅过滤器过滤并将特定日志数据路由到管道中的下一个组件。CloudWatch 导出Lambda函数此Lambda函数接收来自订阅过滤器的已过滤日志数据，旨在转化并准备日志数据以便传送到OpenSearch Ingestion管道。OpenSearch Ingestion OpenSearch Service的一个组成部分，负责处理和丰富从CloudWatch导出Lambda函数接收的日志数据，最终存储在OpenSearch Serverless集合中。OpenSearch Service 一个完全托管的服务，储存和索引日志数据，使其可搜索并便于分析和可视化。OpenSearch Service提供两种配置：预置域和无服务器。在本设置中，我们使用无服务器的配置，这是一种为OpenSearch Service自动扩展的方式。

前提条件

一个AWS账户在您AWS环境中设置的CloudWatch日志创建的OpenSearch Serverless集合VPC和子网配置

部署解决方案

所有前提条件就绪后，您可以创建并部署解决方案的各个组件。

步骤1：为引入创建PipelineRole

打开AWS管理控制台。选择策略，然后选择创建策略。选择JSON并粘贴以下策略：

json{ Version 20121017 Statement [ { Action [ aossBatchGetCollection aossAPIAccessAll ] Effect Allow Resource arnawsaossuseast1{accountId}collection/{collectionId} } { Action [ aossCreateSecurityPolicy aossGetSecurityPolicy aossUpdateSecurityPolicy ] Effect Allow Resource Condition { StringEquals { aosscollection {collection} } } } ]}

将 Amazon CloudWatch 日志传送到 Amazon OpenSearch Server

注意请用您的信息替换 {accountId}、 {collectionId} 和 {collection}。

npv下载选择下一步，命名策略为collectionpipelinepolicy。选择创建策略。接下来，创建一个角色并将策略附加到它。选择角色，然后选择创建角色。选择自定义信任策略并粘贴以下策略：

json{ Version20121017 Statement[ { EffectAllow Principal{ Serviceosispipelinesamazonawscom } ActionstsAssumeRole } ]}

选择下一步，然后搜索并选择刚刚创建的collectionpipelinepolicy。选择下一步，命名该角色为PipelineRole。选择创建角色。

步骤2：配置OpenSearch集合的网络和数据策略

在OpenSearch Service控制台中，导航到无服务器菜单。按照为OpenSearch Serverless创建接口端点的说明创建VPC端点。转到安全性，选择网络策略。选择创建网络策略。配置以下策略：

json[ { Rules [ { Resource [ collection/{collection name} ] ResourceType collection } ] AllowFromPublic false SourceVPCEs [ {VPC Endpoint Id} ] } { Rules [ { Resource [ collection/{collection name} ] ResourceType dashboard } ] AllowFromPublic true }]

转到安全性，选择数据访问策略。选择创建访问策略。配置以下策略：

json[ { Rules [ { Resource [ index/{collection name}/ ] Permission [ aossCreateIndex aossUpdateIndex aossDescribeIndex aossReadDocument aossWriteDocument ] ResourceType index } ] Principal [ arnawsiam{accountId}role/PipelineRole arnawsiam{accountId}role/Admin ] Description 规则1 }]

步骤3：创建OpenSearch Ingestion管道

导航到OpenSearch Service。转到Ingestion管道部分。选择创建管道。定义管道配置：

yamlversion 2 cwlogsingestionpipeline

source

 http   path /logs/ingest

sink

  opensearch     # 提供AWS OpenSearch Service域端点     hosts [https//{collectionId}{region}aossamazonawscom]     index cwl{yyyyMMdd}     aws       # 提供一个可以访问该域的Role ARN。该角色应与osispipelinesamazonawscom有信任关系       stsrolearn arnawsiam{accountId}role/PipelineRole       # 提供域的区域。       region {region}       serverless true       serverlessoptions         networkpolicyname {Network policy name}

获取占位符值的方法1 {collectionId} 通过在AWS管理控制台中导航到Amazon OpenSearch Serverless集合并单击该集合，collection ID会显示在“概述”部分下。2 {region} 这是您Amazon OpenSearch Service域所在的AWS区域。在AWS管理控制台中导航到域时，可以找到此信息。3 {accountId} 这是您的AWS账户ID。在AWS管理控制台右上角单击用户名，选择“我的账户”即可找到账户ID。4 {Network policy name} 这是您为Amazon OpenSearch Serverless集合配置的网络策略名称。如果尚未配置网络策略，可以保留此占位符，或从配置中删除它。

步骤4：创建Lambda函数

创建一个用于请求和sigv4包的Lambda层。在AWS CloudShell中运行以下命令：

bashmkdir lambdalayerscd lambdalayersmkdir pythoncd pythonpip install requests t /pip install requestsauthawssigv4 t /cd zip r pythonmoduleszip aws lambda publishlayerversion layername Datarequests description My Python layer zipfile fileb//pythonmoduleszip compatibleruntimes python3x

创建Python 3x运行环境的函数。请参见创建您的第一个Lambda函数。

pythonimport base64import gzipimport jsonimport loggingimport jmespathimport requestsfrom datetime import datetimefrom requestsauthawssigv4 import AWSSigV4import boto3

LOGGER = logginggetLogger(name)LOGGERsetLevel(loggingINFO)

def lambdahandler(event context)

提取事件中的数据data = jmespathsearch(awslogsdata event)解压日志cwLogs = decompressjsondata(data)构建要发送到OpenSearch Ingestion的有效载荷payload = preparepayload(cwLogs)print(payload)将事件集传送到管道response = ingestData(payload)return {    statusCode 200}

def decompressjsondata(data)

compresseddata = base64b64decode(data)uncompresseddata = gzipdecompress(compresseddata)return jsonloads(uncompresseddata)

def preparepayload(cwLogs)

payload = []logEvents = cwLogs[logEvents]for logEvent in logEvents    request = {}    request[id] = logEvent[id]    dt = datetimefromtimestamp(logEvent[timestamp] / 1000)    request[timestamp] = dtisoformat()    request[message] = logEvent[message]    request[owner] = cwLogs[owner]    request[loggroup] = cwLogs[logGroup]    request[logstream] = cwLogs[logStream]    payloadappend(request)return payload

def ingestData(payload)

ingestionEndpoint = {OpenSearch Pipeline Endpoint}endpoint = https//  ingestionEndpointheaders = {ContentType application/json Acceptapplication/json}r = requestsrequest(POST f{endpoint}/logs/ingest json=payload auth=AWSSigV4(osis) headers=headers)LOGGERinfo(收到的响应   rtext)return r

将 {OpenSearch Pipeline Endpoint} 替换为您OpenSearch Ingestion管道的端点。在执行角色中附加以下内联策略：

json{ Version 20121017 Statement [ { Sid PermitsWriteAccessToPipeline Effect Allow Action osisIngest Resource arnawsosis{region}{accountId}pipeline/{OpenSearch Pipeline Name} } ]}

部署该函数。

步骤5：设置CloudWatch Logs订阅

授权特定AWS服务或AWS账户调用指定的Lambda函数。以下命令授权CloudWatch Logs服务调用指定日志组的cloudlogs Lambda函数，这是因为CloudWatch Logs无法直接调用Lambda函数，必须先授予权限。在CloudShell中运行以下命令以添加权限。

bashaws lambda addpermission functionname {function name} statementid {function name} principal logsamazonawscom action lambdaInvokeFunction sourcearn arnawslogs{region}{accountId}loggroup{loggroup} sourceaccount {accountId}

为日志组创建订阅过滤器。以下命令在日志组上创建订阅过滤器，将所有日志事件因为过滤器模式为空字符串转发到Lambda函数。在CloudShell中运行以下命令以创建订阅过滤器。

bashaws logs putsubscriptionfilter loggroupname {loggroup} filtername {filter name} filterpattern destinationarn arnawslambda{region}{accountId}function{function name}

步骤6：测试和验证

在您的CloudWatch日志组中生成一些日志。在CloudShell中运行以下命令以在日志组中创建示例日志。

bashaws logs putlogevents loggroupname {loggroup} logstreamname {streamname} logevents [{timestamp{timestamp in millis} message 简单的Lambda测试}]

检查OpenSearch集合，确保日志已正确索引。

清理

在不使用时，请移除本解决方案的基础设施，以避免不必要的费用。

结论

您已经了解了如何设置管道，将CloudWatch日志发送到VPC中的OpenSearch Serverless集合。此集成利用CloudWatch进行日志聚合，利用Lambda进行日志处理，并通过OpenSearch Serverless进行查询和可视化。通过使用OpenSearch Serverless的按需计费模型，您可以优化日志分析的运营成本。

想要进一步探索，可以：

了解更多关于在OpenSearch Dashboards中查询和可视化日志数据。将其他日志源整合，如EC2实例或容器日志，整合到同一管道中。基于日志模式或异常设置警报和通知规则。