Salesforce input

Use the salesforce input to monitor Salesforce events using either the Salesforce EventLogFile (ELF) API or the Salesforce Real-time event monitoring API. Both use the REST API under the hood to run SOQL queries against the Salesforce instance and fetch the relevant events.

The Salesforce input maintains cursor states between requests to track collection progress for each execution. These cursor states are passed to the next event monitoring execution to resume fetching events from the last known position. The cursor states allow the input to pick up where it left off and provide control over the behavior of the input.

Here are some supported authentication methods and event monitoring methods:

• Authentication methods

  • OAuth2

    • User-Password flow
    • JWT Bearer flow

• Event monitoring methods

  • EventLogFile (ELF) using REST API
  • REST API for objects (For monitoring real-time events)

Here are some key points about how cursors are used in the Salesforce input:

• Separate cursor states are maintained for each configured event monitoring method (event_log_file and object).
• The cursor state stores the persisted watermark used by the query templates. In simple incremental queries this is usually the last observed cursor.field value. In bounded object batching it also includes object.progress_time.
• On the first run, the query.default is used to fetch an initial set of events unless the input precomputes a derived cursor for bounded batching.
• On subsequent runs, the query.value template is populated with the cursor state to fetch events since the last execution.
• If the input is restarted, it will resume from the last persisted cursor state rather than starting over from scratch.

Using cursors allows the Salesforce input to reliably keep track of its progress and avoid missing or duplicating events across executions. Choose cursor inputs that advance predictably between runs. For EventLogFile this usually means a monotonically increasing field such as CreatedDate, with Id as a same-timestamp tie-breaker. For bounded object catch-up, use object.progress_time together with object.batch_start_time and object.batch_end_time.

Event Monitoring methods are highly configurable and can be used to monitor any supported object or event log file. The input can be configured to monitor multiple objects or event log files at the same time.

Example configuration:

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: salesforce-instance@user.in
        password: salesforce-instance-password
      jwt_bearer_flow:
        enabled: true
        client.id: client-id
        client.username: salesforce-instance@user.in
        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, Id ASC"
          value: "SELECT Id,CreatedDate,LogDate,LogFile FROM EventLogFile WHERE EventType = 'Login' AND (CreatedDate > [[ .cursor.event_log_file.last_event_time ]][[ if .cursor.event_log_file.last_event_id ]] OR (CreatedDate = [[ .cursor.event_log_file.last_event_time ]] AND Id > '[[ .cursor.event_log_file.last_event_id ]]')[[ end ]]) ORDER BY CreatedDate ASC NULLS FIRST, Id ASC"
        cursor:
          field: "CreatedDate"
      object:
        enabled: true
        interval: 5m
        batch:
          enabled: true
          initial_interval: 24h
          window: 5m
          max_windows_per_run: 12
        query:
          default: "SELECT FIELDS(STANDARD) FROM LoginEvent ORDER BY EventDate DESC"
          value: "SELECT FIELDS(STANDARD) FROM LoginEvent WHERE EventDate > [[ .cursor.object.batch_start_time ]] AND EventDate <= [[ .cursor.object.batch_end_time ]] ORDER BY EventDate DESC"
        cursor:
          field: "EventDate"
		

In order to use this integration, users need to create a new Salesforce Application using OAuth. Follow the steps below to create a connected application in Salesforce:

1. Login to Salesforce with the same user credentials that the user wants to collect data with.

2. Click on Setup on the top right menu bar. On the Setup page, search for App Manager in the Search Setup search box at the top of the page, then select App Manager.

3. Click New Connected App.

4. Provide a name for the connected application. This will be displayed in the App Manager and on its App Launcher tile.

5. Enter the API name. The default is a version of the name without spaces. Only letters, numbers, and underscores are allowed. If the original app name contains any other characters, edit the default name.

6. Enter the contact email for Salesforce.

7. Under the API (Enable OAuth Settings) section of the page, select Enable OAuth Settings.

8. In the Callback URL, enter the Instance URL (Please refer to Salesforce Instance URL).

9. Select the following OAuth scopes to apply to the connected app:

   • Manage user data via APIs (api).
   • Perform requests at any time (refresh_token, offline_access).
   • (Optional) In case of data collection, if any permission issues arise, add the Full access (full) scope.

10. Select Require Secret for the Web Server Flow to require the app’s client secret in exchange for an access token.

11. Select Require Secret for Refresh Token Flow to require the app’s client secret in the authorization request of a refresh token and hybrid refresh token flow.

12. Click Save. It may take approximately 10 minutes for the changes to take effect.

13. Click Continue and then under API details, click Manage Consumer Details. Verify the user account using the Verification Code.

14. Copy Consumer Key and Consumer Secret from the Consumer Details section, which should be populated as values for Client ID and Client Secret respectively in the configuration.

For more details on how to create a Connected App, refer to the Salesforce documentation here.

The salesforce input is a long-running program that retrieves events from a Salesforce instance and sends them to the specified output. The program executes in a loop, fetching events from the Salesforce instance at a preconfigured interval. Each event monitoring method can be configured to run separately and at different intervals.

When multiple event monitoring methods are enabled, they run in the same input loop (single goroutine). After the initial one-shot collection runs, the steady-state ticker loop handles one ready method at a time. If both method tickers are ready at the same time, Go's select chooses one ready case non-deterministically, and that collection runs to completion before the loop handles the next ready ticker. In practice, this means long-running collection in one method can delay the other method.

There are two methods to fetch the events from the Salesforce instance:

• event_log_file: EventLogFile is a standard object in Salesforce and the event monitoring method uses the REST API under the hood to gather the Salesforce org’s operational events from the object. There is a field EventType that helps distinguish between the types of operational events like — Login, Logout, etc. Uses Salesforce’s query language SOQL to query the object. Keep the query filter, sort order, and cursor.field aligned on the same watermark field. The built-in module templates use CreatedDate, with Id as a same-timestamp tie-breaker, and downloaded CSV rows are processed as a stream to reduce memory pressure.
• object: This method is a general way of retrieving events from a Salesforce instance by using the REST API. It can be used for monitoring objects in real-time. In real-time event monitoring, subscribing to the events is a common practice, but the events are also stored in Salesforce org (if configured), specifically in big object tables that are preconfigured for each event type. With this method, we query the object using Salesforce’s query language (SOQL). The collection happens at the configured scrape interval. For high-volume objects, enable bounded batching so one large catch-up query does not have to process the full backlog at once.

The salesforce input supports the following configuration options plus the Common options described later.

Whether the input is enabled or not. Default: false.

The version of the Salesforce API to use. Minimum supported version is 46.

The authentication settings for the Salesforce instance.

The OAuth2 authentication options for the Salesforce instance.

There are two OAuth2 authentication flows supported:

• user_password_flow: User-Password flow
• jwt_bearer_flow: JWT Bearer flow

Whether to use the user-password flow for authentication. Default: false.

The client ID for the user-password flow.

The client secret for the user-password flow.

The Salesforce OAuth host or canonical token endpoint for the user-password flow. You can provide either a host URL such as https://login.salesforce.com, https://test.salesforce.com, or your My Domain host, or the canonical full endpoint ending in https://login.salesforce.com/services/oauth2/token. Avoid query parameters and fragments on this value.

The username for the user-password flow.

The password for the user-password flow.

Whether to use the JWT bearer flow for authentication. Default: false.

The client ID for the JWT bearer flow.

The username for the JWT bearer flow.

The path to the private key file for the JWT bearer flow. The file must be PEM encoded PKCS1 or PKCS8 private key and must have the right permissions set to have read access for the user running the program.

The URL for the JWT bearer flow.

Optional Salesforce OAuth host or canonical token endpoint override for the JWT bearer flow. You can provide either a host URL such as https://login.salesforce.com, https://test.salesforce.com, or your My Domain host, or the canonical full endpoint ending in https://login.salesforce.com/services/oauth2/token. Avoid query parameters and fragments on this value. If omitted, auth.oauth2.jwt_bearer_flow.url is used.

The URL of the Salesforce instance. Required.

Duration before declaring that the HTTP client connection has timed out. Valid time units are ns, us, ms, s, m, h. Default: 30s.

The maximum number of retries for the HTTP client. Default: 5.

The minimum time to wait before a retry is attempted. Default: 1s.

The maximum time to wait before a retry is attempted. Default: 60s.

The event monitoring method to use. There are two event monitoring methods supported:

• event_log_file: EventLogFile (ELF) using REST API
• object: Real-time event monitoring using REST API (objects)

The event monitoring method to use — event_log_file. Uses the EventLogFile API to fetch the events from the Salesforce instance.

Whether to use the EventLogFile API for event monitoring. Default: false.

The interval to collect the events from the Salesforce instance using the EventLogFile API.

The default query to fetch the events from the Salesforce instance using the EventLogFile API.

In case the cursor state is not available, the default query will be used to fetch the events from the Salesforce instance. The default query must be a valid SOQL query.

The SOQL query to fetch the events from the Salesforce instance using the EventLogFile API but it uses the cursor state to fetch the events from the Salesforce instance. The SOQL query must be a valid SOQL query.

In case of restarts or subsequent executions, the cursor state will be used to fetch the events from the Salesforce instance. The cursor state is the last event time of the last event fetched from the Salesforce instance, optionally paired with event_log_file.last_event_id as a same-timestamp tie-breaker. The cursor state is taken from event_monitoring_method.event_log_file.cursor.field field for the last event fetched from the Salesforce instance.

The default and value queries should keep their filter field, sort field, and cursor.field aligned. For example, if cursor.field is CreatedDate, both queries should filter on CreatedDate and use ORDER BY CreatedDate. If multiple records can share the same cursor value, select Id, add it as a secondary sort, and compare it with event_log_file.last_event_id in the value query.

The field to use to fetch the cursor state from the last event fetched from the Salesforce instance. The field must be a valid field in the SOQL query specified in event_monitoring_method.event_log_file.query.default and event_monitoring_method.event_log_file.query.value i.e., part of the selected fields in the SOQL query.

The event monitoring method to use — object. Uses REST API to fetch the events directly from the objects from the Salesforce instance.

Whether to use the REST API for objects for event monitoring. Default: false.

The interval to collect the events from the Salesforce instance using the REST API from objects.

Whether to split object catch-up into bounded SOQL windows. Default: false.

The historical lookback used to seed the first bounded object window when batching is enabled. The first batched query starts at now - initial_interval.

When you use the built-in Salesforce module templates for login and logout, leaving var.initial_interval unset causes the generated object batching configuration to fall back to var.real_time_interval for the first bounded window. Set var.initial_interval explicitly if you want a deeper first-run catch-up window.

The size of each bounded object batch window. Each run queries at most window worth of object data per batch.

The maximum number of bounded object batch windows to execute in one input run before yielding to the next interval. Default: 1.

The default SOQL query to fetch the events from the Salesforce instance using the REST API from objects.

In case the cursor state is not available, the default query will be used to fetch the events from the Salesforce instance. The default query must be a valid SOQL query.

When object batching is enabled, the input seeds the first batch window from event_monitoring_method.object.batch.initial_interval and executes the value query immediately with derived cursor inputs instead of using an unbounded default query.

The SOQL query to fetch the events from the Salesforce instance using the REST API from objects but it uses the cursor state to fetch the events from the Salesforce instance. The SOQL query must be a valid SOQL query.

In case of restarts or subsequent executions, the cursor state will be used to fetch the events from the Salesforce instance. The cursor state is the last event time of the last event fetched from the Salesforce instance. The cursor state is taken from event_monitoring_method.object.cursor.field field for the last event fetched from the Salesforce instance.

When object batching is enabled, the value query can use object.progress_time, object.batch_start_time, and object.batch_end_time to build bounded windows. This is the recommended pattern for high-volume LoginEvent and LogoutEvent collection.

The field to use to fetch the cursor state from the last event fetched from the Salesforce instance. The field must be a valid field in the SOQL query specified in event_monitoring_method.object.query.default and event_monitoring_method.object.query.value i.e., part of the selected fields in the SOQL query.

For high-volume objects, the query should also order on the same field whenever Salesforce supports it. LoginEvent and LogoutEvent should use ORDER BY EventDate DESC.

The following configuration options are supported by all inputs.

Use the enabled option to enable and disable inputs. By default, enabled is set to true.

A list of tags that Filebeat includes in the tags field of each published event. Tags make it easy to select specific events in Kibana or apply conditional filtering in Logstash. These tags will be appended to the list of tags specified in the general configuration.

Example:

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

Optional fields that you can specify to add additional information to the output. For example, you might add fields that you can use for filtering log data. Fields can be scalar values, arrays, dictionaries, or any nested combination of these. By default, the fields that you specify here will be grouped under a fields sub-dictionary in the output document. To store the custom fields as top-level fields, set the fields_under_root option to true. If a duplicate field is declared in the general configuration, then its value will be overwritten by the value declared here.

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

If this option is set to true, the custom fields are stored as top-level fields in the output document instead of being grouped under a fields sub-dictionary. If the custom field names conflict with other field names added by Filebeat, then the custom fields overwrite the other fields.

A list of processors to apply to the input data.

See Processors for information about specifying processors in your config.

The ingest pipeline ID to set for the events generated by this input.

If this option is set to true, fields with null values will be published in the output document. By default, keep_null is set to false.

If present, this formatted string overrides the index for events from this input (for elasticsearch outputs), or sets the raw_index field of the event’s metadata (for other outputs). This string can only refer to the agent name and version and the event timestamp; for access to dynamic fields, use output.elasticsearch.index or a processor.

Example value: "%{[agent.name]}-myindex-%{+yyyy.MM.dd}" might expand to "filebeat-myindex-2019.11.01".

By default, all events contain host.name. This option can be set to true to disable the addition of this field to all events. The default value is false.