plugin_proxy: enable event_type specification for proxy plugins

include/fluent-bit/flb_plugin_proxy.h

@@ -40,6 +40,7 @@ struct flb_plugin_proxy_def {
     int flags;
     char *name;               /* plugin short name                           */
     char *description;        /* plugin description                          */
+    int event_type;           /* event type (logs/metrics/traces)            */
 };
 
 /* Proxy context */

src/flb_plugin_proxy.c

@@ -362,6 +362,14 @@ static int flb_proxy_register_output(struct flb_plugin_proxy *proxy,
     out->flags = def->flags;
     out->name  = flb_strdup(def->name);
 
+    /* If event_type is unset (0) then default to logs (this is the current behavior) */
+    if (def->event_type == 0) {
+        out->event_type = FLB_OUTPUT_LOGS;
+    }
+    else {
+        out->event_type = def->event_type;
+    }
+
     out->description = def->description;
     mk_list_add(&out->_head, &config->out_plugins);
 
@@ -396,6 +404,7 @@ static int flb_proxy_register_input(struct flb_plugin_proxy *proxy,
     in->flags = def->flags | FLB_INPUT_THREADED;
     in->name  = flb_strdup(def->name);
     in->description = def->description;
+
     mk_list_add(&in->_head, &config->in_plugins);
 
     /*
@@ -612,7 +621,7 @@ struct flb_plugin_proxy *flb_plugin_proxy_create(const char *dso_path, int type,
         return NULL;
     }
 
-    proxy->def = flb_malloc(sizeof(struct flb_plugin_proxy_def));
+    proxy->def = flb_calloc(1, sizeof(struct flb_plugin_proxy_def));
     if (!proxy->def) {
         flb_errno();
         dlclose(handle);

tests/runtime_shell/CMakeLists.txt

@@ -17,6 +17,10 @@ set(UNIT_TESTS_SH
   processor_invalid.sh
   )
 
+if (CMAKE_SYSTEM_NAME STREQUAL "Linux")
+  list(APPEND UNIT_TESTS_SH proxy_logs_expect.sh)
+endif()
+
 # Prepare list of unit tests
 foreach(script ${UNIT_TESTS_SH})
   add_test(NAME ${script}
@@ -27,6 +31,7 @@ foreach(script ${UNIT_TESTS_SH})
     "FLB_ROOT=${PROJECT_SOURCE_DIR};\
 FLB_RUNTIME_SHELL_PATH=${CMAKE_CURRENT_SOURCE_DIR};\
 FLB_RUNTIME_SHELL_CONF=${CMAKE_CURRENT_SOURCE_DIR}/conf;\
-FLB_BIN=${CMAKE_BINARY_DIR}/bin/fluent-bit"
+FLB_BIN=${CMAKE_BINARY_DIR}/bin/fluent-bit;\
+FLB_BUILD=${CMAKE_BINARY_DIR}"
     )
 endforeach()

tests/runtime_shell/conf/proxy_logs_test.conf

@@ -0,0 +1,21 @@
+[SERVICE]
+    Flush        1
+    Grace        2
+    Log_Level    info
+    Daemon       Off
+
+[INPUT]
+    Name         dummy
+    Dummy        {"message": "test log entry", "level": "info"}
+    Samples      3
+    Tag          test.logs
+
+[OUTPUT]
+    Name         test_logs_go
+    Match        test.logs
+
+[OUTPUT]
+    Name         file
+    Match        test.logs
+    File         ${SIGNAL_FILE_PATH}
+    mkdir        on

tests/runtime_shell/go_plugins/build_test_plugins.sh

@@ -0,0 +1,113 @@
+#!/bin/bash
+
+# Build script for Go test plugins
+set -e
+
+GO_PLUGIN_DIR="${FLB_ROOT}/tests/runtime_shell/go_plugins"
+BUILD_DIR="${FLB_ROOT}/build"
+
+install_go_if_needed() {
+    if ! command -v go &> /dev/null; then 
+    echo "Go not found, installing Go..."
+
+    ARCH=$(uname -m)
+    case $ARCH in 
+        x86_64) GO_ARCH="amd64" ;;
+        aarch64|arm64) GO_ARCH="arm64" ;;
+        *) echo "Unsupported architecture: $ARCH"; exit 1 ;;
+    esac
+
+    OS=$(uname -s | tr '[:upper:]' '[:lower:]')
+    GO_VERSION="1.25.4"
+    GO_TARBALL="go${GO_VERSION}.${OS}-${GO_ARCH}.tar.gz"
+    GO_URL="https://golang.org/dl/${GO_TARBALL}"
+
+    echo "Downloading Go from $GO_URL..."
+
+    TEMP_DIR=$(mktemp -d)
+    cd "$TEMP_DIR"
+
+    if command -v curl > /dev/null 2>&1; then
+        curl -L -O "$GO_URL"
+    else
+        echo "Neither wget nor curl is available to download Go."
+        exit 1
+    fi
+
+    echo "Extracting Go tarball..."
+    ls -la
+
+    if [ ! -f "$GO_TARBALL" ]; then
+        echo "Failed to download Go tarball."
+        exit 1
+    fi
+
+    tar -xzf "$GO_TARBALL"
+
+    if [ -w "/usr/local" ]; then
+        if [ -d /usr/local/go ]; then
+            sudo rm -rf /usr/local/go
+        fi
+        sudo mv go /usr/local/go
+        export PATH="/usr/local/go/bin:$PATH"
+    else
+        echo "No write permission to /usr/local. Installing Go to $HOME/.local/go"
+        mkdir -p "$HOME/.local"
+        rm -rf "$HOME/.local/go"
+        mv go "$HOME/.local/go"
+        export PATH="$HOME/.local/go/bin:$PATH"
+    fi
+    cd - > /dev/null
+    rm -rf "$TEMP_DIR"
+    echo "Go installed successfully."
+    go version 
+else
+    echo "Go is already installed."
+fi
+}
+
+verify_go_cgo() {
+    echo "Verifying Go CGO support..."
+    if ! go env CGO_ENABLED | grep -q "1"; then
+        echo "Warning: CGO is not enabled. Attempting to enable CGO..."
+        export CGO_ENABLED=1
+    fi
+
+    TEMP_GO_FILE=$(mktemp --suffix=.go)
+    cat > "$TEMP_GO_FILE" << 'EOF'
+package main
+import "C"
+//export TestFunc
+func TestFunc() {}
+func main() {}
+EOF
+    TEMP_SO_FILE=$(mktemp --suffix=.so)
+    if go build -buildmode=c-shared -o "$TEMP_SO_FILE" "$TEMP_GO_FILE" 2> /dev/null; then
+        echo "CGO is enabled and working."
+        rm -f "$TEMP_GO_FILE" "$TEMP_SO_FILE"
+    else
+        echo "Error: CGO is not enabled or not working properly. Please ensure you have a C compiler installed."
+        rm -f "$TEMP_GO_FILE" "$TEMP_SO_FILE"
+        exit 1
+    fi
+}
+
+build_go_plugins() {
+    echo "Building Go test plugins..."
+
+    echo "Building logs output plugin..."
+    cd "$GO_PLUGIN_DIR"
+    CGO_ENABLED=1 GO111MODULE=on go build -buildmode=c-shared -v -ldflags="-s -w" -o $BUILD_DIR/test_logs_go.so logs_output.go
+    if [ $? -eq 0 ]; then 
+        echo "Go test plugins built successfully!"
+        echo "Logs plugin: $BUILD_DIR/test_logs_go.so"
+    else
+        echo "Failed to build Go test plugins."
+        exit 1
+    fi
+}
+
+echo "Setting up Go build environment..."
+install_go_if_needed
+verify_go_cgo
+build_go_plugins

tests/runtime_shell/go_plugins/go.mod

@@ -0,0 +1,8 @@
+module github.com/fluent/fluent-bit/tests/runtime_shell/go_plugins
+
+go 1.25.1
+
+require (
+    github.com/fluent/fluent-bit-go v0.0.0-20230731091245-a7a013e2473c
+    github.com/ugorji/go/codec v1.1.7 // indirect
+)

tests/runtime_shell/go_plugins/go.sum

@@ -0,0 +1,5 @@
+github.com/fluent/fluent-bit-go v0.0.0-20230731091245-a7a013e2473c h1:yKN46XJHYC/gvgH2UsisJ31+n4K3S7QYZSfU2uAWjuI=
+github.com/fluent/fluent-bit-go v0.0.0-20230731091245-a7a013e2473c/go.mod h1:L92h+dgwElEyUuShEwjbiHjseW410WIcNz+Bjutc8YQ=
+github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
+github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
+github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=

tests/runtime_shell/go_plugins/logs_output.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"C"
+	"fmt"
+	"unsafe"
+
+	"github.com/fluent/fluent-bit-go/output"
+)
+
+//export FLBPluginRegister
+func FLBPluginRegister(def unsafe.Pointer) int {
+	// Register as logs-only output plugin
+	return output.FLBPluginRegister(def, "test_logs_go", "Test Go Output Plugin for Logs")
+}
+
+//export FLBPluginInit
+func FLBPluginInit(plugin unsafe.Pointer) int {
+	return output.FLB_OK
+}
+
+//export FLBPluginFlushCtx
+func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, tag *C.char) int {
+	// Write to a stdout to verify it received data
+	dec := output.NewDecoder(data, int(length))
+	var logrecords []string
+	for {
+		ret, _, record := output.GetRecord(dec)
+		if ret != 0 {
+			break
+		}
+		logrecords = append(logrecords, fmt.Sprintf("%v", record))
+	}
+	for _, record := range logrecords {
+		fmt.Printf("%s\n", record)
+	}
+
+	return output.FLB_OK
+}
+
+//export FLBPluginExit
+func FLBPluginExit() int {
+	return output.FLB_OK
+}
+
+func main() {
+}

tests/runtime_shell/proxy_logs_expect.sh

@@ -0,0 +1,38 @@
+#!/bin/sh
+
+# Setup environment if not already set
+if [ -z "$FLB_BIN" ]; then
+    FLB_ROOT=${FLB_ROOT:-$(cd $(dirname $0)/../.. && pwd)}
+    FLB_BIN=${FLB_BIN:-$FLB_ROOT/build/bin/fluent-bit}
+fi
+
+echo "Using Fluent Bit at: $FLB_BIN"
+
+. $FLB_RUNTIME_SHELL_PATH/go_plugins/build_test_plugins.sh
+
+test_proxy_logs_compatibility() {
+    export SIGNAL_FILE_PATH="/tmp/flb_signal_logs_$$.txt"
+    STDOUT_OUTPUT_FILE="/tmp/test_logs_stdout_$$.txt"
+
+    rm -f "$STDOUT_OUTPUT_FILE" "$SIGNAL_FILE_PATH"
+
+    $FLB_BIN -e $FLB_ROOT/build/test_logs_go.so -c $FLB_RUNTIME_SHELL_CONF/proxy_logs_test.conf > "$STDOUT_OUTPUT_FILE" 2>&1 &
+    FLB_PID=$!
+
+    sleep 3
+
+    if [ -f "$STDOUT_OUTPUT_FILE" ]; then
+        echo "SUCCESS: Captured Fluent Bit output"
+        echo "Output contents:"
+        cat "$STDOUT_OUTPUT_FILE"
+    else
+        echo "FAIL: No stdout output captured"
+        return 1
+    fi
+
+    # Clean up
+    rm -f "$STDOUT_OUTPUT_FILE" "$SIGNAL_FILE_PATH"
+}
+
+# Load the runtime shell environment
+. $FLB_RUNTIME_SHELL_PATH/runtime_shell.env