count shared <initlabels> against every section's limit

README.md

@@ -305,7 +305,9 @@ stays unlimited with `0`:
 The label sets are counted per metric name, not per `<metric>` section: sections
 with the same `name`, in one plugin or in two, share one count. Each of them
 refuses a new label set once that shared count reaches its own limit, so a
-section which stays at `0` adds label sets without counting them.
+section which stays at `0` adds label sets without counting them. Its
+`<initlabels>` are counted anyway. They come from the configuration and their
+number is fixed. The metric holds them in every section.
 
 The count is per worker process as well, since a worker has its own registry and
 exposes the metrics it holds itself. With `workers N`, a metric can hold up to N
@@ -317,11 +319,59 @@ not a number, does not consume the limit. A pre-initialized label set
 (`initialized` and `<initlabels>`) consumes it from the start, since the metric
 holds it before any record arrives.
 
-A `<metric>` section is refused at startup with a configuration error when its
-limit is smaller than the number of its own `<initlabels>` label sets: the limit
-is already exceeded before any record arrives, so the metric could never take a
-new label set. A limit equal to that number is fine, since every label set of
-the metric is known in advance.
+A `<metric>` section is refused at startup with a configuration error when the
+`<initlabels>` label sets already fill its limit. The limit is exceeded before
+any record arrives, and the section can never take a new label set.
+
+These label sets are shared by every section with the same `name`. A section can
+be refused because of another section:
+
+```
+<metric>
+  name shared
+  type counter
+  desc Something foo.
+  max_series_per_metric 2
+  initialized true
+  <labels>
+    path $.path
+  </labels>
+  <initlabels>
+    path /a
+  </initlabels>
+  <initlabels>
+    path /b
+  </initlabels>
+</metric>
+<metric>
+  name shared               # the same name, so the 2 label sets count here too
+  type counter
+  desc Something foo.
+  max_series_per_metric 1   # refused: 2 label sets do not fit into 1
+  <labels>
+    path $.path
+  </labels>
+</metric>
+```
+
+The second section is refused even though it declares no `<initlabels>` of its
+own. The same happens when the first section sets `max_series_per_metric 0`.
+The metric holds its label sets in both cases. A limit equal to their number is
+fine because all label sets of the metric are known in advance.
+
+The check runs after every `<metric>` section of a plugin is read. It does not
+depend on the order of the sections. Sections that share a `name` across two
+plugins are only checked against the sections read before them.
+
+The check runs again when Fluentd reloads the configuration. It counts only the
+label sets from `<initlabels>`. It does not count the label sets that records
+brought, so the metric does not make the reload fail with the label sets it took
+while it was running.
+
+A reload does not clear the registry. The metric still holds the label sets from
+the `<initlabels>` of the old configuration, and the check counts them too. So
+it can refuse a new configuration which has fewer `<initlabels>` than the old
+one. Restart the worker to drop them.
 
 ##### Observing what the limit leaves out
 

lib/fluent/plugin/prometheus.rb

@@ -60,10 +60,12 @@ def self.included(klass)
       # holding max_series_per_metric label sets of its own.
       class SeriesSet
         # The set is kept on the client metric, so that it is found again by
-        # every section and goes away with it. Metrics are built at
-        # configuration time, which is single threaded, so no lock is needed.
+        # every section. The registry does not drop its metrics, so the set is
+        # still there after a reload.
         IVAR = :@fluent_plugin_prometheus_series_set
 
+        # Sections are built at configuration time, which is single threaded,
+        # so this needs no lock.
         def self.of(client_metric)
           client_metric.instance_variable_get(IVAR) ||
             client_metric.instance_variable_set(IVAR, new)
@@ -74,8 +76,10 @@ def initialize
           @mutex = Mutex.new
         end
 
-        def size
-          @mutex.synchronize { @series.size }
+        # Only <initlabels> come from the configuration. Counting a label set
+        # a record brought would refuse a good configuration after a reload.
+        def initial_size
+          @mutex.synchronize { @series.count { |_, state| state == :initial } }
         end
 
         # Checking the limit and taking the slot happen under the same lock, so
@@ -100,7 +104,19 @@ def reserve(label, limit, name)
         # side already, so it has to be accounted for even when a concurrent
         # failure gave the reservation back in the meantime.
         def confirm(label)
-          @mutex.synchronize { @series[label] = :confirmed }
+          @mutex.synchronize do
+            # #initial_size has to keep counting it, so a record on it does not
+            # change where it came from
+            next if @series[label] == :initial
+
+            @series[label] = :confirmed
+          end
+        end
+
+        # The limit is not checked here either: a section which cannot hold
+        # these label sets is refused when the configuration is read.
+        def confirm_initial(label)
+          @mutex.synchronize { @series[label] = :initial }
         end
 
         # Gives a reserved slot back when the instrumentation failed, so that a
@@ -221,6 +237,14 @@ def self.parse_metrics_elements(conf, registry, labels = {}, opts = {})
             raise ConfigError, "type option must be 'counter', 'gauge', 'summary' or 'histogram'"
           end
         }
+
+        # <metric> sections with the same name share one client
+        # metric. All of their <initlabels> label sets are known only
+        # after every section is built, so the check runs here and not
+        # in the constructor not to depend on the order of the
+        # sections.
+        metrics.each(&:check_series_limit!)
+
         metrics
       end
 
@@ -427,6 +451,22 @@ def self.get(registry, name, type, docstring)
           metric
         end
 
+        # <initlabels> label sets go to the client as soon as the
+        # configuration is read. Every <metric> section with the same name
+        # shares them, so reject a section which has no room for the label
+        # sets.
+        def check_series_limit!
+          return if @max_series_per_metric <= 0
+          # two <initlabels> blocks with the same values make one label set
+          held = @series_set.initial_size
+          return if held <= @max_series_per_metric
+
+          raise ConfigError, "metric #{@name} already holds #{held} label sets from <initlabels>, " \
+                             "shared by every <metric> section with this name, " \
+                             "but max_series_per_metric is #{@max_series_per_metric} in this section: " \
+                             "the limit is already exceeded before any record arrives"
+        end
+
         private
 
         # The client refuses a value which is not a number, but it is only
@@ -455,28 +495,16 @@ def metric_limit(element, name, default)
         def bind_series_set(client_metric)
           @series_set = SeriesSet.of(client_metric)
 
-          if @initialized
-            # the client is given them at startup, so they take their slots now
-            @base_initlabels.each do |initlabels|
-              confirm_series(normalize_label_set(initlabels))
-            end
-            check_initlabels_fit_series_limit!
-          end
-        end
+          return unless @initialized
 
-        # The client is given these label sets at startup, so a limit which
-        # does not fit them is already exceeded before any record arrives and
-        # the metric could never take a new one. Stop instead of running that
-        # way. A record on one of them is still counted, since the metric
-        # already holds its label set.
-        def check_initlabels_fit_series_limit!
-          return if @max_series_per_metric <= 0
-          # two <initlabels> blocks with the same values make one label set
-          return if @series_set.size <= @max_series_per_metric
-
-          raise ConfigError, "metric #{@name} holds #{@series_set.size} label sets from <initlabels>, " \
-                             "but max_series_per_metric is #{@max_series_per_metric}: " \
-                             "the limit is already exceeded before any record arrives"
+          # The client gets them even when this section has no limit, so they
+          # take their slots in both cases. A section with the same name shares
+          # this set and has to see them. Their number is fixed by the
+          # configuration. Counting them cannot leak like the label sets that
+          # records bring.
+          @base_initlabels.each do |initlabels|
+            @series_set.confirm_initial(normalize_label_set(initlabels))
+          end
         end
 
         # The SeriesSet keys a label set by its values, so the same value has to
@@ -494,8 +522,9 @@ def normalize_label_set(label)
         # Returns true when this call took the slot, which tells #with_label_set
         # whether it has something to give back on failure.
         def reserve_series!(label)
-          # nothing is counted with the limit off, otherwise the set would grow
-          # with every label set and leak what the limit is there to prevent
+          # If the limit is off, a label set from a record is not counted.
+          # If it is counted, the set grows with every new label set and
+          # uses too much memory.
           return false if @max_series_per_metric <= 0
 
           @series_set.reserve(label, @max_series_per_metric, @name)

spec/fluent/plugin/prometheus/series_limit_spec.rb

@@ -33,6 +33,31 @@ def instrument(path, value = 1)
     metric.instrument({'foo' => value, 'path' => path}, expander)
   end
 
+  def build_metrics(*elements)
+    Fluent::Plugin::Prometheus.parse_metrics_elements(
+      Fluent::Config::Element.new('ROOT', '', {}, elements), registry, {}, {}
+    )
+  end
+
+  # A <metric> section named 'shared', so that two of them instrument the same
+  # client metric.
+  def counter_element(limit, initlabels = nil)
+    attributes = {
+      'name' => 'shared',
+      'type' => 'counter',
+      'desc' => 'Something foo.',
+      'key' => 'foo',
+      'max_series_per_metric' => limit.to_s,
+    }
+    attributes['initialized'] = 'true' if initlabels
+
+    Fluent::Config::Element.new(
+      'metric', '', attributes,
+      [Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, [])] +
+        Array(initlabels).map { |path| Fluent::Config::Element.new('initlabels', '', {'path' => path}, []) }
+    )
+  end
+
   describe 'max_series_per_metric' do
     it 'refuses a new label set once the limit is reached' do
       instrument('/a')
@@ -275,8 +300,9 @@ def instrument_other(path, value = 1)
       let(:max_series_per_metric) { 1 }
 
       it 'stops at startup instead of dropping every record' do
-        expect { metric }.to raise_error(Fluent::ConfigError,
-                                         /holds 3 label sets from <initlabels>.*max_series_per_metric is 1/)
+        expect { build_metrics(element) }
+          .to raise_error(Fluent::ConfigError,
+                          /already holds 3 label sets from <initlabels>.*max_series_per_metric is 1/)
       end
     end
 
@@ -286,7 +312,7 @@ def instrument_other(path, value = 1)
       let(:max_series_per_metric) { 3 }
 
       it 'accepts the config' do
-        expect { metric }.not_to raise_error
+        expect { build_metrics(element) }.not_to raise_error
       end
 
       it 'still counts a record on an <initlabels> label set' do
@@ -306,20 +332,95 @@ def instrument_other(path, value = 1)
       let(:max_series_per_metric) { 1 }
 
       it 'counts the label sets and not the <initlabels> blocks' do
-        expect { metric }.not_to raise_error
+        expect { build_metrics(element) }.not_to raise_error
       end
     end
 
     context 'without a limit' do
       let(:max_series_per_metric) { 0 }
 
       it 'accepts any number of <initlabels> label sets' do
-        expect { metric }.not_to raise_error
+        expect { build_metrics(element) }.not_to raise_error
         expect { instrument('/d') }.not_to raise_error
       end
     end
   end
 
+  describe '<initlabels> shared by <metric> sections with the same name' do
+    # Every section with this name instruments the same client metric. The
+    # label sets from <initlabels> count in every section, even in one that
+    # declares none. A section with no limit still adds its own to the count.
+    let(:five_initlabels) { ['/a', '/b', '/c', '/d', '/e'] }
+
+    it 'refuses a section that has no <initlabels>' do
+      expect { build_metrics(counter_element(10, five_initlabels), counter_element(3)) }
+        .to raise_error(Fluent::ConfigError,
+                        /already holds 5 label sets from <initlabels>.*max_series_per_metric is 3/)
+    end
+
+    it 'counts the <initlabels> of a section that has no limit' do
+      expect { build_metrics(counter_element(0, five_initlabels), counter_element(3, ['/z'])) }
+        .to raise_error(Fluent::ConfigError,
+                        /already holds 6 label sets from <initlabels>.*max_series_per_metric is 3/)
+    end
+
+    it 'does not depend on the order of the sections' do
+      expect { build_metrics(counter_element(3), counter_element(10, five_initlabels)) }
+        .to raise_error(Fluent::ConfigError, /max_series_per_metric is 3/)
+    end
+
+    it 'accepts a config when every limit fits the shared label sets' do
+      expect { build_metrics(counter_element(5, five_initlabels), counter_element(10)) }
+        .not_to raise_error
+    end
+
+    it 'counts them against the limit at runtime as well' do
+      # the section with no limit puts them on the client, so the section with
+      # a limit has to see them: none of its six slots is left
+      _unlimited, limited = build_metrics(counter_element(0, five_initlabels),
+                                          counter_element(6, ['/z']))
+
+      expect { limited.instrument({'foo' => 1, 'path' => '/new'}, expander) }
+        .to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError)
+    end
+  end
+
+  describe 'reading the configuration again' do
+    # A reload builds the <metric> sections again against the registry the
+    # process already has, so the client metric keeps its label sets. Reading
+    # the configuration twice here does the same.
+    def instrument_through(metric, path)
+      metric.instrument({'foo' => 1, 'path' => path}, expander)
+    end
+
+    it 'does not count the label sets that records brought' do
+      # the section with the wider limit fills five slots of the set both
+      # sections share, which leaves none for the limit of three
+      wide, _narrow = build_metrics(counter_element(10), counter_element(3))
+      ['/a', '/b', '/c', '/d', '/e'].each { |path| instrument_through(wide, path) }
+
+      expect { build_metrics(counter_element(10), counter_element(3)) }.not_to raise_error
+    end
+
+    it 'accepts the same <initlabels> again' do
+      metric, = build_metrics(counter_element(3, ['/a', '/b', '/c']))
+      instrument_through(metric, '/a')
+
+      expect { build_metrics(counter_element(3, ['/a', '/b', '/c'])) }.not_to raise_error
+    end
+
+    it 'keeps counting an <initlabels> label set a record came on' do
+      # the client still holds the three label sets, so a limit of 2 does not
+      # fit them even though a record made one of them look like its own
+      metric, = build_metrics(counter_element(3, ['/a', '/b', '/c']))
+      instrument_through(metric, '/a')
+
+      expect { build_metrics(counter_element(2, ['/a', '/b', '/c'])) }
+        .to raise_error(Fluent::ConfigError,
+                        /already holds 3 label sets from <initlabels>.*max_series_per_metric is 2/)
+    end
+  end
+
   describe '<metric> overriding the plugin limit' do
     # the plugin is configured with 100, which <metric> has to win over
     let(:metric) do