› › ›

Salesforce 输入

使用 salesforce 输入通过 Salesforce EventLogFile (ELF) API 或 Salesforce 实时事件监控 API 来监控 Salesforce 事件。两者都使用 REST API（在 Salesforce 实例中执行 SOQL 查询）来查询相关对象以获取事件。

Salesforce 输入在请求之间维护游标状态，以跟踪每次执行中检索到的最后一个事件。这些游标状态被传递到下一个事件监控执行，以从最后一个已知位置恢复获取事件。游标状态允许输入从中断的地方继续，并提供对输入行为的控制。

以下是一些支持的身份验证方法和事件监控方法

身份验证方法
- OAuth2
  - 用户密码流
  - JWT 持有者流
事件监控方法
- 使用 REST API 的 EventLogFile (ELF)
- 用于对象的 REST API（用于监控实时事件）

以下是有关游标如何在 Salesforce 输入中使用的几个要点

为每个配置的事件监控方法（event_log_file 和 object）维护单独的游标状态。
游标状态根据配置中指定的 cursor.field 存储检索到的最后一个事件的唯一标识符。
首次运行时，使用 query.default 来获取初始事件集。
在后续运行时，使用游标状态填充 query.value 模板，以获取自上次执行以来的事件。
如果输入重新启动，它将从上次持久化的游标状态恢复，而不是从头开始。

使用游标允许 Salesforce 输入可靠地跟踪其进度，并避免在执行过程中丢失或重复事件。应仔细选择游标字段，使其对于每个新事件具有单调递增的值。

事件监控方法是高度可配置的，可用于监控任何支持的对象或事件日志文件。可以将输入配置为同时监控多个对象或事件日志文件。

配置示例

filebeat.inputs:
  - type: salesforce
    enabled: true
    version: 56
    auth.oauth2:
      user_password_flow:
        enabled: true
        client.id: client-id
        client.secret: client-secret
        token_url: https://instance-id.develop.my.salesforce.com
        username: [email protected]
        password: salesforce-instance-password
      jwt_bearer_flow:
        enabled: true
        client.id: client-id
        client.username: [email protected]
        client.key_path: server_client.key
        url: https://login.salesforce.com
    url: https://instance-id.develop.my.salesforce.com
    event_monitoring_method:
      event_log_file:
        enabled: true
        interval: 1h
        query:
          default: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE EventType = 'Login' ORDER BY CreatedDate ASC NULLS FIRST"
          value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE EventType = 'Login' AND CreatedDate > [[ .cursor.event_log_file.last_event_time ]] ORDER BY CreatedDate ASC NULLS FIRST"
        cursor:
          field: "CreatedDate"
      object:
        enabled: true
        interval: 5m
        query:
          default: "SELECT FIELDS(STANDARD) FROM LoginEvent"
          value: "SELECT FIELDS(STANDARD) FROM LoginEvent WHERE EventDate > [[ .cursor.object.first_event_time ]]"
        cursor:
          field: "EventDate"

在 Salesforce 中设置 OAuth 应用程序

编辑

为了使用此集成，用户需要使用 OAuth 创建一个新的 Salesforce 应用程序。按照以下步骤在 Salesforce 中创建连接的应用程序

使用用户想要收集数据的相同用户凭据登录到 Salesforce。
单击右上角菜单栏上的“设置”。在“设置”页面上，在页面顶部的 搜索设置 搜索框中搜索 应用程序管理器，然后选择 应用程序管理器。
单击新建连接的应用程序。
为连接的应用程序提供一个名称。这将在“应用程序管理器”及其“应用程序启动器”磁贴上显示。
输入 API 名称。默认值是不带空格的名称版本。只允许字母、数字和下划线。如果原始应用程序名称包含任何其他字符，请编辑默认名称。
输入 Salesforce 的联系人电子邮件。
在页面的 API（启用 OAuth 设置）部分下，选择启用 OAuth 设置。
在回调 URL 中，输入实例 URL（请参阅 Salesforce 实例 URL）。
选择以下 OAuth 范围以应用于连接的应用程序
- 通过 API 管理用户数据 (api)。
- 随时执行请求 (refresh_token, offline_access)。
- （可选）在数据收集的情况下，如果出现任何权限问题，请添加完全访问权限 (full) 范围。
选择需要 Web 服务器流的密钥，以要求应用程序的客户端密钥来换取访问令牌。
选择需要刷新令牌流的密钥，以要求应用程序的客户端密钥在刷新令牌和混合刷新令牌流的授权请求中。
单击“保存”。更改生效可能需要大约 10 分钟。
单击“继续”，然后在“API 详细信息”下，单击“管理消费者详细信息”。使用验证码验证用户帐户。
从“消费者详细信息”部分复制 消费者密钥和 消费者密钥，它们应该分别在配置中填充为客户端 ID 和客户端密钥的值。

有关如何创建连接的应用程序的更多详细信息，请参阅 Salesforce 文档此处。

启用实时事件

要开始使用实时事件，请转到“设置”，然后在快速查找中搜索事件管理器。企业版和无限版环境默认可以访问注销事件，但其余事件需要获得许可才能访问 Shield 事件监控。

执行

编辑

salesforce 输入是一个长时间运行的程序，它从 Salesforce 实例检索事件并将它们发送到指定的输出。该程序在一个循环中执行，以预配置的间隔从 Salesforce 实例获取事件。可以将每个事件监控方法配置为单独运行，并以不同的间隔运行。为了防止内存使用量突然飙升，如果配置了多个事件监控方法，则会安排它们一次运行一个。即使间隔重叠，也只会随机执行一种方法，而另一种方法将在第一种方法完成后执行。

有两种方法可以从 Salesforce 实例获取事件

event_log_file：EventLogFile 是 Salesforce 中的标准对象，事件监控方法在后台使用 REST API 从该对象收集 Salesforce 组织的运营事件。有一个字段 EventType 可以帮助区分操作事件的类型，如登录、注销等。使用 Salesforce 的查询语言 SOQL 来查询该对象。
object：此方法是一种通过使用 REST API 从 Salesforce 实例检索事件的通用方法。它可用于监控对象的实时状态。在实时事件监控中，订阅事件是一种常见做法，但事件也存储在 Salesforce 组织中（如果已配置），特别是为每种事件类型预配置的大对象表中。使用此方法，我们使用 Salesforce 的查询语言 (SOQL) 查询对象。收集发生在配置的抓取 interval 中。

Salesforce 对象和 SOQL 查询字段排序限制

每个 Salesforce 对象都包含一组字段，但 SOQL 查询对可以排序的字段和特定的排序方法有限制。Salesforce 开发人员页面上的对象描述提供了有关这些限制的信息。例如，登录对象只允许按 EventDate 字段以降序排序。

当使用游标随时间收集数据时，可以使用以下游标输入

object.first_event_time：此游标输入存储使用 object 方法进行数据收集期间遇到的第一个事件的游标值。
object.last_event_time：此游标输入存储使用 object 方法进行数据收集期间遇到的最后一个事件的游标值。
event_log_file.first_event_time：此游标输入存储使用事件日志文件方法进行数据收集期间遇到的第一个事件的游标值。
event_log_file.last_event_time：此游标输入存储使用事件日志文件方法进行数据收集期间遇到的最后一个事件的游标值。

通过选择上述游标输入之一，用户可以按所需的顺序从对象和事件日志文件中收集数据。游标配置可以根据用户的特定需求进行自定义。

配置选项

编辑

salesforce 输入支持以下配置选项以及稍后描述的通用选项。

`enabled`

编辑

是否启用输入。默认值：false。

`version`

编辑

要使用的 Salesforce API 的版本。支持的最低版本为 46。

`auth`

编辑

Salesforce 实例的身份验证设置。

`auth.oauth2`

编辑

Salesforce 实例的 OAuth2 身份验证选项。

支持两种 OAuth2 身份验证流

user_password_flow：用户密码流
jwt_bearer_flow：JWT 持有者流

`auth.oauth2.user_password_flow.enabled`

编辑

是否使用用户密码流进行身份验证。默认值：false。

一次只能启用一个身份验证流。

`auth.oauth2.user_password_flow.client.id`

编辑

用户密码流的客户端 ID。

`auth.oauth2.user_password_flow.client.secret`

编辑

用户密码流的客户端密钥。

`auth.oauth2.user_password_flow.token_url`

编辑

用户密码流的令牌 URL。

`auth.oauth2.user_password_flow.username`

编辑

用户密码流的用户名。

`auth.oauth2.user_password_flow.password`

编辑

用户密码流的密码。

`auth.oauth2.jwt_bearer_flow.enabled`

编辑

是否使用 JWT 持有者流进行身份验证。默认值：false。

一次只能启用一个身份验证流。

`auth.oauth2.jwt_bearer_flow.client.id`

编辑

JWT 持有者流程的客户端 ID。

`auth.oauth2.jwt_bearer_flow.client.username`

编辑

JWT 持有者流程的用户名。

`auth.oauth2.jwt_bearer_flow.client.key_path`

编辑

JWT 持有者流程的私钥文件路径。该文件必须是 PEM 编码的 PKCS1 或 PKCS8 私钥，并且必须设置正确的权限，以便运行该程序的用户具有读取访问权限。

`auth.oauth2.jwt_bearer_flow.url`

编辑

JWT 持有者流程的 URL。

`url`

编辑

Salesforce 实例的 URL。必需。

`resource.timeout`

编辑

声明 HTTP 客户端连接超时的持续时间。有效的时间单位是 ns、us、ms、s、m、h。默认值：30s。

`resource.retry.max_attempts`

编辑

HTTP 客户端的最大重试次数。默认值：5。

`resource.retry.wait_min`

编辑

尝试重试之前等待的最小时间。默认值：1s。

`resource.retry.wait_max`

编辑

尝试重试之前等待的最大时间。默认值：60s。

`event_monitoring_method`

编辑

要使用的事件监控方法。支持两种事件监控方法

event_log_file：使用 REST API 的 EventLogFile (ELF)
object：使用 REST API 的实时事件监控（对象）

`event_monitoring_method.event_log_file`

编辑

要使用的事件监控方法 — event_log_file。使用 EventLogFile API 从 Salesforce 实例获取事件。

`event_monitoring_method.event_log_file.enabled`

编辑

是否使用 EventLogFile API 进行事件监控。默认值：false。

`event_monitoring_method.event_log_file.interval`

编辑

使用 EventLogFile API 从 Salesforce 实例收集事件的间隔。

`event_monitoring_method.event_log_file.query.default`

编辑

使用 EventLogFile API 从 Salesforce 实例获取事件的默认查询。

如果游标状态不可用，将使用默认查询从 Salesforce 实例获取事件。默认查询必须是有效的 SOQL 查询。如果 event_monitoring_method.event_log_file.query.value 中的 SOQL 查询无效，则将使用默认查询从 Salesforce 实例获取事件。

`event_monitoring_method.event_log_file.query.value`

编辑

使用 EventLogFile API 从 Salesforce 实例获取事件的 SOQL 查询，但它使用游标状态从 Salesforce 实例获取事件。SOQL 查询必须是有效的 SOQL 查询。如果 SOQL 查询无效，则将使用默认查询从 Salesforce 实例获取事件。

在重新启动或后续执行的情况下，将使用游标状态从 Salesforce 实例获取事件。游标状态是从 Salesforce 实例获取的最后一个事件的最后一个事件时间。游标状态取自从 Salesforce 实例获取的最后一个事件的 event_monitoring_method.event_log_file.cursor.field 字段。

`event_monitoring_method.event_log_file.cursor.field`

编辑

用于从 Salesforce 实例获取的最后一个事件中获取游标状态的字段。该字段必须是 event_monitoring_method.event_log_file.query.default 和 event_monitoring_method.event_log_file.query.value 中指定的 SOQL 查询中的有效字段，即 SOQL 查询中选定字段的一部分。

`event_monitoring_method.object`

编辑

要使用的事件监控方法 — object。使用 REST API 直接从 Salesforce 实例的对象中获取事件。

`event_monitoring_method.object.enabled`

编辑

是否使用 REST API 从对象进行事件监控。默认值：false。

`event_monitoring_method.object.interval`

编辑

使用来自对象的 REST API 从 Salesforce 实例收集事件的间隔。

`event_monitoring_method.object.query.default`

编辑

使用来自对象的 REST API 从 Salesforce 实例获取事件的默认 SOQL 查询。

如果游标状态不可用，将使用默认查询从 Salesforce 实例获取事件。默认查询必须是有效的 SOQL 查询。如果 event_monitoring_method.object.query.value 中的 SOQL 查询无效，则将使用默认查询从 Salesforce 实例获取事件。

`event_monitoring_method.object.query.value`

编辑

使用来自对象的 REST API 从 Salesforce 实例获取事件的 SOQL 查询，但它使用游标状态从 Salesforce 实例获取事件。SOQL 查询必须是有效的 SOQL 查询。如果 SOQL 查询无效，则将使用默认查询从 Salesforce 实例获取事件。

在重新启动或后续执行的情况下，将使用游标状态从 Salesforce 实例获取事件。游标状态是从 Salesforce 实例获取的最后一个事件的最后一个事件时间。游标状态取自从 Salesforce 实例获取的最后一个事件的 event_monitoring_method.object.cursor.field 字段。

`event_monitoring_method.object.cursor.field`

编辑

用于从 Salesforce 实例获取的最后一个事件中获取游标状态的字段。该字段必须是 event_monitoring_method.object.query.default 和 event_monitoring_method.object.query.value 中指定的 SOQL 查询中的有效字段，即 SOQL 查询中选定字段的一部分。

常用选项

编辑

所有输入都支持以下配置选项。

`enabled`

编辑

使用 enabled 选项启用和禁用输入。默认情况下，enabled 设置为 true。

`tags`

编辑

Filebeat 在每个已发布事件的 tags 字段中包含的标签列表。标签可以轻松地在 Kibana 中选择特定事件或在 Logstash 中应用条件筛选。这些标签将附加到常规配置中指定的标签列表中。

示例

filebeat.inputs:
- type: salesforce
  . . .
  tags: ["json"]

`fields`

编辑

您可以指定的可选字段，用于向输出添加其他信息。例如，您可以添加可用于筛选日志数据的字段。字段可以是标量值、数组、字典或这些值的任何嵌套组合。默认情况下，您在此处指定的字段将分组在输出文档中的 fields 子字典下。要将自定义字段存储为顶级字段，请将 fields_under_root 选项设置为 true。如果在常规配置中声明了重复字段，则其值将被此处声明的值覆盖。

filebeat.inputs:
- type: salesforce
  . . .
  fields:
    app_id: query_engine_12

`fields_under_root`

编辑

如果此选项设置为 true，则自定义字段将作为输出文档中的顶级字段存储，而不是分组在 fields 子字典下。如果自定义字段名称与 Filebeat 添加的其他字段名称冲突，则自定义字段将覆盖其他字段。

`processors`

编辑

要应用于输入数据的处理器列表。

有关在配置中指定处理器的信息，请参阅处理器。

`pipeline`

编辑

为此输入生成的事件设置的 Ingest Pipeline ID。

也可以在 Elasticsearch 输出中配置 Pipeline ID，但此选项通常会生成更简单的配置文件。如果在输入和输出中都配置了 pipeline，则使用输入中的选项。

`keep_null`

编辑

如果此选项设置为 true，则在输出文档中发布具有 null 值的字段。默认情况下，keep_null 设置为 false。

`index`

编辑

如果存在，则此格式化的字符串会覆盖此输入事件的索引（对于 elasticsearch 输出），或者设置事件元数据的 raw_index 字段（对于其他输出）。此字符串只能引用代理名称和版本以及事件时间戳；要访问动态字段，请使用 output.elasticsearch.index 或处理器。

示例值："%{[agent.name]}-myindex-%{+yyyy.MM.dd}" 可能会扩展为 "filebeat-myindex-2019.11.01"。

`publisher_pipeline.disable_host`

编辑

默认情况下，所有事件都包含 host.name。可以将此选项设置为 true 以禁用向所有事件添加此字段。默认值为 false。

« Redis 输入 Stdin 输入 »