Class: Fluent::Plugin::DruidQueryInput

Inherits:
Input
  • Object
show all
Defined in:
lib/fluent/plugin/in_druid_query.rb

Constant Summary collapse

NAME = 
'druid_query'
DEFAULT_TAG = 
NAME
DEFAULT_INTERVAL = 
300
DEFAULT_DRUID_URL = 
'http://localhost:8888'
DEFAULT_DRUID_USER = 
nil
DEFAULT_DRUID_PASSWORD = 
nil
DEFAULT_DRUID_VERIFY_SSL = 
true
DEFAULT_DRUID_USER_AGENT = 
NAME
DEFAULT_DRUID_TIMEOUT = 
30
DEFAULT_CA_CERT = 
nil
DEFAULT_QUERY_CACHE = 
nil
DEFAULT_QUERY_PRIORITY = 
nil
DEFAULT_QUERY_SUBTAG = 
nil
DEFAULT_QUERY_GENERATE_RECORD = 
true
DEFAULT_QUERY_GENERATE_INFO = 
false

Instance Method Summary collapse

Instance Method Details

#check_druid_information ⇒ Object

Raises:

  • (Fluent::ConfigError) 


91
92
93
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 91

def check_druid_information
  raise Fluent::ConfigError, 'druid_url should not be empty' if druid_url.empty?
end

#check_druid_queries ⇒ Object

Raises:

  • (Fluent::ConfigError) 


95
96
97
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 95

def check_druid_queries
  raise Fluent::ConfigError, 'queries should not be empty' if queries.empty?
end

#configure(conf) ⇒ Object

Raises:

  • (Fluent::ConfigError) 


81
82
83
84
85
86
87
88
89
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 81

def configure(conf)
  super

  raise Fluent::ConfigError, 'tag should not be empty' if tag.empty?
  raise Fluent::ConfigError, 'tag_info should not be empty' if tag_info.empty?

  check_druid_information
  check_druid_queries
end

#druid_client ⇒ Object 



135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 135

def druid_client
  @druid_client ||= DruidClient::Api.new(
    url: druid_url,
    username: druid_user,
    password: druid_password,
    user_agent: druid_user_agent,
    verify_ssl: druid_verify_ssl,
    timeout: druid_timeout,
    ca_file: ca_cert,
    log: log
  )
end

#emit_query_info(query:, response:, query_time: Fluent::Engine.now) ⇒ Object 



160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 160

def emit_query_info(query:, response:, query_time: Fluent::Engine.now)
  return unless query.generate_info

  current_tag = [tag_info, query.subtag].compact.join('.')
  info_record = {
    'timestamp' => query_time.to_time.utc.iso8601(3),
    'status' => response.success? ? 'success' : 'failure',
    'status_code' => response.status_code,
    'query_duration' => response.duration
  }
  info_record.update({ 'response_rows_count' => response.body.size }) if response.success?
  info_record.update(query.metadata)
  router.emit(current_tag, query_time, info_record)
end

#emit_query_records(query:, response:, query_time: Fluent::Engine.now) ⇒ Object 



148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 148

def emit_query_records(query:, response:, query_time: Fluent::Engine.now)
  return unless query.generate_record
  return unless response.success?

  current_tag = [tag, query.subtag].compact.join('.')
  query_events = MultiEventStream.new
  response.body.each do |response_entry|
    query_events.add(query_time, response_entry.merge(query.metadata))
  end
  router.emit_stream(current_tag, query_events)
end

#query_context(use_cache: true, priority: 0) ⇒ Object 



125
126
127
128
129
130
131
132
133
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 125

def query_context(use_cache: true, priority: 0)
  {
    useCache: use_cache,
    populateCache: use_cache,
    useResultLevelCache: use_cache,
    populateResultLevelCache: use_cache,
    priority: priority
  }.compact
end

#run_queries ⇒ Object 



106
107
108
109
110
111
112
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 106

def run_queries
  queries.each do |query|
    run_query(query)
  rescue StandardError => e
    log.error "while runnig query: #{query.sql}: #{e}"
  end
end

#run_query(query) ⇒ Object 



114
115
116
117
118
119
120
121
122
123
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 114

def run_query(query)
  query_time = Fluent::Engine.now
  response = druid_client.sql.query(
    query: query.sql,
    header: false,
    context: query_context(use_cache: query.cache, priority: query.priority)
  )
  emit_query_records(query_time: query_time, query: query, response: response)
  emit_query_info(query_time: query_time, query: query, response: response)
end

#start ⇒ Object 



99
100
101
102
103
104
 
# File 'lib/fluent/plugin/in_druid_query.rb', line 99

def start
  super

  timer_execute(:run_queries_first, 1, repeat: false, &method(:run_queries)) if interval > 60
  timer_execute(:run_queries, interval, repeat: true, &method(:run_queries))
end