Class: LogStash::Inputs::CloudWatch_Logs

Inherits:
Base
  • Object
show all
Includes:
PluginMixins::AwsConfig::V2
Defined in:
lib/logstash/inputs/cloudwatch_logs.rb

Overview

Stream events from CloudWatch Logs streams.

Specify an individual log group, and this plugin will scan all log streams in that group, and pull in any new log events.

Optionally, you may set the ‘log_group_prefix` parameter to true which will scan for all log groups matching the specified prefix and ingest all logs available in all of the matching groups.

Instance Method Summary collapse

Instance Method Details

#check_start_position_validity ⇒ Object

Raises:

  • (LogStash::ConfigurationError) 


100
101
102
103
104
105
106
107
 
# File 'lib/logstash/inputs/cloudwatch_logs.rb', line 100

def check_start_position_validity
  raise LogStash::ConfigurationError, "No start_position specified!" unless @start_position

  return if @start_position =~ /^(beginning|end)$/
  return if @start_position.is_a? Integer

  raise LogStash::ConfigurationError, "start_position '#{@start_position}' is invalid! Must be `beginning`, `end`, or an integer."
end

#determine_start_position(groups, sincedb) ⇒ Object 



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/logstash/inputs/cloudwatch_logs.rb', line 162

def determine_start_position(groups, sincedb)
  groups.each do |group|
    if !sincedb.member?(group)
      case @start_position
        when 'beginning'
          sincedb[group] = 0

        when 'end'
          sincedb[group] = DateTime.now.strftime('%Q')

        else
          sincedb[group] = DateTime.now.strftime('%Q').to_i - (@start_position * 1000)
      end # case @start_position
    end
  end
end

#find_log_groups ⇒ Object 



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/logstash/inputs/cloudwatch_logs.rb', line 134

def find_log_groups
  if @log_group_prefix
    @logger.debug("log_group prefix is enabled, searching for log groups")
    groups = []
    next_token = nil
    @log_group.each do |group|
      loop do
        log_groups = @cloudwatch.describe_log_groups(log_group_name_prefix: group, next_token: next_token)
        groups += log_groups.log_groups.map {|n| n.log_group_name}
        next_token = log_groups.next_token
        @logger.debug("found #{log_groups.log_groups.length} log groups matching prefix #{group}")
        break if next_token.nil?
      end
    end
  else
    @logger.debug("log_group_prefix not enabled")
    groups = @log_group
  end
  # Move the most recent groups to the end
  groups.sort{|a,b| priority_of(a) <=> priority_of(b) }
end

#register ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/logstash/inputs/cloudwatch_logs.rb', line 56

def register
  require "digest/md5"
  @logger.debug("Registering cloudwatch_logs input", :log_group => @log_group)
  settings = defined?(LogStash::SETTINGS) ? LogStash::SETTINGS : nil
  @sincedb = {}

  check_start_position_validity

  Aws::ConfigService::Client.new(aws_options_hash)
  @cloudwatch = Aws::CloudWatchLogs::Client.new(aws_options_hash)

  if @sincedb_path.nil?
    if settings
      datapath = File.join(settings.get_value("path.data"), "plugins", "inputs", "cloudwatch_logs")
      # Ensure that the filepath exists before writing, since it's deeply nested.
      FileUtils::mkdir_p datapath
      @sincedb_path = File.join(datapath, ".sincedb_" + Digest::MD5.hexdigest(@log_group.join(",")))
    end
  end

  # This section is going to be deprecated eventually, as path.data will be
  # the default, not an environment variable (SINCEDB_DIR or HOME)
  if @sincedb_path.nil? # If it is _still_ nil...
    if ENV["SINCEDB_DIR"].nil? && ENV["HOME"].nil?
      @logger.error("No SINCEDB_DIR or HOME environment variable set, I don't know where " \
                    "to keep track of the files I'm watching. Either set " \
                    "HOME or SINCEDB_DIR in your environment, or set sincedb_path in " \
                    "in your Logstash config for the file input with " \
                    "path '#{@path.inspect}'")
      raise
    end

    #pick SINCEDB_DIR if available, otherwise use HOME
    sincedb_dir = ENV["SINCEDB_DIR"] || ENV["HOME"]

    @sincedb_path = File.join(sincedb_dir, ".sincedb_" + Digest::MD5.hexdigest(@log_group.join(",")))

    @logger.info("No sincedb_path set, generating one based on the log_group setting",
                 :sincedb_path => @sincedb_path, :log_group => @log_group)
  end

end

#run(queue) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/logstash/inputs/cloudwatch_logs.rb', line 111

def run(queue)
  @queue = queue
  @priority = []
  _sincedb_open
  determine_start_position(find_log_groups, @sincedb)

  while !stop?
    begin
      groups = find_log_groups

      groups.each do |group|
        @logger.debug("calling process_group on #{group}")
        process_group(group)
      end # groups.each
    rescue Aws::CloudWatchLogs::Errors::ThrottlingException
      @logger.debug("reached rate limit")
    end

    Stud.stoppable_sleep(@interval) { stop? }
  end
end