filter_ratelimit: Add message rate limiter

Add a filter to limit messages per configurable source field. The
benefit of this is the overall logging infrastructure can be protected
from a rogue logging source (e.g. an application that spams a high
volume of messages in a short period of time).

Rate limiting is implemented using a typical token based algorithm. Some
n tokens per second are added, with a total burst limit of q. This
limits messages to n messages per second on average, with a max amount
of q per second.

Configuration:

Bucket_Key <filename-field>

Field to use for grouping messages into a rate limited bucket. Rate limits
apply to the bucket. Buckets are independent from each other.

Events_Per_Second 10

Average number of events per second allowed.

Events_Burst 20

Max number of events in a second allowed.

Initial_Burst 100

Max number of events to allow on startup. Useful when expected to load
a lot of log messages at startup from new log files.

Max_Buckets 256

Number of expected active buckets. If too small, rate limiting won't
function very well.

CMakeLists.txt

@@ -112,6 +112,7 @@ option(FLB_FILTER_STDOUT   "Enable stdout filter"               Yes)
 option(FLB_FILTER_PARSER   "Enable parser filter"               Yes)
 option(FLB_FILTER_KUBERNETES "Enable kubernetes filter"         Yes)
 option(FLB_FILTER_RECORD_MODIFIER "Enable record_modifier filter" Yes)
+option(FLB_FILTER_RATELIMIT "Enable ratelimit filter"           Yes)
 
 # Enable all features
 if(FLB_ALL)

include/fluent-bit/flb_hash.h

@@ -29,7 +29,7 @@
 struct flb_hash_entry {
     char *key;
     size_t key_len;
-    char *val;
+    void *val;
     size_t val_size;
     struct mk_list _head;
 };
@@ -48,11 +48,11 @@ struct flb_hash *flb_hash_create(size_t size);
 void flb_hash_destroy(struct flb_hash *ht);
 
 int flb_hash_add(struct flb_hash *ht, char *key, int key_len,
-                 char *val, size_t val_size);
+                 void *val, size_t val_size);
 int flb_hash_get(struct flb_hash *ht, char *key, int key_len,
-                 char **out_buf, size_t *out_size);
-int flb_hash_get_by_id(struct flb_hash *ht, int id, char *key, char **out_buf,
-                       size_t *out_size);
+                 void **out, size_t *out_size);
+int flb_hash_get_by_id(struct flb_hash *ht, int id, char *key, int key_len,
+                       void **out, size_t *out_size);
 int flb_hash_del(struct flb_hash *ht, char *key);
 
 #endif

plugins/CMakeLists.txt

@@ -147,6 +147,7 @@ if(FLB_REGEX)
   REGISTER_FILTER_PLUGIN("filter_parser")
 endif()
 REGISTER_FILTER_PLUGIN("filter_record_modifier")
+REGISTER_FILTER_PLUGIN("filter_ratelimit")
 
 # Register external input and output plugins
 if(EXT_IN_PLUGINS)

plugins/filter_kubernetes/kube_meta.c

@@ -692,6 +692,7 @@ int flb_kube_meta_get(struct flb_kube *ctx,
     int id;
     int ret;
     char *hash_meta_buf;
+    void *out;
     size_t hash_meta_size;
 
     if (ctx->dummy_meta == FLB_TRUE) {
@@ -708,7 +709,8 @@ int flb_kube_meta_get(struct flb_kube *ctx,
     /* Check if we have some data associated to the cache key */
     ret = flb_hash_get(ctx->hash_table,
                        meta->cache_key, meta->cache_key_len,
-                       &hash_meta_buf, &hash_meta_size);
+                       &out, &hash_meta_size);
+    hash_meta_buf = out;
     if (ret == -1) {
         /* Retrieve API server meta and merge with local meta */
         ret = get_and_merge_meta(ctx, meta,
@@ -727,8 +729,9 @@ int flb_kube_meta_get(struct flb_kube *ctx,
              * the outgoing buffer and size.
              */
             flb_free(hash_meta_buf);
-            flb_hash_get_by_id(ctx->hash_table, id, meta->cache_key,
-                               &hash_meta_buf, &hash_meta_size);
+            flb_hash_get_by_id(ctx->hash_table, id, meta->cache_key, meta->cache_key_len,
+                               &out, &hash_meta_size);
+            hash_meta_buf = out;
         }
     }
 

plugins/filter_ratelimit/CMakeLists.txt

@@ -0,0 +1,4 @@
+set(src
+  ratelimit.c)
+
+FLB_PLUGIN(filter_ratelimit "${src}" "")