addition of new metric

fluentd/vendored_gem_src/fluent-plugin-prometheus/lib/fluent/plugin/in_prometheus_tail_monitor.rb

@@ -9,7 +9,7 @@ class PrometheusTailMonitorInput < Fluent::Plugin::Input
 
     helpers :timer
 
-    config_param :interval, :time, default: 5
+    config_param :interval, :time, default: 1
     attr_reader :registry
 
     MONITOR_IVARS = [
@@ -38,24 +38,31 @@ def configure(conf)
         @base_labels[key] = expander.expand(value)
       end
 
-      if defined?(Fluent::Plugin) && defined?(Fluent::Plugin::MonitorAgentInput)
-        # from v0.14.6
-        @monitor_agent = Fluent::Plugin::MonitorAgentInput.new
-      else
-        @monitor_agent = Fluent::MonitorAgentInput.new
-      end
+      @monitor_agent = Fluent::Plugin::MonitorAgentInput.new
     end
 
     def start
       super
 
       @metrics = {
-        position: @registry.gauge(
+        position: get_gauge(
           :fluentd_tail_file_position,
           'Current position of file.'),
-        inode: @registry.gauge(
+        inode: get_gauge(
           :fluentd_tail_file_inode,
           'Current inode of file.'),
+        maxfsize: get_gauge(
+          :maxfsize,
+          'Current max fsize of file on rotation event'),
+        countonrotate: get_gauge(
+          :countonrotate,
+          'No of rotation noticed by fluentd'),
+        totalbytesread: get_gauge(
+          :totalbytesread,
+          'totalbytes read by fluentd IOHandler'),
+        totalbytesavailable: get_gauge(
+          :totalbytesavailable,
+          'totalbytes available at each instance of rotation - to be read by fluentd IOHandler'),
       }
       timer_execute(:in_prometheus_tail_monitor, @interval, &method(:update_monitor_info))
     end
@@ -77,9 +84,18 @@ def update_monitor_info
           # Access to internal variable of internal class...
           # Very fragile implementation
           pe = watcher.instance_variable_get(:@pe)
+          totalbytesread = watcher.instance_variable_get(:@totalbytesread)
+          totalbytesavailable = watcher.instance_variable_get(:@totalbytesavailable)
+          maxfsize = watcher.instance_variable_get(:@maxfsize)
+          countonrotate = watcher.instance_variable_get(:@countonrotate)
           label = labels(info, watcher.path)
           @metrics[:inode].set(label, pe.read_inode)
           @metrics[:position].set(label, pe.read_pos)
+          @metrics[:maxfsize].set(label, maxfsize)
+          @metrics[:countonrotate].set(label, countonrotate)
+          @metrics[:totalbytesread].set(label, totalbytesread)
+          @metrics[:totalbytesavailable].set(label, totalbytesavailable)
+          #@log.info "IN PROMETHEUS PLUGIN pr.read_inode and pe.read_pos #{pe.read_inode} #{pe.read_pos} maxfsize #{maxfsize} countonrotate #{countonrotate} totalbytesread #{totalbytesread} totalbytesavailable #{totalbytesavailable} "
         end
       end
     end
@@ -91,5 +107,13 @@ def labels(plugin_info, path)
         path: path,
       )
     end
+
+    def get_gauge(name, docstring)
+      if @registry.exist?(name)
+        @registry.get(name)
+      else
+        @registry.gauge(name, docstring)
+      end
+    end
   end
 end

fluentd/vendored_gem_src/fluentd/lib/fluent/plugin/in_tail.rb

@@ -510,6 +510,13 @@ def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, e
         @from_encoding = from_encoding
         @encoding = encoding
         @open_on_every_update = open_on_every_update
+        @totalbytesread=0
+        @totalbytesavailable=0
+        @inodelastfsize_map={}
+        @inodereadfsize_map={}
+        @countonrotate=0
+        @maxfsize=0
+        @logloss=0
       end
 
       attr_reader :path
@@ -519,6 +526,11 @@ def initialize(path, rotate_wait, pe, log, read_from_head, enable_watch_timer, e
       attr_accessor :timer_trigger
       attr_accessor :line_buffer, :line_buffer_timer_flusher
       attr_accessor :unwatched  # This is used for removing position entry from PositionFile
+      attr_accessor :totalbytesavailable  # This is used for removing position entry from PositionFile
+      attr_accessor :totalbytesread  # This is used for removing position entry from PositionFile
+      attr_accessor :maxfsize  # This is used for removing position entry from PositionFile
+      attr_accessor :countonrotate  # This is used for removing position entry from PositionFile
+      attr_accessor :logloss  # This is used for removing position entry from PositionFile
 
       def tag
         @parsed_tag ||= @path.tr('/', '.').gsub(/\.+/, '.').gsub(/^\./, '')
@@ -535,7 +547,7 @@ def attach
 
       def detach
         yield self
-        @io_handler.on_notify if @io_handler
+        @io_handler.on_notify(@inodereadfsize_map) if @io_handler
       end
 
       def close
@@ -553,12 +565,28 @@ def on_notify
           stat = nil
         end
 
-        @rotate_handler.on_notify(stat) if @rotate_handler
+        @totalbytesavailable=@rotate_handler.on_notify(stat,@inodelastfsize_map) if @rotate_handler
         @line_buffer_timer_flusher.on_notify(self) if @line_buffer_timer_flusher
-        @io_handler.on_notify if @io_handler
+        @io_handler.on_notify(@inodereadfsize_map) if @io_handler
+
+        @logloss=0
+        if @io_handler && @rotate_handler
+          @totalbytesread=0
+          @countonrotate=0
+          @inodereadfsize_map.each do |k, v|
+            @totalbytesread+=v
+            @countonrotate+=1
+          end
+
+        #@log.info "inodelastfsize_map #{@inodelastfsize_map} inodereadfsize_map #{@inodereadfsize_map}"
+        @logloss=@totalbytesavailable-@totalbytesread
+        #@log.info "totalbytesavailable #{@totalbytesavailable} totalbytesread #{@totalbytesread} logloss #{@logloss} countonrotate #{@countonrotate}"
+        end  
+
       end
 
       def on_rotate(stat)
+        @maxfsize=@pe.read_pos
         if @io_handler.nil?
           if stat
             # first time
@@ -591,6 +619,7 @@ def on_rotate(stat)
             @io_handler = IOHandler.new(self, &method(:wrap_receive_lines))
           else
             @io_handler = NullIOHandler.new
+            @totalbytesread=0
           end
         else
           watcher_needs_update = false
@@ -605,7 +634,7 @@ def on_rotate(stat)
             else # file is rotated and new file found
               watcher_needs_update = true
               # Handle the old log file before renewing TailWatcher [fluentd#1055]
-              @io_handler.on_notify
+              @io_handler.on_notify(@inodereadfsize_map)
             end
           else # file is rotated and new file not found
             # Clear RotateHandler to avoid duplicated file watch in same path.
@@ -732,21 +761,28 @@ def initialize(watcher, &receive_lines)
           @io = nil
           @notify_mutex = Mutex.new
           @watcher.log.info "following tail of #{@watcher.path}"
+          @totalbytesread=0
+          @statatopen=nil
         end
 
-        def on_notify
-          @notify_mutex.synchronize { handle_notify }
+        def on_notify(inodereadfsize_map)
+          @notify_mutex.synchronize { handle_notify(inodereadfsize_map) }
         end
 
-        def handle_notify
+        def handle_notify(inodereadfsize_map)
           with_io do |io|
+          stat = Fluent::FileWrapper.stat(@watcher.path)
+          @statatopen=stat
             begin
               read_more = false
 
               if !io.nil? && @lines.empty?
+                ino=@statatopen.ino
                 begin
                   while true
-                    @fifo << io.readpartial(8192, @iobuf)
+                    buf = io.readpartial(8192, @iobuf)
+                    @totalbytesread+=buf.bytesize
+                    @fifo << buf
                     @fifo.read_lines(@lines)
                     if @lines.size >= @watcher.read_lines_limit
                       # not to use too much memory in case the file is very large
@@ -756,6 +792,7 @@ def handle_notify
                   end
                 rescue EOFError
                 end
+                inodereadfsize_map[ino]=@totalbytesread
               end
 
               unless @lines.empty?
@@ -784,7 +821,7 @@ def opened?
         def open
           io = Fluent::FileWrapper.open(@watcher.path)
           io.seek(@watcher.pe.read_pos + @fifo.bytesize)
-          io
+          return io
         rescue RangeError
           io.close if io
           raise WatcherSetupError, "seek error with #{@watcher.path}: file position = #{@watcher.pe.read_pos.to_s(16)}, reading bytesize = #{@fifo.bytesize.to_s(16)}"
@@ -840,9 +877,10 @@ def initialize(watcher, &on_rotate)
           @inode = nil
           @fsize = -1  # first
           @on_rotate = on_rotate
+          @totalbytesavailable=0
         end
 
-        def on_notify(stat)
+        def on_notify(stat,inodelastfsize_map)
           if stat.nil?
             inode = nil
             fsize = 0
@@ -854,9 +892,18 @@ def on_notify(stat)
           begin
             if @inode != inode || fsize < @fsize
               @on_rotate.call(stat)
+              inodelastfsize_map[@inode]=@fsize
+              @totalbytesavailable=@fsize
             end
             @inode = inode
             @fsize = fsize
+            inodelastfsize_map[inode]=fsize
+            @totalbytesavailable=0
+            inodelastfsize_map.each do | k , v |
+              @totalbytesavailable+=v
+            end
+
+            return @totalbytesavailable
           end
 
         rescue