in_tail: add "tracked_file_count" metrics to see how many log files are being tracked (#4980)

**Which issue(s) this PR fixes**: 
Fixes #

**What this PR does / why we need it**: 
This PR will add a `tracked_file_count` field to the value returned by
statistics method.
This will make it easy to see how many log files are being tracked.

**Docs Changes**:
It appears that there is no documentation at all on plugin-specific
metrics.
Maybe we should add document about in_tail's metrics to
https://docs.fluentd.org/input/tail.

**Release Note**: 
Same as the title.

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/plugin/in_tail.rb

@@ -36,7 +36,7 @@ class TailInput < Fluent::Plugin::Input
     helpers :timer, :event_loop, :parser, :compat_parameters
 
     RESERVED_CHARS = ['/', '*', '%'].freeze
-    MetricsInfo = Struct.new(:opened, :closed, :rotated, :throttled)
+    MetricsInfo = Struct.new(:opened, :closed, :rotated, :throttled, :tracked)
 
     class WatcherSetupError < StandardError
       def initialize(msg)
@@ -206,11 +206,15 @@ def configure(conf)
           @read_bytes_limit_per_second = min_bytes
         end
       end
+
       opened_file_metrics = metrics_create(namespace: "fluentd", subsystem: "input", name: "files_opened_total", help_text: "Total number of opened files")
       closed_file_metrics = metrics_create(namespace: "fluentd", subsystem: "input", name: "files_closed_total", help_text: "Total number of closed files")
       rotated_file_metrics = metrics_create(namespace: "fluentd", subsystem: "input", name: "files_rotated_total", help_text: "Total number of rotated files")
       throttling_metrics = metrics_create(namespace: "fluentd", subsystem: "input", name: "files_throttled_total", help_text: "Total number of times throttling occurs per file when throttling enabled")
-      @metrics = MetricsInfo.new(opened_file_metrics, closed_file_metrics, rotated_file_metrics, throttling_metrics)
+      # The metrics for currently tracking files. Since the value may decrease, it cannot be represented using the counter type, so 'prefer_gauge: true' is used instead.
+      tracked_file_metrics = metrics_create(namespace: "fluentd", subsystem: "input", name: "files_tracked_count", help_text: "Number of tracked files", prefer_gauge: true)
+
+      @metrics = MetricsInfo.new(opened_file_metrics, closed_file_metrics, rotated_file_metrics, throttling_metrics, tracked_file_metrics)
     end
 
     def check_dir_permission
@@ -474,6 +478,7 @@ def refresh_watchers
 
       stop_watchers(removed_hash, unwatched: need_unwatch_in_stop_watchers) unless removed_hash.empty?
       start_watchers(added_hash) unless added_hash.empty?
+      @metrics.tracked.set(@tails.size)
       @startup = false if @startup
     end
 
@@ -814,6 +819,7 @@ def statistics
           'closed_file_count' => @metrics.closed.get,
           'rotated_file_count' => @metrics.rotated.get,
           'throttled_log_count' => @metrics.throttled.get,
+          'tracked_file_count' => @metrics.tracked.get,
         })
       }
       stats

test/plugin/test_in_tail.rb

@@ -3443,4 +3443,32 @@ def test_next_rotation_occurs_very_fast_while_old_TW_still_waiting_rotate_wait
                     d.logs[-2..]
                   ])
   end
+
+  test 'statistics' do
+    config = config_element("", "", {
+      "tag" => "statistics",
+      "path" => "#{@tmp_dir}/statistics*.txt",
+      "format" => "none",
+      "read_from_head" => true,
+    })
+    Fluent::FileWrapper.open("#{@tmp_dir}/statistics1.txt", "w+") do |f|
+      f.puts "foo"
+    end
+
+    d = create_driver(config, false)
+    d.run(expect_records: 1, shutdown: false)
+
+    assert_equal({
+                   "emit_records" => 0,
+                   "emit_size" => 0,
+                   "opened_file_count" => 1,
+                   "closed_file_count" => 0,
+                   "rotated_file_count" => 0,
+                   "throttled_log_count" =>0,
+                   "tracked_file_count" => 1,
+                 },
+                 d.instance.statistics["input"])
+
+    d.instance_shutdown
+  end
 end