作者：SRE运维博客
博客地址：https://www.cnsre.cn/
文章地址：https://www.cnsre.cn/posts/230410659053/
相关话题：https://www.cnsre.cn/tags/lambda/
Lambda函数检查S3文件夹是否存在
作为 AWS 中最常用的对象存储服务，S3 可以用于存储各种类型的文件，包括网站文件、媒体文件、备份文件等等。在 S3 中存储的文件可以通过不同的方式访问，例如在 Web 应用程序中、通过移动应用程序或直接使用 AWS SDK 访问等。
在进行 S3 存储时，如果我们需要将存储的日志同步到另一个桶或区域中，则可以使用 AWS 的 S3 日志同步任务功能。通过将日志同步到其他存储桶或区域中，我们可以更方便地对日志进行分析、监控和管理。
但是，如果 S3 日志同步任务出现故障，我们可能无法及时获取相关的日志信息。因此，为了确保日志同步任务的正常运行，我们需要对任务进行监控。在本文中，我们将介绍如何使用 AWS Lambda 监控 S3 日志同步任务。
介绍
AWS Lambda 是一种无服务器计算服务，可使您在云中运行代码，而无需自己管理服务器。通过使用 Lambda，您可以将代码上传到云中，然后 Lambda 会根据需要自动扩展和缩减计算资源，以满足您的应用程序的请求。Lambda 还支持许多编程语言和库，使您能够编写功能强大的应用程序和服务。
在本文中，我们将使用 Lambda 编写一个函数，该函数将定期检查 S3 存储桶中的文件夹是否存在。如果不存在任何文件夹，则 Lambda 将向指定的 SNS 主题发送一条消息，以便管理员可以及时采取措施。通过使用 Lambda 监控 S3 存储桶中的文件夹，我们可以确保日志同步任务的正常运行。
1. 准备工作
在开始之前，我们需要先准备好以下工作：
- 一个S3桶，用于存储我们要检查的文件夹。
- 一个SNS主题，用于发送消息提醒。
2. 创建Lambda函数
在AWS控制台上创建一个Python Lambda函数，名称为s3-folder-exist-checker，并使用以下代码：
import boto3
from datetime import datetime, timedelta
from dateutil import tz
def lambda_handler(event, context):
print('Lambda 函数已启动.')
s3 = boto3.resource('s3')
bucket_name = 'my_s3_bucket_name'
local_tz = tz.gettz('Asia/Shanghai')
now = datetime.now()
date_prefix = now.strftime('%Y/%m/%d/')
folder_prefixes = ['my_prefixes/' + date_prefix, 'my_prefixes/' + date_prefix, 'RGC-Prod-3in1oven/' + date_prefix]
folder_prefixes = [prefix + '/' if not prefix.endswith('/') else prefix for prefix in folder_prefixes] # 确保每个前缀以斜杠结尾
print('正在检查以下 S3 文件夹：', folder_prefixes)
sns = boto3.client('sns')
topic_arn = 'arn:aws-cn:sns:cn-north-1:1234567890:s3-logs-monitoring'
for prefix in folder_prefixes:
resp = s3.meta.client.list_objects_v2(Bucket=bucket_name, Prefix=prefix, Delimiter='/')
subfolders = [p['Prefix'] for p in resp.get('CommonPrefixes', [])]
if len(subfolders) > 0:
print(f"子文件夹 '{prefix}' 存在:")
for folder in subfolders:
print(f"发现子文件夹: {folder}")
else:
message = f"S3桶'{bucket_name}中'{prefix}'下不存在新增文件夹,即日志同步S3桶任务失败.请检查.'"
sns.publish(TopicArn=topic_arn, Message=message)
print(f"已发送 SNS 消息: {message}")
print('Lambda 函数已完成.')
return {
'statusCode': 200,
'body': 'S3 文件夹存在性检查已完成.'
}
接下来是对 Lambda 函数中的一些细节进行讲解。首先，我们定义了 S3 的资源，并且指定了 S3 桶的名称：
s3 = boto3.resource('s3')
bucket_name = 'my_bucket_name'
接着，我们获取当前的时间，并且根据当前时间生成一个目录前缀。我们使用 dateutil 模块中的 tz.gettz 函数来获取一个本地的时区信息。为了确保时区的准确性，我们建议在使用 Lambda 函数时，显式地设置时区信息：
local_tz = tz.gettz('Asia/Shanghai')
now = datetime.now(local_tz)
date_prefix = now.strftime('%Y/%m/%d/')
在 Lambda 函数中，我们使用了 list_objects_v2 方法来列举指定的文件夹。具体来说，我们使用了 CommonPrefixes 参数，该参数可以返回指定前缀下的子文件夹列表。如果返回的子文件夹列表为空，则说明指定的文件夹不存在。如果子文件夹列表不为空，则说明文件夹存在，并且我们可以将每个子文件夹的路径打印出来：
resp = s3.meta.client.list_objects_v2(Bucket=bucket_name, Prefix=prefix, Delimiter='/')
subfolders = [p['Prefix'] for p in resp.get('CommonPrefixes', [])]
if len(subfolders) > 0:
print(f"子文件夹 '{prefix}' 存在:")
for folder in subfolders:
print(f"发现子文件夹: {folder}")
如果子文件夹列表为空，则说明文件夹不存在。在这种情况下，我们可以使用 SNS 服务发送一条消息来通知管理员：
if len(subfolders) == 0:
message = f"S3桶'{bucket_name}中'{prefix}'下不存在新增文件夹,即日志同步S3桶任务失败.请检查.'"
sns.publish(TopicArn=topic_arn, Message=message)
print(f"已发送 SNS 消息: {message}")
最后，我们返回了一个包含状态码和消息的字典，以便可以在 Lambda 函数执行过程中监控执行状态：
return {
'statusCode': 200,
'body': 'S3 文件夹存在性检查已完成.'
}
确保您的 Lambda 函数有权限访问 S3 和 SNS
-
在 AWS Lambda 控制台中，创建一个新的 Lambda 函数。在函数代码中将 Python 代码粘贴到代码编辑器中。请确保您选择了正确的运行时环境，并设置以下环境变量：
-
BUCKET_NAME：您的 S3 桶名称 -
SNS_TOPIC_ARN：SNS 主题的 ARN
-
配置 Lambda 函数的基本设置和高级设置，包括内存和超时。
在 Lambda 控制台中，测试 Lambda 函数，以确保 Lambda 函数能够访问 S3 桶和 SNS 主题。为了测试该函数，您可以创建一个测试事件，该事件需要一个空的 JSON 对象，例如：
最后，您需要在 Amazon CloudWatch 中设置 CloudWatch Events 规则以定期触发 Lambda 函数。这样您的 Lambda 函数就能在您预定的时间检查 S3 文件夹是否存在并发送通知。
总结
在这篇文章中，我们介绍了一个使用 AWS Lambda、S3 和 SNS 的自动化任务，该任务定期检查 S3 文件夹是否存在并发送通知。我们解释了如何编写 Python 代码来实现此任务，并提供了一个详细的代码示例。我们还介绍了如何在 AWS Lambda 和 Amazon SNS 控制台上配置 Lambda 函数和 SNS 主题，并在 Amazon CloudWatch 中创建定期触发器来触发 Lambda 函数。最后，我们提供了一些最佳实践和注意事项，以确保您的 Lambda 函数和 SNS 主题能够正常工作。
希望这篇文章对您有所帮助！如果您有任何疑问或建议，请在下面的评论区留言。
参考文献
作者：SRE运维博客
博客地址：https://www.cnsre.cn/
文章地址：https://www.cnsre.cn/posts/230410659053/
相关话题：https://www.cnsre.cn/tags/lambda/
