Measure emitting records counts and their sizes for input and output metrics and used it in in_monitor_agent (revised)

[cosmo0920]

Which issue(s) this PR fixes:
None

What this PR does / why we need it:
Currently, Fluentd metrics do not have emit size for calculating flow rate (per sec or per hour).
Fluentd Input plugin and Output should have emit_records and emit_size metrics.

But, currently, I disabled input metrics and size metrics by default for performance concerns.

enable_input_metrics in system config is false by default.
enable_size_metrics in system config is also false by default.

Docs Changes:
Needed.

Release Note:

Same as title.

[cosmo0920]

Sorry, I accidentally deleted PRed branch on #3411.... 😢

[ashie]

Oh... 😅

[cosmo0920]

And I measured simple benchmark testing for filter case (not included in this PR):

#!/usr/bin/env ruby

require 'benchmark/ips'

require_relative 'test/helper'
require 'fluent/plugin/filter'
require 'fluent/event'
require 'fluent/engine'
require 'fluent/event_router'

class DummyStreamFilter < Fluent::Plugin::Filter
  def filter_stream(tag, es)
    es
  end
end

class DummyFilter < Fluent::Plugin::Filter
  def filter(tag, time, record)
    record
  end
end

def create_filter_stream(router)
  f = DummyStreamFilter.new
  f.context_router = router
  f.configure(config_element())
  f
end

def create_filter(router)
  f = DummyFilter.new
  f.context_router = router
  f.configure(config_element())
  f
end

router = Fluent::EventRouter.new(Fluent::NoMatchMatch.new($log), self)

fs1 = create_filter_stream(router)
fs1.use_size_metric = false

fs2 = create_filter_stream(router)

f1 = create_filter(router)
f1.use_size_metric = false

f2 = create_filter(router)

t = event_time()
es = Fluent::ArrayEventStream.new([ [t, {"key" => "value1"}], [t, {"key" => "value2"}] ])

Benchmark.ips do |x|
  x.report("before -- filter stream") do
    fs1.filter_stream('tag', es)
  end

  x.report("after -- filter stream") do
    fs2.filter_stream('tag', es)
  end

  x.report("before -- filter") do
    f1.filter('tag', t, {"key" => "value1"})
  end

  x.report("after -- filter") do
    f2.filter('tag', t, {"key" => "value1"})
  end
end

$ bundle exec ruby bench_filter_metrics.rb
Warming up --------------------------------------
before -- filter stream
                       377.028k i/100ms
after -- filter stream
                       373.687k i/100ms
    before -- filter   217.949k i/100ms
     after -- filter   209.939k i/100ms
Calculating -------------------------------------
before -- filter stream
                          3.629M (± 6.5%) i/s -     18.097M in   5.010721s
after -- filter stream
                          3.614M (± 6.6%) i/s -     18.311M in   5.091728s
    before -- filter      2.427M (± 7.7%) i/s -     12.205M in   5.058706s
     after -- filter      2.434M (± 8.3%) i/s -     12.176M in   5.043038s

Output performance regression is not observed in filter, but for the base classes' consistency, I'll add enable_size_metrics system config handling for filter base class and bare_output and multi_output base classes.

[cosmo0920]

I resolved conflicts. I'll merge this PR after CIs are finished.