http_server: add new API to configure the maximum buffer size

include/fluent-bit/http_server/flb_http_server.h

@@ -46,6 +46,7 @@
 #define HTTP_SERVER_SUCCESS                    0
 #define HTTP_SERVER_PROVIDER_ERROR            -1
 #define HTTP_SERVER_ALLOCATION_ERROR          -2
+#define HTTP_SERVER_BUFFER_LIMIT_EXCEEDED     -3
 
 #define HTTP_SERVER_UNINITIALIZED              0
 #define HTTP_SERVER_INITIALIZED                1
@@ -76,6 +77,7 @@ struct flb_http_server {
     flb_http_server_request_processor_callback
                            request_callback;
     void                  *user_data;
+    size_t                 buffer_max_size;
 };
 
 struct flb_http_server_session {
@@ -121,6 +123,10 @@ int flb_http_server_stop(struct flb_http_server *session);
 
 int flb_http_server_destroy(struct flb_http_server *session);
 
+void flb_http_server_set_buffer_max_size(struct flb_http_server *server, size_t size);
+
+size_t flb_http_server_get_buffer_max_size(struct flb_http_server *server);
+
 /* HTTP SESSION */
 
 int flb_http_server_session_init(struct flb_http_server_session *session, int version);

plugins/in_elasticsearch/in_elasticsearch.c

@@ -172,6 +172,8 @@ static int in_elasticsearch_bulk_init(struct flb_input_instance *ins,
             return -1;
         }
 
+        flb_http_server_set_buffer_max_size(&ctx->http_server, ctx->buffer_max_size);
+
         ctx->http_server.request_callback = in_elasticsearch_bulk_prot_handle_ng;
 
         flb_input_downstream_set(ctx->http_server.downstream, ctx->ins);

plugins/in_http/http.c

@@ -129,6 +129,8 @@ static int in_http_init(struct flb_input_instance *ins,
             return -1;
         }
 
+        flb_http_server_set_buffer_max_size(&ctx->http_server, ctx->buffer_max_size);
+
         ctx->http_server.request_callback = http_prot_handle_ng;
 
         flb_input_downstream_set(ctx->http_server.downstream, ctx->ins);

plugins/in_http/http_conn.c

@@ -56,13 +56,17 @@ static int http_conn_event(void *data)
 {
     int status;
     size_t size;
+    size_t sent;
     ssize_t available;
     ssize_t bytes;
     size_t request_len;
     struct flb_connection *connection;
     struct http_conn *conn;
     struct mk_event *event;
     struct flb_http *ctx;
+    char *request_too_large = "HTTP/1.1 413 Request Entity Too Large\r\n" \
+                              "Content-Length: 0\r\n" \
+                              "Connection: close\r\n\r\n";
 
     connection = (struct flb_connection *) data;
 
@@ -79,6 +83,9 @@ static int http_conn_event(void *data)
                 flb_plg_trace(ctx->ins,
                               "fd=%i incoming data exceed limit (%zu KB)",
                               event->fd, (ctx->buffer_max_size / 1024));
+
+                flb_io_net_write(conn->connection,
+                                 (void *) request_too_large, strlen(request_too_large), &sent);
                 http_conn_del(conn);
                 return -1;
             }

plugins/in_http/http_prot.c

@@ -129,6 +129,14 @@ static int send_response(struct http_conn *conn, int http_status, char *message)
                        FLB_VERSION_STR,
                        context->success_headers_str);
     }
+    else if (http_status == 413) {
+        flb_sds_printf(&out,
+                       "HTTP/1.1 413 Request Entity Too Large\r\n"
+                       "Server: Fluent Bit v%s\r\n"
+                       "Content-Length: %i\r\n\r\n%s",
+                       FLB_VERSION_STR,
+                       len, message ? message : "");
+    }
     else if (http_status == 400) {
         flb_sds_printf(&out,
                        "HTTP/1.1 400 Bad Request\r\n"
@@ -1007,6 +1015,9 @@ static int send_response_ng(struct flb_http_response *response,
     else if (http_status == 400) {
         flb_http_response_set_message(response, "Bad Request");
     }
+    else if (http_status == 413) {
+        flb_http_response_set_message(response, "Payload Too Large");
+    }
 
     if (http_status == 200 ||
         http_status == 201 ||

plugins/in_opentelemetry/opentelemetry.c

@@ -126,6 +126,8 @@ static int in_opentelemetry_init(struct flb_input_instance *ins,
             return -1;
         }
 
+        flb_http_server_set_buffer_max_size(&ctx->http_server, ctx->buffer_max_size);
+
         ctx->http_server.request_callback = opentelemetry_prot_handle_ng;
 
         flb_input_downstream_set(ctx->http_server.downstream, ctx->ins);

plugins/in_prometheus_remote_write/prom_rw.c

@@ -126,6 +126,8 @@ static int prom_rw_init(struct flb_input_instance *ins,
             return -1;
         }
 
+        flb_http_server_set_buffer_max_size(&ctx->http_server, ctx->buffer_max_size);
+
         ctx->http_server.request_callback = prom_rw_prot_handle_ng;
 
         flb_input_downstream_set(ctx->http_server.downstream, ctx->ins);

plugins/in_splunk/splunk.c

@@ -130,6 +130,8 @@ static int in_splunk_init(struct flb_input_instance *ins,
             return -1;
         }
 
+        flb_http_server_set_buffer_max_size(&ctx->http_server, ctx->buffer_max_size);
+
         ctx->http_server.request_callback = splunk_prot_handle_ng;
 
         flb_input_downstream_set(ctx->http_server.downstream, ctx->ins);

src/http_server/flb_http_server.c

@@ -18,6 +18,7 @@
  */
 
 #include <fluent-bit/flb_mem.h>
+#include <string.h>
 
 #include <fluent-bit/http_server/flb_http_server.h>
 
@@ -28,8 +29,12 @@
 
 static int flb_http_server_session_read(struct flb_http_server_session *session)
 {
-    unsigned char input_buffer[1024];
+    size_t sent;
     ssize_t result;
+    unsigned char input_buffer[1024];
+    char *request_too_large = "HTTP/1.1 413 Request Entity Too Large\r\n"
+                              "Content-Length: 0\r\n"
+                              "Connection: close\r\n\r\n";
 
     result = flb_io_net_read(session->connection,
                              (void *) &input_buffer,
@@ -43,7 +48,11 @@ static int flb_http_server_session_read(struct flb_http_server_session *session)
                                                       input_buffer,
                                                       result);
 
-    if (result < 0) {
+    if (result == HTTP_SERVER_BUFFER_LIMIT_EXCEEDED) {
+        flb_io_net_write(session->connection, (void *) request_too_large, strlen(request_too_large), &sent);
+        return -1;
+    }
+    else if (result < 0) {
         return -1;
     }
 
@@ -328,6 +337,7 @@ int flb_http_server_init(struct flb_http_server *session,
     session->system_context = system_context;
 
     session->downstream = NULL;
+    session->buffer_max_size = HTTP_SERVER_MAXIMUM_BUFFER_SIZE;
 
     cfl_list_init(&session->clients);
 
@@ -419,6 +429,17 @@ int flb_http_server_destroy(struct flb_http_server *server)
     return 0;
 }
 
+void flb_http_server_set_buffer_max_size(struct flb_http_server *server,
+                                         size_t size)
+{
+    server->buffer_max_size = size;
+}
+
+size_t flb_http_server_get_buffer_max_size(struct flb_http_server *server)
+{
+    return server->buffer_max_size;
+}
+
 /* HTTP SESSION */
 
 int flb_http_server_session_init(struct flb_http_server_session *session, int version)
@@ -517,8 +538,14 @@ int flb_http_server_session_ingest(struct flb_http_server_session *session,
                             unsigned char *buffer,
                             size_t length)
 {
-    cfl_sds_t resized_buffer;
     int       result;
+    size_t    max_size;
+    cfl_sds_t resized_buffer;
+
+    max_size = flb_http_server_get_buffer_max_size(session->parent);
+    if (session->parent != NULL && cfl_sds_len(session->incoming_data) + length > max_size) {
+        return HTTP_SERVER_BUFFER_LIMIT_EXCEEDED;
+    }
 
     if (session->version == HTTP_PROTOCOL_VERSION_AUTODETECT ||
         session->version <= HTTP_PROTOCOL_VERSION_11) {