此页面由 Cloud Translation API 翻译。

收集 Sentry 日志

支持的平台：

Google SecOps SIEM

本文档介绍了如何使用 Amazon S3 将 Sentry 日志注入到 Google Security Operations。Sentry 会生成运营数据，包括事件、问题、性能监控数据和错误跟踪信息。通过此集成，您可以将这些日志发送到 Google SecOps 进行分析和监控，从而了解 Sentry 监控的应用中的应用错误、性能问题和用户互动。

准备工作

请确保满足以下前提条件：

Google SecOps 实例。
对 Sentry 租户的特权访问权限（具有 API 范围的身份验证令牌）。
对 AWS（S3、Identity and Access Management (IAM)、Lambda、EventBridge）的特权访问权限。

收集 Sentry 前提条件（ID、API 密钥、组织 ID、令牌）

登录 Sentry。
查找您的组织处理后标题：
- 依次前往设置 > 组织 > 设置 > 组织 ID（组织 ID 显示在组织名称旁边）。
创建身份验证令牌：
- 依次前往设置 > 开发者设置 > 个人令牌。
- 点击新建
- 范围（最低要求）：org:read、project:read、event:read。
- 复制令牌值（仅显示一次）。此参数用作：Authorization: Bearer <token>。
（如果是自行托管）：记下您的基本网址（例如 https://<your-domain>）；否则请使用 https://sentry.io。

为 Google SecOps 配置 AWS S3 存储桶和 IAM

按照以下用户指南创建 Amazon S3 存储桶：创建存储桶
保存存储桶名称和区域以供日后参考（例如 sentry-logs）。
按照以下用户指南创建用户：创建 IAM 用户。
选择创建的用户。
选择安全凭据标签页。
在访问密钥部分中，点击创建访问密钥。
选择第三方服务作为使用情形。
点击下一步。
可选：添加说明标记。
点击创建访问密钥。
点击下载 CSV 文件，保存访问密钥和秘密访问密钥，以供日后参考。
点击完成。
选择权限标签页。
在权限政策部分中，点击添加权限。
选择添加权限。
选择直接附加政策。
搜索 AmazonS3FullAccess 政策。
选择相应政策。
点击下一步。
点击添加权限。

为 S3 上传配置 IAM 政策和角色

在 AWS 控制台中，依次前往 IAM > 政策。
依次点击创建政策 > JSON 标签页。
复制并粘贴以下政策。

政策 JSON（如果您输入了其他存储桶名称，请替换 sentry-logs）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::sentry-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::sentry-logs/sentry/events/state.json"
    }
  ]
}

依次点击下一步 > 创建政策。
依次前往 IAM > 角色 > 创建角色 > AWS 服务 > Lambda。
附加新创建的政策。
将角色命名为 WriteSentryToS3Role，然后点击创建角色。

创建 Lambda 函数

在 AWS 控制台中，依次前往 Lambda > 函数 > 创建函数。
点击从头开始创作。
提供以下配置详细信息：

设置值

名称 sentry_to_s3

运行时 Python 3.13

架构 x86_64

执行角色 WriteSentryToS3Role

设置	值
名称	`sentry_to_s3`
运行时	Python 3.13
架构	x86_64
执行角色	`WriteSentryToS3Role`

创建函数后，打开 Code 标签页，删除桩代码并粘贴以下代码 (sentry_to_s3.py)。

#!/usr/bin/env python3
# Lambda: Pull Sentry project events (raw JSON) to S3 using Link "previous" cursor for duplicate-safe polling

import os, json, time
from urllib.request import Request, urlopen
from urllib.parse import urlencode, urlparse, parse_qs
import boto3

ORG = os.environ["SENTRY_ORG"].strip()
TOKEN = os.environ["SENTRY_AUTH_TOKEN"].strip()
S3_BUCKET = os.environ["S3_BUCKET"]
S3_PREFIX = os.environ.get("S3_PREFIX", "sentry/events/")
STATE_KEY = os.environ.get("STATE_KEY", "sentry/events/state.json")
BASE = os.environ.get("SENTRY_API_BASE", "https://sentry.io").rstrip("/")
MAX_PROJECTS = int(os.environ.get("MAX_PROJECTS", "100"))
MAX_PAGES_PER_PROJECT = int(os.environ.get("MAX_PAGES_PER_PROJECT", "5"))

s3 = boto3.client("s3")
HDRS = {"Authorization": f"Bearer {TOKEN}", "Accept": "application/json", "User-Agent": "chronicle-s3-sentry-lambda/1.0"}

def _get_state() -> dict:
    try:
        obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        raw = obj["Body"].read()
        return json.loads(raw) if raw else {"projects": {}}
    except Exception:
        return {"projects": {}}

def _put_state(state: dict):
    s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=json.dumps(state, separators=(",", ":")).encode("utf-8"))

def _req(path: str, params: dict | None = None):
    url = f"{BASE}{path}"
    if params:
        url = f"{url}?{urlencode(params)}"
    req = Request(url, method="GET", headers=HDRS)
    with urlopen(req, timeout=60) as r:
        data = json.loads(r.read().decode("utf-8"))
        link = r.headers.get("Link")
        return data, link

def _parse_link(link_header: str | None):
    """Return (prev_cursor, prev_has_more, next_cursor, next_has_more)."""
    if not link_header:
        return None, False, None, False
    prev_cursor, next_cursor = None, None
    prev_more, next_more = False, False
    parts = [p.strip() for p in link_header.split(",")]
    for p in parts:
        if "<" not in p or ">" not in p:
            continue
        url = p.split("<", 1)[1].split(">", 1)[0]
        rel = "previous" if 'rel="previous"' in p else ("next" if 'rel="next"' in p else None)
        has_more = 'results="true"' in p
        try:
            q = urlparse(url).query
            cur = parse_qs(q).get("cursor", [None])[0]
        except Exception:
            cur = None
        if rel == "previous":
            prev_cursor, prev_more = cur, has_more
        elif rel == "next":
            next_cursor, next_more = cur, has_more
    return prev_cursor, prev_more, next_cursor, next_more

def _write_page(project_slug: str, payload: object, page_idx: int) -> str:
    ts = time.gmtime()
    key = f"{S3_PREFIX.rstrip('/')}/{time.strftime('%Y/%m/%d', ts)}/sentry-{project_slug}-{page_idx:05d}.json"
    s3.put_object(Bucket=S3_BUCKET, Key=key, Body=json.dumps(payload, separators=(",", ":")).encode("utf-8"))
    return key

def list_projects(max_projects: int):
    projects, cursor = [], None
    while len(projects) < max_projects:
        params = {"cursor": cursor} if cursor else {}
        data, link = _req(f"/api/0/organizations/{ORG}/projects/", params)
        for p in data:
            slug = p.get("slug")
            if slug:
                projects.append(slug)
                if len(projects) >= max_projects:
                    break
        # advance pagination
        _, _, next_cursor, next_more = _parse_link(link)
        cursor = next_cursor if next_more else None
        if not next_more:
            break
    return projects

def fetch_project_events(project_slug: str, start_prev_cursor: str | None):
    # If we have a stored "previous" cursor, poll forward (newer) until no more results.
    # If not (first run), fetch the latest page, then optionally follow "next" (older) for initial backfill up to the limit.
    pages = 0
    total = 0
    latest_prev_cursor_to_store = None

    def _one(cursor: str | None):
        nonlocal pages, total, latest_prev_cursor_to_store
        params = {"cursor": cursor} if cursor else {}
        data, link = _req(f"/api/0/projects/{ORG}/{project_slug}/events/", params)
        _write_page(project_slug, data, pages)
        total += len(data) if isinstance(data, list) else 0
        prev_c, prev_more, next_c, next_more = _parse_link(link)
        # capture the most recent "previous" cursor observed to store for the next run
        latest_prev_cursor_to_store = prev_c or latest_prev_cursor_to_store
        pages += 1
        return prev_c, prev_more, next_c, next_more

    if start_prev_cursor:
        # Poll new pages toward "previous" until no more
        cur = start_prev_cursor
        while pages < MAX_PAGES_PER_PROJECT:
            prev_c, prev_more, _, _ = _one(cur)
            if not prev_more:
                break
            cur = prev_c
    else:
        # First run: start at newest, then (optionally) backfill a few older pages
        prev_c, _, next_c, next_more = _one(None)
        cur = next_c
        while next_more and pages < MAX_PAGES_PER_PROJECT:
            _, _, next_c, next_more = _one(cur)
            cur = next_c

    return {"project": project_slug, "pages": pages, "written": total, "store_prev_cursor": latest_prev_cursor_to_store}

def lambda_handler(event=None, context=None):
    state = _get_state()
    state.setdefault("projects", {})

    projects = list_projects(MAX_PROJECTS)
    summary = []
    for slug in projects:
        start_prev = state["projects"].get(slug, {}).get("prev_cursor")
        res = fetch_project_events(slug, start_prev)
        if res.get("store_prev_cursor"):
            state["projects"][slug] = {"prev_cursor": res["store_prev_cursor"]}
        summary.append(res)

    _put_state(state)
    return {"ok": True, "projects": len(projects), "summary": summary}

if __name__ == "__main__":
    print(lambda_handler())

依次前往配置 > 环境变量。
依次点击修改 > 添加新的环境变量。

输入下表中提供的环境变量，并将示例值替换为您的值。

环境变量

键	示例值	说明
`S3_BUCKET`	`sentry-logs`	将存储数据的 S3 存储桶名称。
`S3_PREFIX`	`sentry/events/`	对象的可选 S3 前缀（子文件夹）。
`STATE_KEY`	`sentry/events/state.json`	可选的状态/检查点文件键。
`SENTRY_ORG`	`your-org-slug`	Sentry 组织 slug。
`SENTRY_AUTH_TOKEN`	`sntrys_************************`	具有组织读取权限、项目读取权限和事件读取权限的 Sentry 身份验证令牌。
`SENTRY_API_BASE`	`https://sentry.io`	Sentry API 基准网址（自托管：`https://<your-domain>`）。
`MAX_PROJECTS`	`100`	要处理的项目数量上限。
`MAX_PAGES_PER_PROJECT`	`5`	每次执行时每个项目的最大页数。

创建函数后，请停留在其页面上（或依次打开 Lambda > 函数 > 您的函数）。
选择配置标签页。
在常规配置面板中，点击修改。
将超时更改为 5 分钟（300 秒），然后点击保存。

创建 EventBridge 计划

依次前往 Amazon EventBridge > 调度器 > 创建调度。
提供以下配置详细信息：
- 周期性安排：费率 (1 hour)。
- 目标：您的 Lambda 函数 sentry_to_s3。
- 名称：sentry-1h。
点击创建时间表。

（可选）为 Google SecOps 创建只读 IAM 用户和密钥

在 AWS 控制台中，依次前往 IAM > 用户。
点击 Add users（添加用户）。
提供以下配置详细信息：
- 用户：输入 secops-reader。
- 访问类型：选择访问密钥 - 以程序化方式访问。
点击创建用户。
附加最低限度的读取政策（自定义）：依次选择用户 > secops-reader > 权限 > 添加权限 > 直接附加政策 > 创建政策。

JSON：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::sentry-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::sentry-logs"
    }
  ]
}

名称 = secops-reader-policy。
依次点击创建政策 > 搜索/选择 > 下一步 > 添加权限。
为 secops-reader 创建访问密钥：安全凭据 > 访问密钥。
点击创建访问密钥。
下载 .CSV。（您需要将这些值粘贴到 Feed 中）。

在 Google SecOps 中配置 Feed 以注入 Sentry 日志

依次前往 SIEM 设置> Feed。
点击 + 添加新 Feed。
在Feed 名称字段中，输入 Feed 的名称（例如 Sentry Logs）。
选择 Amazon S3 V2 作为来源类型。
选择 Sentry 作为日志类型。
点击下一步。
为以下输入参数指定值：
- S3 URI：s3://sentry-logs/sentry/events/
- 来源删除选项：根据您的偏好选择删除选项。
- 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
- 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥。
- 私有访问密钥：具有 S3 存储桶访问权限的用户私有密钥。
- 资产命名空间：资产命名空间。
- 注入标签：应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。