Telegraf agent config for DS/RS (#847)

* add telegraf config options to configmap

Co-authored-by: Amol Agrawal <[email protected]>

Author: pfrcks

Documentation/AgentSettings/ReadMe.md

@@ -18,9 +18,26 @@ Note that this can only be achieved using helm chart today.
 ```
   agent-settings: |-
     # prometheus scrape fluent bit settings for high scale
-    # buffer size should be greater than or equal to chunk size else we set it to chunk size. 
+    # buffer size should be greater than or equal to chunk size else we set it to chunk size.
+    # settings scoped to prometheus sidecar container. all values in mb
     [agent_settings.prometheus_fbit_settings]
       tcp_listener_chunk_size = 10
       tcp_listener_buffer_size = 10
       tcp_listener_mem_buf_limit = 200
+
+    # prometheus scrape fluent bit settings for high scale
+    # buffer size should be greater than or equal to chunk size else we set it to chunk size.
+    # settings scoped to daemonset container. all values in mb
+    # [agent_settings.node_prometheus_fbit_settings]
+      # tcp_listener_chunk_size = 1
+      # tcp_listener_buffer_size = 1
+      # tcp_listener_mem_buf_limit = 10
+
+    # prometheus scrape fluent bit settings for high scale
+    # buffer size should be greater than or equal to chunk size else we set it to chunk size.
+    # settings scoped to replicaset container. all values in mb
+    # [agent_settings.cluster_prometheus_fbit_settings]
+      # tcp_listener_chunk_size = 1
+      # tcp_listener_buffer_size = 1
+      # tcp_listener_mem_buf_limit = 10
 ```

build/common/installer/scripts/tomlparser-agent-config.rb

@@ -57,6 +57,21 @@
 @fbitTailMemBufLimitMBs = 0
 @fbitTailIgnoreOlder = ""
 
+# Checking to see if this is the daemonset or replicaset to parse config accordingly
+@controllerType = ENV["CONTROLLER_TYPE"]
+@daemonset = "daemonset"
+# Checking to see if container is not prometheus sidecar.
+# CONTAINER_TYPE is populated only for prometheus sidecar container.
+@containerType = ENV["CONTAINER_TYPE"]
+
+@promFbitChunkSize = 0
+@promFbitBufferSize = 0
+@promFbitMemBufLimit = 0
+
+@promFbitChunkSizeDefault = "32k" #kb
+@promFbitBufferSizeDefault  = "64k" #kb
+@promFbitMemBufLimitDefault  = "10m" #mb
+
 def is_number?(value)
   true if Integer(value) rescue false
 end
@@ -178,6 +193,35 @@ def populateSettingValuesFromConfigMap(parsedConfig)
           end
         end
       end
+    
+      prom_fbit_config = nil
+      if !@controllerType.nil? && !@controllerType.empty? && @controllerType.strip.casecmp(@daemonset) == 0 && @containerType.nil?
+        prom_fbit_config = parsedConfig[:agent_settings][:node_prometheus_fbit_settings]
+      elsif !@controllerType.nil? && !@controllerType.empty? && @controllerType.strip.casecmp(@daemonset) != 0
+        prom_fbit_config = parsedConfig[:agent_settings][:cluster_prometheus_fbit_settings]
+      end
+
+      if !prom_fbit_config.nil?
+        chunk_size = prom_fbit_config[:tcp_listener_chunk_size]
+        if !chunk_size.nil? && is_number?(chunk_size) && chunk_size.to_i > 0
+          @promFbitChunkSize = chunk_size.to_i
+          puts "Using config map value: AZMON_FBIT_CHUNK_SIZE = #{@promFbitChunkSize.to_s + "m"}"
+        end
+        buffer_size = prom_fbit_config[:tcp_listener_buffer_size]
+        if !buffer_size.nil? && is_number?(buffer_size) && buffer_size.to_i > 0
+          @promFbitBufferSize = buffer_size.to_i
+          puts "Using config map value: AZMON_FBIT_BUFFER_SIZE = #{@promFbitBufferSize.to_s + "m"}"
+          if @promFbitBufferSize < @promFbitChunkSize
+            @promFbitBufferSize = @promFbitChunkSize
+            puts "Setting Fbit buffer size equal to chunk size since it is set to less than chunk size - AZMON_FBIT_BUFFER_SIZE = #{@promFbitBufferSize.to_s + "m"}"
+          end
+        end
+        mem_buf_limit = prom_fbit_config[:tcp_listener_mem_buf_limit]
+        if !mem_buf_limit.nil? && is_number?(mem_buf_limit) && mem_buf_limit.to_i > 0
+          @promFbitMemBufLimit = mem_buf_limit.to_i
+          puts "Using config map value: AZMON_FBIT_MEM_BUF_LIMIT = #{@promFbitMemBufLimit.to_s + "m"}"
+        end
+      end
     end
   rescue => errorStr
     puts "config::error:Exception while reading config settings for agent configuration setting - #{errorStr}, using defaults"
@@ -226,6 +270,24 @@ def populateSettingValuesFromConfigMap(parsedConfig)
     file.write("export FBIT_TAIL_IGNORE_OLDER=#{@fbitTailIgnoreOlder}\n")
   end
 
+  if @promFbitChunkSize > 0
+    file.write("export AZMON_FBIT_CHUNK_SIZE=#{@promFbitChunkSize.to_s + "m"}\n")
+  else
+    file.write("export AZMON_FBIT_CHUNK_SIZE=#{@promFbitChunkSizeDefault}\n")
+  end
+
+  if @promFbitBufferSize > 0
+    file.write("export AZMON_FBIT_BUFFER_SIZE=#{@promFbitBufferSize.to_s + "m"}\n")
+  else
+    file.write("export AZMON_FBIT_BUFFER_SIZE=#{@promFbitBufferSizeDefault}\n")
+  end
+
+  if @promFbitMemBufLimit > 0
+    file.write("export AZMON_FBIT_MEM_BUF_LIMIT=#{@promFbitMemBufLimit.to_s + "m"}\n")
+  else
+    file.write("export AZMON_FBIT_MEM_BUF_LIMIT=#{@promFbitMemBufLimitDefault}\n")
+  end
+
   # Close file after writing all environment variables
   file.close
 else
@@ -262,6 +324,27 @@ def get_command_windows(env_variable_name, env_variable_value)
       commands = get_command_windows("FBIT_TAIL_IGNORE_OLDER", @fbitTailIgnoreOlder)
       file.write(commands)
     end
+    if @promFbitChunkSize > 0
+      commands = get_command_windows("AZMON_FBIT_CHUNK_SIZE", @promFbitChunkSize.to_s + "m")
+      file.write(commands)
+    else
+      commands = get_command_windows("AZMON_FBIT_CHUNK_SIZE", @promFbitChunkSizeDefault)
+      file.write(commands)
+    end
+    if @promFbitBufferSize > 0
+      commands = get_command_windows("AZMON_FBIT_BUFFER_SIZE", @promFbitBufferSize.to_s + "m")
+      file.write(commands)
+    else
+      commands = get_command_windows("AZMON_FBIT_BUFFER_SIZE", @promFbitBufferSizeDefault)
+      file.write(commands)
+    end
+    if @promFbitMemBufLimit > 0
+      commands = get_command_windows("AZMON_FBIT_MEM_BUF_LIMIT", @promFbitMemBufLimit.to_s + "m")
+      file.write(commands)
+    else
+      commands = get_command_windows("AZMON_FBIT_MEM_BUF_LIMIT", @promFbitMemBufLimitDefault)
+      file.write(commands)
+    end
     # Close file after writing all environment variables
     file.close
     puts "****************End Config Processing********************"

build/linux/installer/conf/td-agent-bit-rs.conf

@@ -41,9 +41,9 @@
     Tag         oms.container.perf.telegraf.*
     Listen      0.0.0.0
     Port        25226
-    Chunk_Size  32
-    Buffer_Size 64
-    Mem_Buf_Limit 10m
+    Chunk_Size  ${AZMON_FBIT_CHUNK_SIZE}
+    Buffer_Size ${AZMON_FBIT_BUFFER_SIZE}
+    Mem_Buf_Limit ${AZMON_FBIT_MEM_BUF_LIMIT}
 
 [OUTPUT]
     Name                            oms

build/linux/installer/conf/td-agent-bit.conf

@@ -73,9 +73,9 @@
     Tag         oms.container.perf.telegraf.*
     Listen      0.0.0.0
     Port        25226
-    Chunk_Size  32
-    Buffer_Size 64
-    Mem_Buf_Limit 5m
+    Chunk_Size  ${AZMON_FBIT_CHUNK_SIZE}
+    Buffer_Size ${AZMON_FBIT_BUFFER_SIZE}
+    Mem_Buf_Limit ${AZMON_FBIT_MEM_BUF_LIMIT}
 
 [FILTER]
     Name grep

build/windows/installer/conf/fluent-bit.conf

@@ -46,9 +46,9 @@
     Tag                 oms.container.perf.telegraf.*
     Listen              0.0.0.0
     Port                25229
-    Chunk_Size          32
-    Buffer_Size         64
-    Mem_Buf_Limit       5m
+    Chunk_Size          ${AZMON_FBIT_CHUNK_SIZE}
+    Buffer_Size         ${AZMON_FBIT_BUFFER_SIZE}
+    Mem_Buf_Limit       ${AZMON_FBIT_MEM_BUF_LIMIT}
 
 [FILTER]
     Name grep

kubernetes/container-azm-ms-agentconfig.yaml

@@ -148,11 +148,28 @@ data:
   agent-settings: |-
     # prometheus scrape fluent bit settings for high scale
     # buffer size should be greater than or equal to chunk size else we set it to chunk size.
+    # settings scoped to prometheus sidecar container. all values in mb
     [agent_settings.prometheus_fbit_settings]
       tcp_listener_chunk_size = 10
       tcp_listener_buffer_size = 10
       tcp_listener_mem_buf_limit = 200
 
+    # prometheus scrape fluent bit settings for high scale
+    # buffer size should be greater than or equal to chunk size else we set it to chunk size.
+    # settings scoped to daemonset container. all values in mb
+    # [agent_settings.node_prometheus_fbit_settings]
+      # tcp_listener_chunk_size = 1
+      # tcp_listener_buffer_size = 1
+      # tcp_listener_mem_buf_limit = 10
+
+    # prometheus scrape fluent bit settings for high scale
+    # buffer size should be greater than or equal to chunk size else we set it to chunk size.
+    # settings scoped to replicaset container. all values in mb
+    # [agent_settings.cluster_prometheus_fbit_settings]
+      # tcp_listener_chunk_size = 1
+      # tcp_listener_buffer_size = 1
+      # tcp_listener_mem_buf_limit = 10
+
     # The following settings are "undocumented", we don't recommend uncommenting them unless directed by Microsoft.
     # They increase the maximum stdout/stderr log collection rate but will also cause higher cpu/memory usage.
     ## Ref for more details about Ignore_Older -  https://docs.fluentbit.io/manual/v/1.7/pipeline/inputs/tail

source/plugins/go/src/telemetry.go

@@ -194,6 +194,10 @@ func SendContainerLogPluginMetrics(telemetryPushIntervalProperty string) {
 				if fbitTailMemBufLimitMBs != "" {
 					telemetryDimensions["FbitMemBufLimitSizeMBs"] = fbitTailMemBufLimitMBs
 				}
+				telemetryDimensions["PromFbitChunkSize"] = os.Getenv("AZMON_FBIT_CHUNK_SIZE")
+				telemetryDimensions["PromFbitBufferSize"] = os.Getenv("AZMON_FBIT_BUFFER_SIZE")
+				telemetryDimensions["PromFbitMemBufLimit"] = os.Getenv("AZMON_FBIT_MEM_BUF_LIMIT")
+
 				SendEvent(eventNameDaemonSetHeartbeat, telemetryDimensions)
 				flushRateMetric := appinsights.NewMetricTelemetry(metricNameAvgFlushRate, flushRate)
 				TelemetryClient.Track(flushRateMetric)