output: engine: output_throttle: Implement throttle operations for output plugins

.github/scripts/ctest-wrapper/ctest

@@ -0,0 +1,27 @@
+#!/bin/bash
+
+set -u
+
+if [ -z "${FLB_CTEST_REAL_PATH:-}" ]; then
+    echo "FLB_CTEST_REAL_PATH must name the real ctest executable" >&2
+    exit 2
+fi
+
+wrapper_path="$(cd "$(dirname "$0")" && pwd -P)/$(basename "$0")"
+ctest_path="$(cd "$(dirname "$FLB_CTEST_REAL_PATH")" && pwd -P)/$(basename "$FLB_CTEST_REAL_PATH")"
+
+if [ "$ctest_path" = "$wrapper_path" ]; then
+    echo "FLB_CTEST_REAL_PATH resolves to the ctest timeout wrapper" >&2
+    exit 2
+fi
+
+timeout_seconds="${FLB_CTEST_DEFAULT_TIMEOUT_SECONDS:-300}"
+
+case "$timeout_seconds" in
+    ''|*[!0-9]*|0)
+        echo "FLB_CTEST_DEFAULT_TIMEOUT_SECONDS must be a positive integer" >&2
+        exit 2
+        ;;
+esac
+
+exec "$FLB_CTEST_REAL_PATH" "$@" --timeout "$timeout_seconds"

.github/scripts/tests/test_ctest_timeout_wrapper.sh

@@ -0,0 +1,91 @@
+#!/bin/bash
+
+set -eu
+
+script_dir="$(cd "$(dirname "$0")" && pwd -P)"
+wrapper="$script_dir/../ctest-wrapper/ctest"
+temporary_dir="$(mktemp -d "${TMPDIR:-/tmp}/flb-ctest-wrapper.XXXXXX")"
+
+cleanup()
+{
+    rm -rf "$temporary_dir"
+}
+
+trap cleanup EXIT
+
+fake_ctest="$temporary_dir/fake ctest"
+arguments_file="$temporary_dir/arguments"
+
+cat > "$fake_ctest" <<'EOF'
+#!/bin/bash
+printf '%s\n' "$@" > "$FLB_CTEST_ARGUMENTS_FILE"
+exit "${FLB_CTEST_FAKE_EXIT_CODE:-0}"
+EOF
+chmod +x "$fake_ctest"
+
+FLB_CTEST_REAL_PATH="$fake_ctest" \
+FLB_CTEST_ARGUMENTS_FILE="$arguments_file" \
+FLB_CTEST_DEFAULT_TIMEOUT_SECONDS=17 \
+    "$wrapper" --test-dir "directory with spaces" --output-on-failure
+
+expected_arguments="$temporary_dir/expected-arguments"
+cat > "$expected_arguments" <<'EOF'
+--test-dir
+directory with spaces
+--output-on-failure
+--timeout
+17
+EOF
+cmp "$expected_arguments" "$arguments_file"
+
+set +e
+FLB_CTEST_REAL_PATH="$fake_ctest" \
+FLB_CTEST_ARGUMENTS_FILE="$arguments_file" \
+FLB_CTEST_FAKE_EXIT_CODE=42 \
+    "$wrapper"
+wrapper_status=$?
+set -e
+
+if [ "$wrapper_status" -ne 42 ]; then
+    echo "wrapper returned $wrapper_status instead of the ctest status 42" >&2
+    exit 1
+fi
+
+timeout_project="$temporary_dir/default-timeout"
+mkdir "$timeout_project"
+cat > "$timeout_project/CTestTestfile.cmake" <<'EOF'
+add_test(slow-test /bin/sleep 2)
+EOF
+
+set +e
+FLB_CTEST_REAL_PATH="$(command -v ctest)" \
+FLB_CTEST_DEFAULT_TIMEOUT_SECONDS=1 \
+    "$wrapper" --test-dir "$timeout_project" --output-on-failure
+timeout_status=$?
+set -e
+
+if [ "$timeout_status" -eq 0 ]; then
+    echo "ctest unexpectedly passed a test longer than the default timeout" >&2
+    exit 1
+fi
+
+property_project="$temporary_dir/explicit-timeout"
+mkdir "$property_project"
+cat > "$property_project/CTestTestfile.cmake" <<'EOF'
+add_test(slow-test /bin/sleep 2)
+set_tests_properties(slow-test PROPERTIES TIMEOUT 3)
+EOF
+
+FLB_CTEST_REAL_PATH="$(command -v ctest)" \
+FLB_CTEST_DEFAULT_TIMEOUT_SECONDS=1 \
+    "$wrapper" --test-dir "$property_project" --output-on-failure
+
+set +e
+FLB_CTEST_REAL_PATH="$wrapper" "$wrapper"
+recursion_status=$?
+set -e
+
+if [ "$recursion_status" -ne 2 ]; then
+    echo "wrapper did not reject itself as the real ctest executable" >&2
+    exit 1
+fi

.github/workflows/unit-tests.yaml

@@ -189,12 +189,24 @@ jobs:
           echo "CC = $CC, CXX = $CXX, FLB_OPT = $FLB_OPT"
           brew update
           brew install bison flex openssl || true
+          export FLB_CTEST_REAL_PATH="$(command -v ctest)"
+          export PATH="$GITHUB_WORKSPACE/.github/scripts/ctest-wrapper:$PATH"
           ci/scripts/run-unit-tests.sh
         env:
           CC: gcc
           CXX: g++
           FLB_OPT: ${{ matrix.flb_option }}
 
+      - name: Upload CTest failure diagnostics
+        if: ${{ failure() || cancelled() }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: macos-ctest-logs-${{ matrix.flb_option }}
+          path: |
+            build/Testing/Temporary/LastTest.log
+            build/Testing/Temporary/LastTest.log.tmp
+          if-no-files-found: warn
+
   run-aarch64-unit-tests:
     runs-on: ${{(github.repository == 'fluent/fluent-bit') && 'ubuntu-24.04-arm' || 'ubuntu-latest' }}
     permissions:

benchmarks/CMakeLists.txt

@@ -17,3 +17,9 @@ target_link_libraries(flb-bench-processor_sampling
   fluent-bit-static
   ${CMAKE_THREAD_LIBS_INIT}
 )
+
+add_executable(flb-bench-output_throttle flb-bench-output_throttle.c)
+target_link_libraries(flb-bench-output_throttle
+  fluent-bit-static
+  ${CMAKE_THREAD_LIBS_INIT}
+)

benchmarks/flb-bench-output_throttle.c

@@ -0,0 +1,169 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+#include <inttypes.h>
+#include <pthread.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#include <cfl/cfl_time.h>
+#include <fluent-bit/flb_output_throttle.h>
+
+#define DEFAULT_ITERATIONS 1000000
+#define CONTENTION_THREADS 4
+
+struct benchmark_worker {
+    uint64_t iterations;
+    uint64_t admitted;
+    struct flb_output_throttle *gate;
+};
+
+static void *run_admissions(void *data)
+{
+    uint64_t index;
+    uint64_t generation;
+    struct benchmark_worker *worker;
+
+    worker = data;
+    for (index = 0; index < worker->iterations; index++) {
+        worker->admitted += flb_output_throttle_admit(
+                                worker->gate,
+                                flb_output_throttle_now_ms(),
+                                &generation);
+    }
+
+    return NULL;
+}
+
+static double elapsed_seconds(uint64_t start, uint64_t end)
+{
+    return (double) (end - start) / 1000000000.0;
+}
+
+static void report(const char *name, uint64_t operations, double seconds)
+{
+    printf("%-30s %12" PRIu64 " ops %10.3f Mops/s\n",
+           name, operations, (double) operations / seconds / 1000000.0);
+}
+
+static double run_parallel(struct flb_output_throttle *gate,
+                           uint64_t iterations)
+{
+    int index;
+    int created;
+    uint64_t start;
+    uint64_t end;
+    pthread_t threads[CONTENTION_THREADS];
+    struct benchmark_worker workers[CONTENTION_THREADS];
+
+    created = 0;
+    start = cfl_time_now();
+    for (index = 0; index < CONTENTION_THREADS; index++) {
+        workers[index].iterations = iterations;
+        workers[index].admitted = 0;
+        workers[index].gate = gate;
+        if (pthread_create(&threads[index], NULL,
+                           run_admissions, &workers[index]) != 0) {
+            break;
+        }
+        created++;
+    }
+    for (index = 0; index < created; index++) {
+        pthread_join(threads[index], NULL);
+    }
+    end = cfl_time_now();
+
+    if (created != CONTENTION_THREADS) {
+        return -1.0;
+    }
+    return elapsed_seconds(start, end);
+}
+
+static void report_overhead(const char *name, double baseline, double candidate)
+{
+    printf("%-30s %10.2f%%\n", name,
+           (candidate / baseline - 1.0) * 100.0);
+}
+
+int main(int argc, char **argv)
+{
+    uint64_t iterations;
+    uint64_t start;
+    uint64_t end;
+    uint64_t generation;
+    double disabled_seconds;
+    double enabled_seconds;
+    double disabled_parallel_seconds;
+    double enabled_parallel_seconds;
+    volatile uint64_t baseline;
+    struct benchmark_worker workers[CONTENTION_THREADS];
+    struct flb_output_throttle disabled_gate;
+    struct flb_output_throttle enabled_gate;
+
+    iterations = DEFAULT_ITERATIONS;
+    if (argc > 1) {
+        iterations = strtoull(argv[1], NULL, 10);
+    }
+    if (iterations == 0) {
+        fprintf(stderr, "iterations must be greater than zero\n");
+        return EXIT_FAILURE;
+    }
+
+    if (flb_output_throttle_init(&disabled_gate,
+                                 FLB_FALSE, 1000, 60000) != 0) {
+        fprintf(stderr, "could not initialize throttle gates\n");
+        return EXIT_FAILURE;
+    }
+    if (flb_output_throttle_init(&enabled_gate,
+                                 FLB_TRUE, 1000, 60000) != 0) {
+        fprintf(stderr, "could not initialize throttle gates\n");
+        flb_output_throttle_destroy(&disabled_gate);
+        return EXIT_FAILURE;
+    }
+
+    baseline = 0;
+    start = cfl_time_now();
+    for (generation = 0; generation < iterations; generation++) {
+        baseline += flb_output_throttle_now_ms() != UINT64_MAX;
+    }
+    end = cfl_time_now();
+    report("clock-only baseline", iterations, elapsed_seconds(start, end));
+
+    workers[0].iterations = iterations;
+    workers[0].admitted = 0;
+    workers[0].gate = &disabled_gate;
+    start = cfl_time_now();
+    run_admissions(&workers[0]);
+    end = cfl_time_now();
+    disabled_seconds = elapsed_seconds(start, end);
+    report("disabled gate", iterations, disabled_seconds);
+
+    workers[0].admitted = 0;
+    workers[0].gate = &enabled_gate;
+    start = cfl_time_now();
+    run_admissions(&workers[0]);
+    end = cfl_time_now();
+    enabled_seconds = elapsed_seconds(start, end);
+    report("enabled ready gate", iterations, enabled_seconds);
+    report_overhead("enabled vs disabled overhead",
+                    disabled_seconds, enabled_seconds);
+
+    disabled_parallel_seconds = run_parallel(&disabled_gate, iterations);
+    enabled_parallel_seconds = run_parallel(&enabled_gate, iterations);
+    if (disabled_parallel_seconds < 0 || enabled_parallel_seconds < 0) {
+        fprintf(stderr, "could not create benchmark threads\n");
+        flb_output_throttle_destroy(&enabled_gate);
+        flb_output_throttle_destroy(&disabled_gate);
+        return EXIT_FAILURE;
+    }
+    report("disabled gate, 4 threads", iterations * CONTENTION_THREADS,
+           disabled_parallel_seconds);
+    report("enabled ready gate, 4 threads",
+           iterations * CONTENTION_THREADS, enabled_parallel_seconds);
+    report_overhead("4-thread enabled overhead",
+                    disabled_parallel_seconds, enabled_parallel_seconds);
+
+    flb_output_throttle_destroy(&enabled_gate);
+    flb_output_throttle_destroy(&disabled_gate);
+    return baseline == 0;
+}

include/fluent-bit/flb_http_retry_after.h

@@ -0,0 +1,46 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_HTTP_RETRY_AFTER_H
+#define FLB_HTTP_RETRY_AFTER_H
+
+#include <fluent-bit/flb_macros.h>
+
+#include <stddef.h>
+#include <stdint.h>
+
+enum flb_retry_after_status {
+    FLB_RETRY_AFTER_ABSENT = 0,
+    FLB_RETRY_AFTER_VALID,
+    FLB_RETRY_AFTER_INVALID,
+    FLB_RETRY_AFTER_SATURATED
+};
+
+FLB_EXPORT int flb_http_retry_after_parse(const char *value,
+                                          size_t length,
+                                          int64_t received_wall_time_ms,
+                                          uint64_t *delay_ms);
+
+FLB_EXPORT int flb_http_retry_after_parse_headers(const char *headers,
+                                                  size_t length,
+                                                  int64_t received_wall_time_ms,
+                                                  uint64_t *delay_ms,
+                                                  size_t *invalid_count);
+
+#endif

include/fluent-bit/flb_macros.h

@@ -30,6 +30,7 @@
 #define FLB_ERROR   0
 #define FLB_OK      1
 #define FLB_RETRY   2
+#define FLB_THROTTLE 3
 
 /* ala-printf format check */
 #if defined(__GNUC__) || defined(__clang__)