导出 Cloud Logging 的场景：Elasticsearch

此场景显示了如何将选定的日志从 Logging 导出到 Elasticsearch 集群。Elasticsearch 是一个开源文档数据库，用于提取、索引和分析非结构化数据，例如日志、指标和其他遥测数据。将通过 Logging 收集的日志与您在 Elasticsearch 中可能拥有的日志集成，可提供统一的日志分析解决方案。

此场景是导出 Logging 设计模式系列的一部分。

简介

Elastic Stack 具有多种用于将数据提取到 Elasticsearch 集群中的解决方案。Logstash 和 Beats 是用于收集、转换和提取数据的核心产品。根据您的数据量、提取率和延迟时间要求，在 Logstash 和 Beats 之间进行选择。此解决方案侧重于 Elastic Stack 的 Logstash 组件，因为 Logstash 是处理从 Logging 中导出的日志的最灵活的选项。

您可以使用 Logstash 从多个来源和目标位置收集、转换和导出日志。如果日志量较大，则最好使用 Logstash。您可以使用 Pub/Sub 输入插件和 Cloud Storage 输入插件将 Logstash 与 Elasticsearch 集成。

Beats 是一个开源平台，用于将数据以轻量级方式传输到 Elasticsearch 中。您可以使用 pubsubbeat 传输工具将 Pub/Sub 消息提取到 Elasticsearch 中。虽然有 gcsbeat 传输工具，但它适用于在存储分区的根级轻量传输文件，因此它不是导入由 Logging 导出的日志的最佳解决方案。

在以下情况下可使用 Beats：

• 您的日志量较小（可由一台计算机处理）。
• 您不希望承担运行 Logstash 的开销。
• 您可以通过 Elastic 提取流水线来满足您的日志扩充要求。

在以下情况下可使用 Logstash：

• 您的日志量大。
• 除了 Elastic 提取流水线功能之外，您还需要通过其他功能来丰富自己的日志。
• 您需要能够将日志导出到 Elasticsearch 以外的系统。

	量大或互动性强	量小且互动性弱
实时	logstash-input-google_pubsub	pubsubbeat
最终一致	logstash-input-google_cloud_storage	gcsbeat

注意：您可以运行多个 pubsubbeat 实例，而无需担心处理重复文件。

下图显示了使用 Logstash 批量和实时导出到 Elasticsearch 的架构。

设置实时日志记录导出

在本部分，您可以使用 Pub/Sub 创建用于将日志从 Logging 实时导出到 Logstash 的流水线。

设置 Pub/Sub 主题

• 按照 Pub/Sub 主题设置说明操作，该主题将用于接收导出的日志。

打开所有服务的审核日志

默认情况下，系统禁用数据访问的审核日志（BigQuery 除外）。要启用所有审核日志，请使用审核政策文档中列出的配置并按照更新 IAM 政策说明执行操作。步骤包括：

• 将当前的 IAM 政策下载为文件。
• 将审核日志政策 JSON 或 YAML 对象添加到当前的政策文件。
• 使用更改后的政策文件更新 Google Cloud 项目。

以下是一个示例 JSON 对象，它启用了全部服务的全部审核日志。

"auditConfigs": [
    {
        "service": "allServices",
        "auditLogConfigs": [
            { "logType": "ADMIN_READ" },
            { "logType": "DATA_READ"  },
            { "logType": "DATA_WRITE" }
        ]
    }
],

配置日志导出

设置集合导出或日志导出后，需要优化日志过滤条件以导出审核日志、与虚拟机相关的日志、存储日志和数据库日志。以下日志过滤条件涵盖了管理员活动和数据访问审核日志以及特定资源类型的日志。

logName:"logs/cloudaudit.googleapis.com" OR
resource.type:gce OR
resource.type=gcs_bucket OR
resource.type=bigquery_resource

在 gcloud 命令行工具中，使用 gcloud logging sinks create 命令或 organizations.sinks.create API 调用来创建具有适当过滤条件的接收器。以下 gcloud 命令示例会为组织创建一个名为 gcp_logging_sink_pubsub 的接收器。该接收器包括所有子项目，并指定过滤条件以选择特定的审核日志。

ORG_ID=your-org-id
PROJECT_ID=$(gcloud config get-value project)
gcloud logging sinks create gcp_logging_sink_pubsub \
    pubsub.googleapis.com/projects/${PROJECT_ID}/topics/logs-export-topic \
    --log-filter='logName="logs/cloudaudit.googleapis.com" OR resource.type:"gce" OR resource.type="gcs_bucket" OR resource.type="bigquery_resource"' \
    --include-children \
    --organization=${ORG_ID}

输出的结果应类似于以下内容：

Created [https://logging.googleapis.com/v2/organizations/your-org-id/sinks/gcp_logging_export_pubsub_sink].
Please remember to grant `serviceAccount:[e-mail]` Pub/Sub Publisher role to the topic.
More information about sinks can be found at /logging/docs/export/configure_export

在 API 调用返回的 serviceAccount 条目中，响应中包含身份 [e-mail]。该身份表示为导出创建的 Google Cloud 服务帐号。在您授予此身份对目标主题的发布访问权限之前，从此接收器导出日志条目的操作将会失败。如需了解详情，请参阅下一部分或授予资源访问权限文档。

为 Pub/Sub 主题设置 IAM 政策权限

通过将服务帐号 [e-mail] 添加到 pubsub.googleapis.com/projects/${PROJECT_ID}/topics/logs-export-topic 主题并为其设置 Pub/Sub Publisher 权限，可授予该服务帐号向该主题发布消息的权限。在您添加这些权限之前，接收器导出将失败。

如需向服务帐号添加权限，请按以下步骤操作：

1. 在 Cloud Console 中，打开 Cloud Pub/Sub 主题页面：

   转到“主题”页面

2. 选择主题名称。

3. 点击显示信息面板，然后配置权限。确保已选择 Pub/Sub Publisher 权限。

   

使用此过滤条件创建日志记录导出后，日志文件将开始填充到已配置的项目的 Pub/Sub 主题中。您可以使用 Cloud Monitoring 中的 Metrics Explorer 确认主题正在接收消息。使用以下资源类型和指标，观察短时间内的消息发送操作数。如果您已正确配置导出，则将在图表上看到活动数大于 0，如以下屏幕截图所示。

• 资源类型：pubsub_topic
• 指标：pubsub.googleapis.com/topic/send_message_operation_count
• 过滤条件：topic_id="logs-export-topic"

设置日志记录导出

由于 Logging 的日志保留期限有限，因此创建到 Cloud Storage 的 Logging 导出是最佳做法。使用 Logstash 提取导出的日志，您可以搜索默认保留期限以外的日志，还有助于在需要时重新填充 Elasticsearch 数据库。

配置 Logstash

1. 下载最新的 logstash 版本。
2. 为 gcsbeat 创建 IAM 服务帐号和 JSON 帐号密钥，并为其授予 Cloud Storage Admin 和 Pub/Sub Subscriber 角色。

3. 从您的 Logstash 安装目录运行以下命令，为 Logstash 安装 Cloud Storage 和 Pub/Sub 输入插件：

   ./bin/logstash-plugin install \
       logstash-input-google_cloud_storage \
       logstash-input-exec \
       logstash-input-google_pubsub
   

创建批量提取流水线

• 如下所示，为 Cloud Storage 创建一个 Logstash 配置文件，将 my-cloud-logs 替换为存储分区的 ID：

  input {
    google_cloud_storage {
      type => "gcs"
      interval => 60
      bucket_id => "my-cloud-logs"
      json_key_file => "logstash-sa.json"
      file_matches => ".*\.json"
      codec => "json"
    }
  }
  
  output {
    elasticsearch {
      document_id => "%{[@metadata][gcs][line_id]}"
    }
  }
  

  如需查看所有选项的列表，请参阅 google_cloud_storage 插件配置文档。

Cloud Storage 中 Logstash 处理过的对象会使用元数据项 x-goog-meta-ls-gcs-input:processed 进行标记。输入插件会跳过带有此元数据项的所有对象。如果您打算重新处理数据，则必须手动移除元数据标签。

创建实时流水线

• 为 Pub/Sub 源创建一个 logstash 配置文件，格式如下：

  input {
    google_pubsub {
        type => "pubsub"
        project_id => "my-project-id"
        topic => "logs-export-topic"
        subscription => "logstash"
        json_key_file => "logstash-sa.json"
    }
  }
  
  output {
    elasticsearch {
      document_id => "%{[@metadata][pubsub_message][messageId]}"
    }
  }
  

如需查看配置项的完整列表，请参阅插件文档。

幂等插入

Elasticsearch 将文档的 _id 字段用作唯一标识符。在 Logstash 中，您可以使用 [@metadata] 项和其他消息字段，根据 Logging 中的日志消息类型创建唯一文档 ID。

google_cloud_storage 插件元数据文档包含可用的 Logstash 元数据字段列表。在创建幂等文档 ID 的用例中，系统会明确插入 [@metadata][gcs][line_id] 字段。

google_pubsub plugin 还包含一个元数据字段列表，[@metadata][pubsub_message][messageId] 常用于去重。

使用导出的日志

将导出的日志提取到 Elasticsearch 中并将实时 Beat 部署为服务后，您可以使用 Elasticsearch 和 Elastic Stack 中的产品执行以下任务：

• 搜索日志。
• 关联复杂事件。
• 创建自动提醒。
• 以图表的形式了解数据中的关系。
• 使用 Kibana 信息中心直观呈现结果。

后续步骤

• 将日志导出到 Elastic Cloud 中
• 了解其他导出场景：
  • 场景 - 为满足合规性要求而导出
  • 场景 - 为安全性和访问分析而导出
  • 场景 - 导出到 Splunk
  • 场景 - 导出到 Datadog
• 试用其他 Google Cloud 功能。查阅我们的教程。