Feature/support managed identity (#57)
* * Attempt to address issue #50 with managed identity * * Update version * Fix review comments
This commit is contained in:
Ramachandran A G
GitHub
Родитель
f27b03097f
Коммит
33cbb65732
|
|
@ -53,7 +53,8 @@ More information about configuring Logstash can be found in the [logstash config
|
|||
| --- | --- | --- |
|
||||
| **path** | The plugin writes events to temporary files before sending them to ADX. This parameter includes a path where files should be written and a time expression for file rotation to trigger an upload to the ADX service. The example above shows how to rotate the files every minute and check the Logstash docs for more information on time expressions. | Required
|
||||
| **ingest_url** | The Kusto endpoint for ingestion-related communication. See it on the Azure Portal.| Required|
|
||||
| **app_id, app_key, app_tenant**| Credentials required to connect to the ADX service. Be sure to use an application with 'ingest' privileges. | Required|
|
||||
| **app_id, app_key, app_tenant**| Credentials required to connect to the ADX service. Be sure to use an application with 'ingest' privileges. | Optional|
|
||||
| **managed_identity**| Managed Identity to authenticate. For user-based managed ID, use the Client ID GUID. For system-based, use the value `system`. The ID needs to have 'ingest' privileges on the cluster. | Optional|
|
||||
| **database**| Database name to place events | Required |
|
||||
| **table** | Target table name to place events | Required
|
||||
| **json_mapping** | Maps each attribute from incoming event JSON strings to the appropriate column in the table. Note that this must be in JSON format, as this is the interface between Logstash and Kusto | Required |
|
||||
|
|
|
|||
|
|
@ -73,11 +73,13 @@ class LogStash::Outputs::Kusto < LogStash::Outputs::Base
|
|||
|
||||
# The following are the credentails used to connect to the Kusto service
|
||||
# application id
|
||||
config :app_id, validate: :string, required: true
|
||||
config :app_id, validate: :string, required: false
|
||||
# application key (secret)
|
||||
config :app_key, validate: :password, required: true
|
||||
config :app_key, validate: :password, required: false
|
||||
# aad tenant id
|
||||
config :app_tenant, validate: :string, default: nil
|
||||
# managed identity id
|
||||
config :managed_identity, validate: :string, default: nil
|
||||
|
||||
# The following are the data settings that impact where events are written to
|
||||
# Database name
|
||||
|
|
@ -150,7 +152,7 @@ class LogStash::Outputs::Kusto < LogStash::Outputs::Base
|
|||
max_queue: upload_queue_size,
|
||||
fallback_policy: :caller_runs)
|
||||
|
||||
@ingestor = Ingestor.new(ingest_url, app_id, app_key, app_tenant, database, table, final_mapping, delete_temp_files, proxy_host, proxy_port,proxy_protocol, @logger, executor)
|
||||
@ingestor = Ingestor.new(ingest_url, app_id, app_key, app_tenant, managed_identity, database, table, final_mapping, delete_temp_files, proxy_host, proxy_port,proxy_protocol, @logger, executor)
|
||||
|
||||
# send existing files
|
||||
recover_past_files if recovery
|
||||
|
|
|
|||
|
|
@ -20,15 +20,34 @@ class LogStash::Outputs::Kusto < LogStash::Outputs::Base
|
|||
LOW_QUEUE_LENGTH = 3
|
||||
FIELD_REF = /%\{[^}]+\}/
|
||||
|
||||
def initialize(ingest_url, app_id, app_key, app_tenant, database, table, json_mapping, delete_local, proxy_host , proxy_port , proxy_protocol,logger, threadpool = DEFAULT_THREADPOOL)
|
||||
def initialize(ingest_url, app_id, app_key, app_tenant, managed_identity_id, database, table, json_mapping, delete_local, proxy_host , proxy_port , proxy_protocol,logger, threadpool = DEFAULT_THREADPOOL)
|
||||
@workers_pool = threadpool
|
||||
@logger = logger
|
||||
validate_config(database, table, json_mapping,proxy_protocol)
|
||||
validate_config(database, table, json_mapping,proxy_protocol,app_id, app_key, managed_identity_id)
|
||||
@logger.info('Preparing Kusto resources.')
|
||||
|
||||
kusto_java = Java::com.microsoft.azure.kusto
|
||||
apache_http = Java::org.apache.http
|
||||
kusto_connection_string = kusto_java.data.auth.ConnectionStringBuilder.createWithAadApplicationCredentials(ingest_url, app_id, app_key.value, app_tenant)
|
||||
# kusto_connection_string = kusto_java.data.auth.ConnectionStringBuilder.createWithAadApplicationCredentials(ingest_url, app_id, app_key.value, app_tenant)
|
||||
# If there is managed identity, use it. This means the AppId and AppKey are empty/nil
|
||||
is_managed_identity = (app_id.nil? && app_key.empty?)
|
||||
# If it is system managed identity, propagate the system identity
|
||||
is_system_assigned_managed_identity = is_managed_identity && 0 == "system".casecmp(managed_identity_id)
|
||||
# Is it direct connection
|
||||
is_direct_conn = (proxy_host.nil? || proxy_host.empty?)
|
||||
# Create a connection string
|
||||
kusto_connection_string = if is_managed_identity
|
||||
if is_system_assigned_managed_identity
|
||||
@logger.info('Using system managed identity.')
|
||||
kusto_java.data.auth.ConnectionStringBuilder.createWithAadManagedIdentity(ingest_url)
|
||||
else
|
||||
@logger.info('Using user managed identity.')
|
||||
kusto_java.data.auth.ConnectionStringBuilder.createWithAadManagedIdentity(ingest_url, managed_identity_id)
|
||||
end
|
||||
else
|
||||
kusto_java.data.auth.ConnectionStringBuilder.createWithAadApplicationCredentials(ingest_url, app_id, app_key.value, app_tenant)
|
||||
end
|
||||
|
||||
#
|
||||
@logger.debug(Gem.loaded_specs.to_s)
|
||||
# Unfortunately there's no way to avoid using the gem/plugin name directly...
|
||||
|
|
@ -41,7 +60,7 @@ class LogStash::Outputs::Kusto < LogStash::Outputs::Base
|
|||
kusto_connection_string.setConnectorDetails("Logstash",version_for_tracing.to_s,"","",false,"", tuple_utils.Pair.emptyArray());
|
||||
|
||||
@kusto_client = begin
|
||||
if proxy_host.nil? || proxy_host.empty?
|
||||
if is_direct_conn
|
||||
kusto_java.ingest.IngestClientFactory.createClient(kusto_connection_string)
|
||||
else
|
||||
kusto_http_client_properties = kusto_java.data.HttpClientProperties.builder().proxy(apache_http.HttpHost.new(proxy_host,proxy_port,proxy_protocol)).build()
|
||||
|
|
@ -57,7 +76,12 @@ class LogStash::Outputs::Kusto < LogStash::Outputs::Base
|
|||
@logger.debug('Kusto resources are ready.')
|
||||
end
|
||||
|
||||
def validate_config(database, table, json_mapping,proxy_protocol)
|
||||
def validate_config(database, table, json_mapping, proxy_protocol, app_id, app_key, managed_identity_id)
|
||||
# Add an additional validation and fail this upfront
|
||||
if app_id.nil? && app_key.empty? && managed_identity_id.empty?
|
||||
@logger.error('managed_identity_id is not provided and app_id/app_key is empty.')
|
||||
raise LogStash::ConfigurationError.new('managed_identity_id is not provided and app_id/app_key is empty.')
|
||||
end
|
||||
if database =~ FIELD_REF
|
||||
@logger.error('database config value should not be dynamic.', database)
|
||||
raise LogStash::ConfigurationError.new('database config value should not be dynamic.')
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ describe LogStash::Outputs::Kusto::Ingestor do
|
|||
let(:app_id) { "myid" }
|
||||
let(:app_key) { LogStash::Util::Password.new("mykey") }
|
||||
let(:app_tenant) { "mytenant" }
|
||||
let(:managed_identity) { "managed_identity" }
|
||||
let(:database) { "mydatabase" }
|
||||
let(:table) { "mytable" }
|
||||
let(:proxy_host) { "localhost" }
|
||||
|
|
@ -24,7 +25,7 @@ describe LogStash::Outputs::Kusto::Ingestor do
|
|||
# note that this will cause an internal error since connection is being tried.
|
||||
# however we still want to test that all the java stuff is working as expected
|
||||
expect {
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol, logger)
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol, logger)
|
||||
ingestor.stop
|
||||
}.not_to raise_error
|
||||
end
|
||||
|
|
@ -35,7 +36,7 @@ describe LogStash::Outputs::Kusto::Ingestor do
|
|||
dynamic_name_array.each do |test_database|
|
||||
it "with database: #{test_database}" do
|
||||
expect {
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, test_database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity, test_database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
|
||||
ingestor.stop
|
||||
}.to raise_error(LogStash::ConfigurationError)
|
||||
end
|
||||
|
|
@ -46,7 +47,7 @@ describe LogStash::Outputs::Kusto::Ingestor do
|
|||
dynamic_name_array.each do |test_table|
|
||||
it "with database: #{test_table}" do
|
||||
expect {
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, database, test_table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity,database, test_table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
|
||||
ingestor.stop
|
||||
}.to raise_error(LogStash::ConfigurationError)
|
||||
end
|
||||
|
|
@ -57,7 +58,7 @@ describe LogStash::Outputs::Kusto::Ingestor do
|
|||
dynamic_name_array.each do |json_mapping|
|
||||
it "with database: #{json_mapping}" do
|
||||
expect {
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity,database, table, json_mapping, delete_local, proxy_host, proxy_port,proxy_protocol,logger)
|
||||
ingestor.stop
|
||||
}.to raise_error(LogStash::ConfigurationError)
|
||||
end
|
||||
|
|
@ -67,7 +68,16 @@ describe LogStash::Outputs::Kusto::Ingestor do
|
|||
context 'proxy protocol has to be http or https' do
|
||||
it "with proxy protocol: socks" do
|
||||
expect {
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, database, table, json_mapping, delete_local, proxy_host, proxy_port,'socks',logger)
|
||||
ingestor = described_class.new(ingest_url, app_id, app_key, app_tenant, managed_identity,database, table, json_mapping, delete_local, proxy_host, proxy_port,'socks',logger)
|
||||
ingestor.stop
|
||||
}.to raise_error(LogStash::ConfigurationError)
|
||||
end
|
||||
end
|
||||
|
||||
context 'one of appid or managedid has to be provided' do
|
||||
it "with empty managed identity and appid" do
|
||||
expect {
|
||||
ingestor = described_class.new(ingest_url, "", app_key, app_tenant, "",database, table, json_mapping, delete_local, proxy_host, proxy_port,'socks',logger)
|
||||
ingestor.stop
|
||||
}.to raise_error(LogStash::ConfigurationError)
|
||||
end
|
||||
|
|
|
|||
2
version
2
version
|
|
@ -1 +1 @@
|
|||
2.0.0
|
||||
2.0.1
|
||||
Загрузка…
Ссылка в новой задаче