Merge branch 'fluent:master' into dietWall/fix_machine_id_in_docker

Author: dietWall

.github/workflows/staging-release.yaml

@@ -842,9 +842,20 @@ jobs:
           tag_name: v${{ inputs.version }}
           make_latest: false
 
-      - name: Release 4.1 and latest
+      - name: Release 4.1 - not latest
         uses: softprops/action-gh-release@v2
         if: startsWith(inputs.version, '4.1')
+        with:
+          body: "https://fluentbit.io/announcements/v${{ inputs.version }}/"
+          draft: false
+          generate_release_notes: true
+          name: "Fluent Bit ${{ inputs.version }}"
+          tag_name: v${{ inputs.version }}
+          make_latest: false
+
+      - name: Release 4.2 and latest
+        uses: softprops/action-gh-release@v2
+        if: startsWith(inputs.version, '4.2')
         with:
           body: "https://fluentbit.io/announcements/v${{ inputs.version }}/"
           draft: false
@@ -969,13 +980,20 @@ jobs:
           token: ${{ secrets.GH_PA_TOKEN }}
           ref: '4.0'
 
-      - name: Release 4.1 and latest
+      - name: Release 4.1 - not latest
         if: startsWith(inputs.version, '4.1')
         uses: actions/checkout@v5
         with:
           repository: fluent/fluent-bit-docs
           token: ${{ secrets.GH_PA_TOKEN }}
 
+      - name: Release 4.2 and latest
+        if: startsWith(inputs.version, '4.2')
+        uses: actions/checkout@v5
+        with:
+          repository: fluent/fluent-bit-docs
+          token: ${{ secrets.GH_PA_TOKEN }}
+
       - name: Ensure we have the script we need
         run: |
           if [[ ! -f update-release-version-docs.sh ]] ; then
@@ -1076,6 +1094,12 @@ jobs:
       - name: Release 4.1
         if: startsWith(inputs.version, '4.1')
         uses: actions/checkout@v5
+        with:
+          ref: 4.1
+
+      - name: Release 4.2
+        if: startsWith(inputs.version, '4.2')
+        uses: actions/checkout@v5
         with:
           ref: master
 

.github/workflows/unit-tests.yaml

@@ -80,6 +80,23 @@ jobs:
     permissions:
       contents: read
     steps:
+      - name: Free disk space
+        run: |
+          echo "::group::Disk usage before cleanup"
+          df -h
+          echo "::endgroup::"
+
+          sudo rm -rf /usr/share/dotnet || true
+          sudo rm -rf /usr/local/lib/android || true
+          sudo rm -rf /opt/ghc || true
+          sudo rm -rf /usr/lib/jvm || true
+
+          sudo apt-get clean || true
+
+          echo "::group::Disk usage after cleanup"
+          df -h
+          echo "::endgroup::"
+
       - name: Setup environment
         run: |
           sudo apt-get update

CMakeLists.txt

@@ -8,7 +8,7 @@ set(CMAKE_POLICY_DEFAULT_CMP0069 NEW)
 # Fluent Bit Version
 set(FLB_VERSION_MAJOR  4)
 set(FLB_VERSION_MINOR  2)
-set(FLB_VERSION_PATCH  0)
+set(FLB_VERSION_PATCH  1)
 set(FLB_VERSION_STR "${FLB_VERSION_MAJOR}.${FLB_VERSION_MINOR}.${FLB_VERSION_PATCH}")
 
 set(CMAKE_POSITION_INDEPENDENT_CODE ON)

dockerfiles/Dockerfile

@@ -13,7 +13,7 @@
 # docker buildx build --platform "linux/amd64,linux/arm64,linux/arm/v7,linux/s390x" -f ./dockerfiles/Dockerfile.multiarch --build-arg FLB_TARBALL=https://github.com/fluent/fluent-bit/archive/v1.8.11.tar.gz ./dockerfiles/
 
 # Set this to the current release version: it gets done so as part of the release.
-ARG RELEASE_VERSION=4.1.2
+ARG RELEASE_VERSION=4.2.1
 
 # For multi-arch builds - assumption is running on an AMD64 host
 FROM multiarch/qemu-user-static:x86_64-arm AS qemu-arm32

fluent-bit-4.2.0.bb renamed to fluent-bit-4.2.1.bb

@@ -16,7 +16,7 @@ LIC_FILES_CHKSUM = "file://LICENSE;md5=2ee41112a44fe7014dce33e26468ba93"
 SECTION = "net"
 
 PR = "r0"
-PV = "4.2.0"
+PV = "4.2.1"
 
 SRCREV = "v${PV}"
 SRC_URI = "git://github.com/fluent/fluent-bit.git;nobranch=1"

plugins/in_forward/fw.c

@@ -124,28 +124,37 @@ static int fw_unix_create(struct flb_in_fw_config *ctx)
 static int in_fw_collect(struct flb_input_instance *ins,
                          struct flb_config *config, void *in_context)
 {
+    int                      state_backup;
     struct flb_connection   *connection;
     struct fw_conn          *conn;
     struct flb_in_fw_config *ctx;
 
     ctx = in_context;
 
+    state_backup = ctx->state;
+    ctx->state = FW_INSTANCE_STATE_ACCEPTING_CLIENT;
+
     connection = flb_downstream_conn_get(ctx->downstream);
 
     if (connection == NULL) {
         flb_plg_error(ctx->ins, "could not accept new connection");
+        ctx->state = state_backup;
 
         return -1;
     }
 
     if (!config->is_ingestion_active) {
         flb_downstream_conn_release(connection);
+        ctx->state = state_backup;
+
         return -1;
     }
 
     if(ctx->is_paused) {
         flb_downstream_conn_release(connection);
         flb_plg_trace(ins, "TCP connection will be closed FD=%i", connection->fd);
+        ctx->state = state_backup;
+
         return -1;
     }
 
@@ -154,9 +163,17 @@ static int in_fw_collect(struct flb_input_instance *ins,
     conn = fw_conn_add(connection, ctx);
     if (!conn) {
         flb_downstream_conn_release(connection);
+        ctx->state = state_backup;
+
         return -1;
     }
 
+    ctx->state = state_backup;
+
+    if (ctx->state == FW_INSTANCE_STATE_PAUSED) {
+        fw_conn_del_all(ctx);
+    }
+
     return 0;
 }
 
@@ -263,6 +280,7 @@ static int in_fw_init(struct flb_input_instance *ins,
         return -1;
     }
 
+    ctx->state = FW_INSTANCE_STATE_RUNNING;
     ctx->coll_fd = -1;
     ctx->ins = ins;
     mk_list_init(&ctx->connections);
@@ -386,7 +404,10 @@ static void in_fw_pause(void *data, struct flb_config *config)
             return;
         }
 
-        fw_conn_del_all(ctx);
+        if (ctx->state == FW_INSTANCE_STATE_RUNNING) {
+            fw_conn_del_all(ctx);
+        }
+
         ctx->is_paused = FLB_TRUE;
         ret = pthread_mutex_unlock(&ctx->conn_mutex);
         if (ret != 0) {
@@ -406,6 +427,8 @@ static void in_fw_pause(void *data, struct flb_config *config)
     if (config->is_ingestion_active == FLB_FALSE) {
         fw_conn_del_all(ctx);
     }
+
+    ctx->state = FW_INSTANCE_STATE_PAUSED;
 }
 
 static void in_fw_resume(void *data, struct flb_config *config) {
@@ -427,6 +450,8 @@ static void in_fw_resume(void *data, struct flb_config *config) {
             flb_plg_error(ctx->ins, "cannot unlock collector mutex");
             return;
         }
+
+        ctx->state = FW_INSTANCE_STATE_RUNNING;
     }
 }
 

plugins/in_forward/fw.h

@@ -25,6 +25,12 @@
 #include <fluent-bit/flb_log_event_decoder.h>
 #include <fluent-bit/flb_log_event_encoder.h>
 
+#define FW_INSTANCE_STATE_RUNNING           0
+#define FW_INSTANCE_STATE_ACCEPTING_CLIENT  1
+#define FW_INSTANCE_STATE_PROCESSING_PACKET 2
+#define FW_INSTANCE_STATE_PAUSED            3
+
+
 enum {
     FW_HANDSHAKE_HELO        = 1,
     FW_HANDSHAKE_PINGPONG    = 2,
@@ -76,6 +82,8 @@ struct flb_in_fw_config {
 
     pthread_mutex_t conn_mutex;
 
+    int state;
+    
     /* Plugin is paused */
     int is_paused;
 };

plugins/in_forward/fw_conn.c

@@ -28,8 +28,7 @@
 #include "fw_prot.h"
 #include "fw_conn.h"
 
-/* Callback invoked every time an event is triggered for a connection */
-int fw_conn_event(void *data)
+static int fw_conn_event_internal(struct flb_connection *connection)
 {
     int ret;
     int bytes;
@@ -39,9 +38,6 @@ int fw_conn_event(void *data)
     struct fw_conn *conn;
     struct mk_event *event;
     struct flb_in_fw_config *ctx;
-    struct flb_connection *connection;
-
-    connection = (struct flb_connection *) data;
 
     conn = connection->user_data;
 
@@ -127,6 +123,37 @@ int fw_conn_event(void *data)
     return 0;
 }
 
+/* Callback invoked every time an event is triggered for a connection */
+int fw_conn_event(void *data)
+{
+    struct flb_in_fw_config *ctx;
+    struct fw_conn          *conn;
+    int                      result;
+    struct flb_connection   *connection;
+    int                      state_backup;
+
+    connection = (struct flb_connection *) data;
+
+    conn = connection->user_data;
+
+    ctx = conn->ctx;
+
+    state_backup = ctx->state;
+
+    ctx->state = FW_INSTANCE_STATE_PROCESSING_PACKET;
+
+    result = fw_conn_event_internal(connection);
+
+    if (ctx->state == FW_INSTANCE_STATE_PROCESSING_PACKET) {
+        ctx->state = state_backup;
+    }
+    else if (ctx->state == FW_INSTANCE_STATE_PAUSED) {
+        fw_conn_del_all(ctx);
+    }
+
+    return result;
+}
+
 /* Create a new Forward request instance */
 struct fw_conn *fw_conn_add(struct flb_connection *connection, struct flb_in_fw_config *ctx)
 {

snap/snapcraft.yaml

@@ -1,6 +1,6 @@
 name: fluent-bit
 base: core20
-version: '4.2.0'
+version: '4.2.1'
 summary: High performance logs and stream processor
 description: |
   Fluent Bit is a high performance log processor and stream processor for Linux.