scripts: Example Lua function to rate limit by size

conf/rate_limit_by_size.conf

@@ -0,0 +1,26 @@
+[SERVICE]
+    Flush        1
+    Daemon       Off
+    Log_Level    info
+    Parsers_File parsers.conf
+
+[INPUT]
+    Name          tail
+    Path          /var/log/containers/*.log
+    Parser        docker
+    Tag           kube.*
+    Mem_Buf_Limit 5MB
+
+[FILTER]
+    Name   kubernetes
+    Match  kube.*
+
+[FILTER]
+    Name   lua
+    Match  kube.*
+    script rate_limit.lua
+    call   rate_limit_by_size
+
+[OUTPUT]
+    Name  stdout
+    Match *

scripts/rate_limit.lua

@@ -1,26 +1,32 @@
 --[[
    This Lua script is to do the rate limiting of logs based on some key. The Throttle filter in fluent-bit doesn't allow to do the rate limiting based on key
-
-   sample configuration:
-    [FILTER]
-     Name lua
-     Match kube.*
-     script rate_limit.lua
-     call rate_limit
 ]]
 
 local counter = {}
 local time = 0
 local group_key = "docker_id" -- Used to group logs. Groups are rate limited independently.
+local rate_limit_field = "log" -- The field in the record whose size is used to determine the rate limit
 local group_bucket_period_s = 60 -- This is the period of of time in seconds over which group_bucket_limit applies.
 local group_bucket_limit = 6000 -- Maximum number logs allowed per groups over the period of group_bucket_period_s.
-
--- with above values, each and every containers running on the kubernetes will have a limit of 6000 logs for every 60 seconds since contianers have unique kubernetes.docker_id value
+local group_bucket_limit_bytes = 30000 -- Maximum size of rate_limit_field in bytes allowed per kubernetes.group_key over the period of group_bucket_period_s.
 
 local function get_current_time(timestamp)
     return math.floor(timestamp / group_bucket_period_s)
 end
 
+--[[
+    This function is used to rate limit logs based on the number of logs of kubernetes.group_key.
+    If the number of logs in a group exceeds group_bucket_limit, the log is dropped.
+    E.g. With above values for the local variables, each and every containers running on Kubernetes will 
+    have a limit of 6000 logs for every 60 seconds since contianers have unique kubernetes.docker_id value
+
+    sample configuration:
+    [FILTER]
+        Name lua
+        Match kube.*
+        script rate_limit.lua
+        call rate_limit
+]]
 function rate_limit(tag, timestamp, record)
     local t = os.time()
     local current_time = get_current_time(t)
@@ -39,4 +45,40 @@ function rate_limit(tag, timestamp, record)
         end
     end
     return 0, 0, 0 -- keep the log
-end
+end
+
+--[[
+    This function is used to rate limit logs based on the size of the content of kubernetes.group_key.
+    E.g. With above values for the local variables, each and every container running on Kubernetes will
+    have a limit of 30000 bytes for every 60 seconds.
+
+    sample configuration:
+    [FILTER]
+        Name lua
+        Match kube.*
+        script rate_limit.lua
+        call rate_limit_by_size
+]]
+function rate_limit_by_size(tag, timestamp, record)
+    local t = os.time()
+    local current_time = get_current_time(t)
+    if current_time ~= time then
+        time = current_time
+        counter = {} -- reset the counter
+    end
+
+    if counter[group_key] == -1 then
+        return -1, 0, 0 -- Log group already rate limited. Hence drop it.
+    else
+        if counter[group_key] == nil then
+            counter[group_key] = #record[rate_limit_field]
+        else
+            counter[group_key] = counter[group_key] + #record[rate_limit_field]
+        end
+        if counter[group_key] > group_bucket_limit_bytes then
+            counter[group_key] = -1 -- value of -1 indicates that this group has been rate limited
+            return -1, 0, 0 -- drop the log     
+        end
+        return 0, 0, 0
+    end
+end