明察秋毫 — 轻松实现 S3 用量细颗粒度追踪

互联网发展至今，很多企业意识到云带来的优势可以让他们将精力放在发展业务上而不需要过多担心软件和底层的运维。在这样的潮流下，云上共享存储的场景越来越被使用到，其中 SaaS 化的多租户概念则是非常典型的使用模式。对于 SaaS 服务供应商来说，如何实现租户之间的数据隔离、服务的权限控制、计费管理等，成为了其构建用户系统最核心的问题之一。本文旨在探讨如何利用 S3 服务功能，在不改变现有工作流程和数据集的情况下对请求者级别的用量进行追踪。

1. 背景介绍

S3 作为 SaaS 服务常用的存储服务之一，现有的功能很方便地支持了单租户或多租户的数据隔离和权限控制。大致上有三种模式：

桶级别租户模型

前缀级别租户模型

数据库匹配租户模型

S3 的成本构成中，主要的存储、请求&检索、数据传输产生的费用会由租户的具体用量决定，因此把请求和数据传输用量追踪起来有助于 SaaS 服务供应商制定合理的收费策略和服务套餐体系。在这三种租户隔离模式下，虽然 S3 仍可以通过打标签与清单报告搭配来精细化管理存储用量和权限控制，但是对于具体的请求用量和数据传输用量尚没有很好的追踪办法。

那我们来看看，有哪些现有的服务或者功能可以被利用起来：

可以看到亚马逊对于 S3 服务提供了多种监控工具，并且各自都有适合的场景。针对具体到请求者级别的信息粒度，对比下来只有 S3 服务器访问日志能够达到我们的需求，但是仅仅依靠原始的日志功能是不够的。

2. 方案概述

主要组件

数据来源— S3 服务器访问日志/清单报告
存储— S3 桶
处理— Lambda + Athena
可视化— QuickSight

工作流程

S3 的服务器访问日志记录提供对 S3 存储桶发出的请求的详细记录，因此可以利用这些日志来分析具体租户的 API 请求和数据传输量。其中唯一标识字段 Requester、Request-URI、Access Point ARN 等能够直接通过查询分析日志来定位到具体用户。Lambda 首次触发创建 Athena 的视图对日志信息进行筛选分析，后续定时触发更新表单中的变量。最后使用 QuickSight 模版将 Athena 中的视图进行可视化，生成请求数及数据传输的监控看板。

需要注意的是， Requester 、Request-URI、Access Point ARN 这些字段需要分别搭配 IAM 权限隔离、前缀管理、访问点等先决条件才能提供有效追踪。如果现有服务没有使用到上述任何一种功能，那么还可以通过自定义访问日志信息功能对 S3 日志注入查询字符串来实现。您可以将 S3 的自定义查询字符串参数添加到请求的 URL 中，而 S3 服务器在处理请求时会忽略以“x-”开头的查询字符串参数，但是会将这些参数包含在请求的访问日志记录中，以作为日志记录的 Request-URI 字段的一部分。这个经常被忽略的功能成为我们用量追踪方案的突破点。

由于 S3 扁平化的架构，任何在请求路径中携带有查询字符串会被服务器认为是专有对象名，因此我们无法简单地将查询字符作为对象名的后缀。然而现有的 SDK 中，还没有专门的参数支持对查询字符串的写入，所以可行的方法只有在 SDK 在往服务节点发送 HTTP 请求前进行注入，同时还要生成携带有查询字符串的签名以通过 S3 服务器的验证。

3. 实战演练

演练中假设存在两个租户团队 Tenant A 和 Tenant B 中各有多个用户在访问对应权限范围内的文件，因此看板中需要两个租户团体的用户请求用量及存储使用量数据；应用端会使用“x-user”开头的查询字符串来将用户信息注入请求。

3.1 数据源建表

S3 服务器日志与清单报告具有固定的格式，因此创建对应的 Athena 表格可以分别参考官方文档–使用 Amazon Athena 查询访问日志中的请求与使用 Amazon Athena 查询 Amazon S3 清单。

3.2 Lambda 创建/更新视图

Lambda 会临时从 Parameter Store 获取需要查询的用户名单，经处理后生成初步的查询语句来更新 Athena 中对于 S3 服务器访问日志的视图：

import time
import boto3
import logging
import json
from datetime import date
from datetime import timedelta

logger = logging.getLogger()
logger.setLevel(logging.INFO)  
          
DATABASE = 'example_db_name'
output='s3://example-bucket/athena-output/'

# fetch user list from parameter store
ssm = boto3.client('ssm', 'us-west-2')
def get_parameters():
    response = ssm.get_parameters(
        Names=['accesslog_user_list'],WithDecryption=True
    )
    value=[]
    for parameter in response['Parameters']:
        return parameter['Value']

# main code to run query template and update Athena view
def lambda_handler(event, context):
    value = get_parameters().replace("'","")
    username= value.split(",")
    step = 'update x-user view'
    logger.info('Trying to %s',step)
    query ="""
    CREATE OR REPLACE VIEW x_user_requester AS
    SELECT REPLACE(request_uri,request_uri,'test') AS requester,bytessent
    FROM "mybucket_logs_for_ab2"
    WHERE requestdatetime LIKE '%/Oct/2022%'
    AND operation LIKE '%GET%'
    AND request_uri LIKE '%x-user=test%'
    """
    for name in username:
        query+=f"""UNION ALL
    SELECT REPLACE(request_uri,request_uri,'{name}'),bytessent
    FROM "mybucket_logs_for_ab2"
    WHERE requestdatetime LIKE '%/Oct/2022%'
    AND operation LIKE '%GET%'
    AND request_uri LIKE '%x-user={name}%'
    """
    client = boto3.client('athena')
    # Execution
    response = client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={
            'Database': DATABASE
        },
        ResultConfiguration={
            'OutputLocation': output,
        }
    )
    logger.info(f"Step {step} completed")

Lambda 继续更新针对清单报告生成查询语句来更新 Athena 视图。其中引入了前缀层级划分来统计前缀级别的存储用量：

    # get the date
    today = date.today()
    yesterday =str(today - timedelta(days = 1))
    
    step = 'update tenantA view'
    logger.info('Trying to %s',step)
    logger.info('Today is %s',today)
    querya=f"""
    CREATE OR REPLACE VIEW tenant_a_prefix_usage AS
    SELECT prefix_depth, common_prefix, sum(size) AS Total_Bytes
    FROM (SELECT prefix_depth, key,
        CASE CARDINALITY(SPLIT(key, '/'))
            WHEN prefix_depth THEN key
            ELSE ARRAY_JOIN(SLICE(SPLIT(key, '/'), 1, prefix_depth), '/') || '/'
        END AS common_prefix,
             size
    FROM inventory_test, UNNEST(SEQUENCE(1, 2)) d(prefix_depth) 
    WHERE dt = '{yesterday}-00-00' AND size <> 0)
    GROUP BY 1, 2 ORDER BY 1 ASC, 2 DESC
    """
    # Execution
    responsea = client.start_query_execution(
        QueryString=querya,
        QueryExecutionContext={
            'Database': DATABASE
        },
        ResultConfiguration={
            'OutputLocation': output,
        }
    )
    logger.info(f"Step {step} completed")

3.3 查询字符串注入 — Python 篇

利用 Python 的事件系统来扩展 boto3 函数的能力。通过对 debug 日志的拆解，可以看到 boto3 对请求签名的逻辑大致如下图所示。

在事件 provide-client-params 中，程序会从 API 调用中读取请求参数并作验证；那么在这一步中需要将用户请求的查询字符串剥离，存入自由参数 context 以通过验证。最后在事件 before-sign 中需要将查询字符串的值从自由参数中取回并添加到 URL 的最后，确保 boto3 在使用 SigV4 生成签名的 URL 已经携带了查询字符串，避免 S3 服务器的 SignatureDoesNotMatch 报错。

import boto3
import time

from botocore.client import Config
from urllib.parse import urlencode
#boto3.set_stream_logger(name='botocore')

# Ensure signature V4 mode
s3 = boto3.client("s3", config=Config(signature_version="s3v4"))

# Customize query string check
def is_custom(k):
    return k.lower().startswith("x-user")

# Event to seprate the query string
def client_param_handler(*, params, context, **_kw):
    # Store custom parameters in context for later event handlers
    context["custom_params"] = params['Key'].split('?')[1]
    # Remove custom parameters from client parameters because validation would fail on them
    params['Key']=params['Key'].split('?')[0]
    return params

# Event to append query string before sign
def request_param_injector(*, request, **_kw):
    request.url += "&" if "?" in request.url else "?"
    request.url += request.context["custom_params"]

# Register the event for client method
s3.meta.events.register("provide-client-params.s3.GetObject", client_param_handler)
s3.meta.events.register("before-sign.s3.GetObject", request_param_injector)

3.4 查询字符串注入 — Java 篇

由于构架企业应用仍然以Java语言生态为主，尤其是开源大数据生态中，Java是主流的语言，因此针对 Java 的解决办法可以极大提高注入查询字符串的可操性。在 Java SDK v2 中，GetObjectRequest 继承 AwsRequest 类，而 AwsRequest 类支持将客户端请求重写，因此可以借助 putRawQueryParameter 方法将额外的查询字符串附加进 HTTP请求 URI 中。

import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
import software.amazon.awssdk.core.ResponseBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.S3Exception;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.core.RequestOverrideConfiguration;
import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
import software.amazon.awssdk.core.SdkRequestOverrideConfiguration;
import software.amazon.awssdk.http.apache.ApacheHttpClient;
import software.amazon.awssdk.http.SdkHttpClient;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;

public static void main( String[] args )
    {
        
        SdkHttpClient httpClient = ApacheHttpClient.builder()
                .build();

        Region region = Region.US_WEST_2;
        S3Client s3 = S3Client.builder()
                    .region(region)
                    .httpClient(httpClient)
                    .build();

        GetObjectRequest getObjectRequest = GetObjectRequest.builder()
                    .bucket("s3tracker-test")
                    .key("test.yaml")
                    .build();

        RequestOverrideConfiguration config = SdkRequestOverrideConfiguration.builder()
                .putRawQueryParameter("x-user", "test")
                .build();

        s3.getObject(getObjectRequest.toBuilder().overrideConfiguration(
                AwsRequestOverrideConfiguration.from(config)
              ).build()
            );
        System.out.println( "Request complete!" );
        httpClient.close();;

    }

3.5 结果展示

使用 Python 编写的小程序来模拟前端用户尝试访问对象。附加的事件系统对于用户的日常访问是无感知的。

注入查询字符串后的 S3 服务器请求日志，在 Request-URI 字段中可以看到完整的对象键名及“x-user”信息：

0b821b5dfdecb9b7b1c3ec3ebcf474c7644324c905a184417dc2ecf3af9448d3 blur-bucketname [19/Sep/2022:08:19:38 +0000] 205.206.233.234 arn:aws:iam::<accountid>:user/toolman MK7JF71WRK6Z7P01 REST.GET.OBJECT vpc-1.yaml "GET /vpc-1.yaml?x-user=testuser HTTP/1.1" 200 - 2030 2030 59 58 "-" "Boto3/1.24.80 Python/3.10.5 Darwin/21.4.0 Botocore/1.27.80" - qqc5t2N9y2pUaCfU8juNy4taOfAQKKh0kdpr8H8kh+GZ+hukH3lS5BnmiJme5+41g8s+OUoH5pc= SigV4 ECDHE-RSA-AES128-GCM-SHA256 AuthHeader blur-bucketname.s3.amazonaws.com TLSv1.2 -
0b821b5dfdecb9b7b1c3ec3ebcf474c7644324c905a184417dc2ecf3af9448d3 blur-bucketname [19/Sep/2022:08:19:06 +0000] 205.206.233.234 arn:aws:iam::<accountid>:user/toolman JCRS0A95XT8RE3QH REST.GET.OBJECT vpc-1.yaml "GET /vpc-1.yaml?x-user=testuser HTTP/1.1" 200 - 2030 2030 64 63 "-" "Boto3/1.24.80 Python/3.10.5 Darwin/21.4.0 Botocore/1.27.80" - o3A8VuLDJ7h7fs+EJ18jbBbkzGDrcpyhHJJ6qJsRpZNUY40MsGOreKzaRFVzS3LK3+4zC3PmyfA= SigV4 ECDHE-RSA-AES128-GCM-SHA256 AuthHeader blur-bucketname.s3.amazonaws.com TLSv1.2 -

在应用端成功将查询字符串注入请求后，就可以通过 QuickSight 将 Athena 分析好的用户请求数量及总共的数据传输字节进行呈现：

总结

以上方案给大家展示了如何在 S3 里按请求者级别颗粒度进行用量追踪，帮助 S3 用户实现对请求量与数据传输量的精细化管理。此外，在不借助其他诸如 IAM 、前缀、标签等额外功能情况下，本文也提供了 SDK 层面注入查询字符串的解决办法，适合对已有多租户或共享存储场景做轻量化的改造。当拥有了请求者级别颗粒度的用量信息后，SaaS 供应商能够根据租户的实际用量进行计费，并且能够基于这些数据制定更加贴合业务的收费策略及套餐。

亚马逊AWS官方博客