Skip to content

refactor: Add support for plain SOQL query #33

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,17 @@
## 4.0.0
- refactor: Change behavior to support SOQL queries
- feat: Include deleted records

## 3.4.0
- feat: Add support for plain SOQL query

## 3.3.0
- feat: Added `timeout` configuration to control RESTForce timeout settings (defaults to `60`)
- feat: Added support for reference fields (`parent__r.child`)
- feat: Added support for built-in SOQL functions (etc. `toLabel(field__c) field`)
- refactor: Removed field_type inferring (elastic can infer it / logstash mapping can be configured for special cases)
- fix: update Apache licence to suppress compiling warning

## 3.2.0
- Added `use_tooling_api` configuration to connect to the Salesforce Tooling API instead of the regular Rest API. [#26](https://github.com/logstash-plugins/logstash-input-salesforce/pull/26)

Expand Down
76 changes: 30 additions & 46 deletions lib/logstash/inputs/salesforce.rb
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
# username => '[email protected]'
# password => 'super-secret'
# security_token => 'SECURITY TOKEN FOR THIS USER'
# sfdc_object_name => 'Opportunity'
# sfdc_soql_query => 'SELECT Id FROM Account'
# }
# }
#
Expand All @@ -58,6 +58,8 @@ class LogStash::Inputs::Salesforce < LogStash::Inputs::Base
# Set this to true to connect to a sandbox sfdc instance
# logging in through test.salesforce.com
config :use_test_sandbox, :validate => :boolean, :default => false
# Include deleted records
config :include_deleted, :validate => :boolean, :default => false
# Set this to the instance url of the sfdc instance you want
# to connect to already during login. If you have configured
# a MyDomain in your sfdc instance you would provide
Expand All @@ -66,6 +68,8 @@ class LogStash::Inputs::Salesforce < LogStash::Inputs::Base
# By default, this uses the default Restforce API version.
# To override this, set this to something like "32.0" for example
config :api_version, :validate => :string, :required => false
# RESTForce request timeout in seconds.
config :timeout, :validate => :number, :required => false
# Consumer Key for authentication. You must set up a new SFDC
# connected app with oath to use this output. More information
# can be found here:
Expand All @@ -83,45 +87,41 @@ class LogStash::Inputs::Salesforce < LogStash::Inputs::Base
# generting a security token, see:
# https://help.salesforce.com/apex/HTViewHelpDoc?id=user_security_token.htm
config :security_token, :validate => :string, :required => true
# The name of the salesforce object you are creating or updating
config :sfdc_object_name, :validate => :string, :required => true
# These are the field names to return in the Salesforce query
# If this is empty, all fields are returned.
config :sfdc_fields, :validate => :array, :default => []
# These options will be added to the WHERE clause in the
# SOQL statement. Additional fields can be filtered on by
# adding field1 = value1 AND field2 = value2 AND...
config :sfdc_filters, :validate => :string, :default => ""
# Plain SOQL query
config :sfdc_soql_query, :validate => :string, :required => true
# Setting this to true will convert SFDC's NamedFields__c to named_fields__c
config :to_underscores, :validate => :boolean, :default => false

public
def register
require 'restforce'
obj_desc = client.describe(@sfdc_object_name)
@sfdc_field_types = get_field_types(obj_desc)
@sfdc_fields = get_all_fields if @sfdc_fields.empty?
@sfdc_fields = get_sfdc_fields
end # def register

public
def run(queue)
results = client.query(get_query())
if @include_deleted
results = client.query_all(@sfdc_soql_query)
else
results = client.query(@sfdc_soql_query)
end

@logger.debug("Query results:", :results => results)
if results && results.first
results.each do |result|
event = LogStash::Event.new()
decorate(event)
@sfdc_fields.each do |field|
field_type = @sfdc_field_types[field]
# PARENT.CHILD => PARENT
# function(field) field => field
field = field.split(/\./).first.split(/\s/).last
value = result.send(field)

# Remove RESTForce's nested 'attributes' field for reference fields
value.is_a?(Hash) ? value = value.tap { |hash| hash.delete(:attributes)} : value

event_key = @to_underscores ? underscore(field) : field
if not value.nil?
case field_type
when 'datetime', 'date'
event.set(event_key, format_time(value))
else
event.set(event_key, value)
end
end
event.set(event_key, value)
end
queue << event
end
Expand Down Expand Up @@ -155,33 +155,17 @@ def client_options
options.merge!({ :host => "test.salesforce.com" })
end
options.merge!({ :api_version => @api_version }) if @api_version
options.merge!({ :timeout => @timeout }) if @timeout
return options
end

private
def get_query()
query = ["SELECT",@sfdc_fields.join(','),
"FROM",@sfdc_object_name]
query << ["WHERE",@sfdc_filters] unless @sfdc_filters.empty?
query << "ORDER BY LastModifiedDate DESC" if @sfdc_fields.include?('LastModifiedDate')
query_str = query.flatten.join(" ")
@logger.debug? && @logger.debug("SFDC Query", :query => query_str)
return query_str
end

private
def get_field_types(obj_desc)
field_types = {}
obj_desc.fields.each do |f|
field_types[f.name] = f.type
end
@logger.debug? && @logger.debug("Field types", :field_types => field_types.to_s)
return field_types
end

private
def get_all_fields
return @sfdc_field_types.keys
def get_sfdc_fields
extracted_fields = sfdc_soql_query.gsub(/SELECT (.+) FROM.+/i, '\1')
extracted_fields_as_array = extracted_fields.split(',').collect { |item| item.strip }
@logger.debug("Extracted fields: ", :extracted_fields_as_array => extracted_fields_as_array.to_s)

return extracted_fields_as_array;
end

private
Expand Down
6 changes: 3 additions & 3 deletions logstash-input-salesforce.gemspec
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
Gem::Specification.new do |s|
s.name = 'logstash-input-salesforce'
s.version = '3.2.0'
s.licenses = ['Apache License (2.0)']
s.version = '4.0.0'
s.licenses = ['Apache-2.0']
s.summary = "Creates events based on a Salesforce SOQL query"
s.description = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
s.authors = ["Russ Savage"]
Expand All @@ -20,7 +20,7 @@ Gem::Specification.new do |s|
# Gem dependencies
s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
s.add_runtime_dependency "logstash-codec-plain"
s.add_runtime_dependency "restforce", ">= 5", "< 5.2"
s.add_runtime_dependency "restforce", ">= 5", "< 5.3"
s.add_development_dependency 'logstash-devutils'
s.add_development_dependency 'vcr'
s.add_development_dependency 'webmock'
Expand Down