diff --git a/.github/workflows/call-build-macos.yaml b/.github/workflows/call-build-macos.yaml index cd0592c2834..eb48cbbc1aa 100644 --- a/.github/workflows/call-build-macos.yaml +++ b/.github/workflows/call-build-macos.yaml @@ -90,7 +90,7 @@ jobs: - name: Install dependencies run: | brew update - brew install bison flex libyaml openssl pkgconfig || true + brew install bison flex libyaml openssl pkgconfig rabbitmq-c || true - name: Install cmake uses: jwlawson/actions-setup-cmake@v2 diff --git a/CMakeLists.txt b/CMakeLists.txt index e122d222891..34d76b0637e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1443,6 +1443,14 @@ if (FLB_IN_EBPF) endif() +# AMQP +# ========== +find_package(rabbitmq-c 0.12) +if(FLB_IN_AMQP AND (NOT rabbitmq-c_FOUND)) + message(STATUS "rabbitmq-c is not found. Disabling AMQP support.") + FLB_OPTION(FLB_IN_AMQP OFF) +endif() + # Pthread Local Storage # ===================== # By default we expect the compiler already support thread local storage diff --git a/cmake/plugins_options.cmake b/cmake/plugins_options.cmake index 7fbd0a96bd5..50fb8f50307 100644 --- a/cmake/plugins_options.cmake +++ b/cmake/plugins_options.cmake @@ -11,6 +11,7 @@ option(FLB_MINIMAL "Enable minimal build configuration" No) # Inputs (sources, data collectors) # ================================= +DEFINE_OPTION(FLB_IN_AMQP "Enable AMQP input plugin" ON) DEFINE_OPTION(FLB_IN_BLOB "Enable Blob input plugin" ON) DEFINE_OPTION(FLB_IN_CALYPTIA_FLEET "Enable Calyptia Fleet input plugin" ON) DEFINE_OPTION(FLB_IN_COLLECTD "Enable Collectd input plugin" ON) diff --git a/cmake/windows-setup.cmake b/cmake/windows-setup.cmake index dac41f2220e..b59175ccdcb 100644 --- a/cmake/windows-setup.cmake +++ b/cmake/windows-setup.cmake @@ -48,6 +48,7 @@ if(FLB_WINDOWS_DEFAULTS) # INPUT plugins # ============= + set(FLB_IN_AMQP No) set(FLB_IN_CPU No) set(FLB_IN_DISK No) set(FLB_IN_EXEC Yes) diff --git a/packaging/distros/almalinux/Dockerfile b/packaging/distros/almalinux/Dockerfile index 8fc050b02e7..f8da8412d8e 100644 --- a/packaging/distros/almalinux/Dockerfile +++ b/packaging/distros/almalinux/Dockerfile @@ -20,11 +20,14 @@ RUN yum -y update && \ yum install -y rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # almalinux/8.arm64v8 base image # hadolint ignore=DL3029 @@ -40,7 +43,8 @@ RUN yum -y update && \ yum install -y rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off @@ -48,6 +52,8 @@ ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" ENV FLB_JEMALLOC_OPTIONS=$FLB_JEMALLOC_OPTIONS +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP FROM almalinux:9 AS almalinux-9-base @@ -59,11 +65,14 @@ RUN yum -y update && \ yum install -y --allowerasing rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # almalinux/8.arm64v8 base image # hadolint ignore=DL3029 @@ -79,7 +88,8 @@ RUN yum -y update && \ yum install -y --allowerasing rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off @@ -87,6 +97,8 @@ ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" ENV FLB_JEMALLOC_OPTIONS=$FLB_JEMALLOC_OPTIONS +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP FROM almalinux:10 AS almalinux-10-base @@ -99,12 +111,15 @@ RUN yum -y update && \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ - libzstd-devel && \ + libzstd-devel \ + librabbitmq-devel && \ yum clean all # Use system installed libzstd library. ARG FLB_PREFER_SYSTEM_LIB_ZSTD=On ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # almalinux/8.arm64v8 base image # hadolint ignore=DL3029 @@ -121,7 +136,8 @@ RUN yum -y update && \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ - libzstd-devel && \ + libzstd-devel \ + librabbitmq-devel && \ yum clean all # Use system installed libzstd library. @@ -130,6 +146,8 @@ ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" ENV FLB_JEMALLOC_OPTIONS=$FLB_JEMALLOC_OPTIONS +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Common build for all distributions now # hadolint ignore=DL3006 @@ -175,6 +193,7 @@ RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ -DFLB_CHUNK_TRACE="${FLB_CHUNK_TRACE}" \ -DFLB_UNICODE_ENCODER="${FLB_UNICODE_ENCODER}" \ -DFLB_PREFER_SYSTEM_LIB_ZSTD="${FLB_PREFER_SYSTEM_LIB_ZSTD}" \ + -DFLB_IN_AMQP="${FLB_IN_AMQP}" \ ../ VOLUME [ "/output" ] diff --git a/packaging/distros/amazonlinux/Dockerfile b/packaging/distros/amazonlinux/Dockerfile index 4d96c2993d2..baff33b9e0e 100644 --- a/packaging/distros/amazonlinux/Dockerfile +++ b/packaging/distros/amazonlinux/Dockerfile @@ -24,6 +24,7 @@ RUN yum -y update && \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel \ postgresql-devel postgresql-libs glibc-devel \ libyaml-devel zlib-devel libcurl-devel pkgconf-pkg-config \ + librabbitmq-devel \ tar gzip && \ yum clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -49,6 +50,7 @@ RUN yum -y update && \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel \ postgresql-devel postgresql-libs \ libyaml-devel zlib-devel libcurl-devel pkgconf-pkg-config \ + librabbitmq-devel \ tar gzip && \ yum clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -60,6 +62,8 @@ RUN yum -y update && \ ENV PATH="${CMAKE_HOME}/bin:${PATH}" ARG FLB_KAFKA=Off ENV FLB_KAFKA=$FLB_KAFKA +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP FROM amazonlinux:2023 AS amazonlinux-2023-base @@ -74,6 +78,7 @@ RUN yum -y update && \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel \ postgresql-devel postgresql-libs \ libyaml-devel zlib-devel libcurl-devel pkgconf-pkg-config \ + librabbitmq-devel \ tar gzip && \ yum clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -84,6 +89,8 @@ RUN yum -y update && \ ENV PATH="${CMAKE_HOME}/bin:${PATH}" ARG FLB_KAFKA=On ENV FLB_KAFKA=$FLB_KAFKA +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # hadolint ignore=DL3029 FROM --platform=arm64 amazonlinux:2023 AS amazonlinux-2023.arm64v8-base @@ -101,6 +108,7 @@ RUN yum -y update && \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel \ postgresql-devel postgresql-libs \ libyaml-devel zlib-devel libcurl-devel pkgconf-pkg-config \ + librabbitmq-devel \ tar gzip && \ yum clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -111,6 +119,8 @@ RUN yum -y update && \ ENV PATH="${CMAKE_HOME}/bin:${PATH}" ARG FLB_KAFKA=On ENV FLB_KAFKA=$FLB_KAFKA +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Common build for all distributions now @@ -152,6 +162,7 @@ RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ -DFLB_NIGHTLY_BUILD="$FLB_NIGHTLY_BUILD" \ -DFLB_JEMALLOC="${FLB_JEMALLOC}" \ -DFLB_CHUNK_TRACE="${FLB_CHUNK_TRACE}" \ + -DFLB_IN_AMQP="${FLB_IN_AMQP}" \ ../ VOLUME [ "/output" ] diff --git a/packaging/distros/centos/Dockerfile b/packaging/distros/centos/Dockerfile index a710abd16ed..90a224db6b9 100644 --- a/packaging/distros/centos/Dockerfile +++ b/packaging/distros/centos/Dockerfile @@ -24,6 +24,7 @@ RUN sed -i -e "s/^mirrorlist=http:\/\/mirrorlist.centos.org/#mirrorlist=http:\/\ wget unzip systemd-devel wget flex bison \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ postgresql-libs postgresql-devel postgresql-server postgresql libyaml-devel \ + librabbitmq-devel \ tar gzip && \ yum install -y epel-release && \ yum install -y cmake3 && \ @@ -45,6 +46,8 @@ ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD ARG FLB_RIPSER=Off ENV FLB_RIPSER=$FLB_RIPSER +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # centos/7.arm64v8 base image FROM arm64v8/centos:7 AS centos-7.arm64v8-base @@ -63,6 +66,7 @@ RUN sed -i -e "s/^mirrorlist=http:\/\/mirrorlist.centos.org/#mirrorlist=http:\/\ wget unzip systemd-devel wget flex bison \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ postgresql-libs postgresql-devel postgresql-server postgresql libyaml-devel \ + librabbitmq-devel \ tar gzip && \ yum install -y epel-release && \ yum install -y cmake3 && \ @@ -85,6 +89,8 @@ ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD ARG FLB_RIPSER=Off ENV FLB_RIPSER=$FLB_RIPSER +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" @@ -111,6 +117,7 @@ RUN yum -y update && \ postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ libyaml-devel zlib-devel \ + librabbitmq-devel \ tar gzip && \ yum clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -130,6 +137,8 @@ ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD ARG FLB_RIPSER=On ENV FLB_RIPSER=$FLB_RIPSER +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # centos/8.arm64v8 base image FROM arm64v8/centos:8 AS centos-8.arm64v8-base @@ -154,6 +163,7 @@ RUN yum -y update && \ postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ libyaml-devel zlib-devel \ + librabbitmq-devel \ tar gzip && \ yum clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -173,6 +183,8 @@ ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD ARG FLB_RIPSER=On ENV FLB_RIPSER=$FLB_RIPSER +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" @@ -192,6 +204,7 @@ RUN dnf -y install 'dnf-command(config-manager)' && dnf -y config-manager --set- postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ libyaml-devel zlib-devel \ + librabbitmq-devel \ tar gzip && \ dnf clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -205,6 +218,8 @@ ARG FLB_OUT_PGSQL=On ENV FLB_OUT_PGSQL=$FLB_OUT_PGSQL ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # hadolint ignore=DL3029 FROM --platform=arm64 quay.io/centos/centos:stream9 AS centos-9.arm64v8-base @@ -223,6 +238,7 @@ RUN dnf -y install 'dnf-command(config-manager)' && dnf -y config-manager --set- postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ libyaml-devel zlib-devel \ + librabbitmq-devel \ tar gzip && \ dnf clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -242,6 +258,8 @@ ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD ARG FLB_RIPSER=On ENV FLB_RIPSER=$FLB_RIPSER +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" @@ -261,6 +279,7 @@ RUN dnf -y install 'dnf-command(config-manager)' && dnf -y config-manager --set- postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ libyaml-devel zlib-devel libzstd-devel \ + librabbitmq-devel \ tar gzip && \ dnf clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -275,6 +294,8 @@ ENV FLB_OUT_PGSQL=$FLB_OUT_PGSQL # Use system installed libzstd library. ARG FLB_PREFER_SYSTEM_LIB_ZSTD=On ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # hadolint ignore=DL3029 FROM --platform=arm64 quay.io/centos/centos:stream10 AS centos-10.arm64v8-base @@ -293,6 +314,7 @@ RUN dnf -y install 'dnf-command(config-manager)' && dnf -y config-manager --set- postgresql-libs postgresql-devel postgresql-server postgresql \ cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libcurl-devel pkgconf-pkg-config \ libyaml-devel zlib-devel libzstd-devel \ + librabbitmq-devel \ tar gzip && \ dnf clean all && \ mkdir -p "${CMAKE_HOME}" && \ @@ -313,6 +335,8 @@ ARG FLB_PREFER_SYSTEM_LIB_ZSTD=On ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD ARG FLB_RIPSER=On ENV FLB_RIPSER=$FLB_RIPSER +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" @@ -360,6 +384,7 @@ RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ -DFLB_UNICODE_ENCODER="${FLB_UNICODE_ENCODER}" \ -DFLB_PREFER_SYSTEM_LIB_ZSTD="${FLB_PREFER_SYSTEM_LIB_ZSTD}" \ -DFLB_RIPSER="${FLB_RIPSER}" \ + -DFLB_IN_AMQP="${FLB_IN_AMQP}" \ ../ VOLUME [ "/output" ] diff --git a/packaging/distros/debian/Dockerfile b/packaging/distros/debian/Dockerfile index 82edf687423..4e28f51b652 100644 --- a/packaging/distros/debian/Dockerfile +++ b/packaging/distros/debian/Dockerfile @@ -29,6 +29,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl1.1 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -37,6 +38,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # debian/buster.arm64v8 base image FROM arm64v8/debian:buster-slim AS debian-buster.arm64v8-base @@ -60,6 +63,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl1.1 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -68,6 +72,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # debian/bullseye base image FROM debian:bullseye-slim AS debian-bullseye-base @@ -84,6 +90,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl1.1 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -92,6 +99,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # debian/bullseye.arm64v8 base image FROM arm64v8/debian:bullseye-slim AS debian-bullseye.arm64v8-base @@ -110,6 +119,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl1.1 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -118,6 +128,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # debian/bookworm base image FROM debian:bookworm-slim AS debian-bookworm-base @@ -134,6 +146,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl3 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -142,6 +155,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # debian/bookworm.arm64v8 base image FROM arm64v8/debian:bookworm-slim AS debian-bookworm.arm64v8-base @@ -160,6 +175,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl3 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -168,6 +184,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # debian/trixie base image FROM debian:trixie-slim AS debian-trixie-base @@ -184,6 +202,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl3 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -192,6 +211,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # debian/trixie.arm64v8 base image FROM arm64v8/debian:trixie-slim AS debian-trixie.arm64v8-base @@ -210,6 +231,7 @@ RUN apt-get -qq update && \ libsystemd-dev zlib1g-dev flex bison \ libssl3 libssl-dev libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -218,6 +240,8 @@ RUN apt-get -qq update && \ curl -jksSL "${cmake_download_url}" | tar -xzf - -C "${CMAKE_HOME}" --strip-components 1 ENV PATH="${CMAKE_HOME}/bin:${PATH}" +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Common build for all distributions now # hadolint ignore=DL3006 @@ -258,6 +282,7 @@ RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ -DFLB_NIGHTLY_BUILD="$FLB_NIGHTLY_BUILD" \ -DFLB_JEMALLOC="${FLB_JEMALLOC}" \ -DFLB_CHUNK_TRACE="${FLB_CHUNK_TRACE}" \ + -DFLB_IN_AMQP="${FLB_IN_AMQP}" \ ../ VOLUME [ "/output" ] diff --git a/packaging/distros/raspbian/Dockerfile b/packaging/distros/raspbian/Dockerfile index 10dd6627258..ec4f6b87c02 100644 --- a/packaging/distros/raspbian/Dockerfile +++ b/packaging/distros/raspbian/Dockerfile @@ -18,7 +18,8 @@ RUN apt-get update && \ make bash sudo wget unzip dh-make \ libsystemd-dev zlib1g-dev flex bison \ libssl1.1 libssl-dev libpq-dev postgresql-server-dev-all \ - libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config && \ + libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev && \ apt-get install -y --reinstall lsb-base lsb-release # raspbian/bullseye base image @@ -31,7 +32,8 @@ RUN apt-get update && \ cmake make bash sudo wget unzip dh-make \ libsystemd-dev zlib1g-dev flex bison \ libssl1.1 libssl-dev libpq-dev postgresql-server-dev-all \ - libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config && \ + libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev && \ apt-get install -y --reinstall lsb-base lsb-release # raspbian/bookworm base image @@ -44,7 +46,8 @@ RUN apt-get update && \ cmake make bash sudo wget unzip dh-make \ libsystemd-dev zlib1g-dev flex bison \ libssl3 libssl-dev libpq-dev postgresql-server-dev-all \ - libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config && \ + libsasl2-2 libsasl2-dev libyaml-dev libcurl4-openssl-dev pkg-config \ + librabbitmq-dev && \ apt-get install -y --reinstall lsb-base lsb-release # Common build for all distributions now @@ -72,6 +75,7 @@ ARG FLB_OUT_KAFKA=On ARG FLB_OUT_PGSQL=On ARG FLB_JEMALLOC=On ARG FLB_CHUNK_TRACE=On +ARG FLB_IN_AMQP=On #Tell raspbian packages should be using armv7. ARG WAMR_BUILD_TARGET=ARMV7A @@ -89,6 +93,7 @@ RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ -DFLB_JEMALLOC="${FLB_JEMALLOC}" \ -DFLB_CHUNK_TRACE="${FLB_CHUNK_TRACE}" \ -DWAMR_BUILD_TARGET="${WAMR_BUILD_TARGET}" \ + -DFLB_IN_AMQP="${FLB_IN_AMQP}" \ ../ VOLUME [ "/output" ] diff --git a/packaging/distros/rockylinux/Dockerfile b/packaging/distros/rockylinux/Dockerfile index 8a30978fbd7..4b688295e1f 100644 --- a/packaging/distros/rockylinux/Dockerfile +++ b/packaging/distros/rockylinux/Dockerfile @@ -20,11 +20,14 @@ RUN yum -y update && \ yum install -y rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # rockylinux/8.arm64v8 base image # hadolint ignore=DL3029 @@ -40,7 +43,8 @@ RUN yum -y update && \ yum install -y rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off @@ -48,6 +52,8 @@ ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" ENV FLB_JEMALLOC_OPTIONS=$FLB_JEMALLOC_OPTIONS +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP FROM rockylinux/rockylinux:9 AS rockylinux-9-base @@ -59,11 +65,14 @@ RUN yum -y update && \ yum install -y --allowerasing rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP # rockylinux/9.arm64v8 base image # hadolint ignore=DL3029 @@ -79,7 +88,8 @@ RUN yum -y update && \ yum install -y --allowerasing rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + librabbitmq-devel && \ yum clean all ARG FLB_PREFER_SYSTEM_LIB_ZSTD=Off @@ -87,6 +97,8 @@ ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" ENV FLB_JEMALLOC_OPTIONS=$FLB_JEMALLOC_OPTIONS +ARG FLB_IN_AMQP=Off +ENV FLB_IN_AMQP=$FLB_IN_AMQP FROM rockylinux/rockylinux:10 AS rockylinux-10-base @@ -98,13 +110,16 @@ RUN yum -y update && \ yum install -y --allowerasing rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ - libzstd-devel && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + libzstd-devel \ + librabbitmq-devel && \ yum clean all # Use system installed libzstd library. ARG FLB_PREFER_SYSTEM_LIB_ZSTD=On ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # rockylinux/10.arm64v8 base image # hadolint ignore=DL3029 @@ -120,8 +135,9 @@ RUN yum -y update && \ yum install -y --allowerasing rpm-build curl ca-certificates gcc gcc-c++ cmake make bash \ wget unzip systemd-devel wget flex bison \ postgresql-libs postgresql-devel postgresql-server postgresql \ - cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ - libzstd-devel && \ + cyrus-sasl-lib cyrus-sasl-devel openssl openssl-libs openssl-devel libyaml-devel pkgconf-pkg-config \ + libzstd-devel \ + librabbitmq-devel && \ yum clean all # Use system installed libzstd library. @@ -130,6 +146,8 @@ ENV FLB_PREFER_SYSTEM_LIB_ZSTD=$FLB_PREFER_SYSTEM_LIB_ZSTD # Need larger page size ARG FLB_JEMALLOC_OPTIONS="--with-lg-page=16 --with-lg-quantum=3" ENV FLB_JEMALLOC_OPTIONS=$FLB_JEMALLOC_OPTIONS +ARG FLB_IN_AMQP=On +ENV FLB_IN_AMQP=$FLB_IN_AMQP # Common build for all distributions now # hadolint ignore=DL3006 @@ -175,6 +193,7 @@ RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ -DFLB_CHUNK_TRACE="${FLB_CHUNK_TRACE}" \ -DFLB_UNICODE_ENCODER="${FLB_UNICODE_ENCODER}" \ -DFLB_PREFER_SYSTEM_LIB_ZSTD="${FLB_PREFER_SYSTEM_LIB_ZSTD}" \ + -DFLB_IN_AMQP="${FLB_IN_AMQP}" \ ../ VOLUME [ "/output" ] diff --git a/packaging/distros/ubuntu/Dockerfile b/packaging/distros/ubuntu/Dockerfile index 42e1c5d2bda..a979bbbe164 100644 --- a/packaging/distros/ubuntu/Dockerfile +++ b/packaging/distros/ubuntu/Dockerfile @@ -28,6 +28,7 @@ RUN apt-get update && \ software-properties-common libyaml-dev apt-transport-https \ pkg-config libsasl2-2 libsasl2-dev openssl libssl-dev libssl1.0 \ libcurl4-openssl-dev zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ wget -q -O - https://apt.kitware.com/keys/kitware-archive-latest.asc 2>/dev/null | \ gpg --dearmor - | tee /etc/apt/trusted.gpg.d/kitware.gpg >/dev/null && \ @@ -58,6 +59,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl1.1 libcurl4-openssl-dev \ software-properties-common libyaml-dev apt-transport-https pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ wget -q -O - https://apt.kitware.com/keys/kitware-archive-latest.asc 2>/dev/null | \ gpg --dearmor - | tee /etc/apt/trusted.gpg.d/kitware.gpg >/dev/null && \ @@ -92,6 +94,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl1.1 libcurl4-openssl-dev \ software-properties-common libyaml-dev apt-transport-https pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ wget -q -O - https://apt.kitware.com/keys/kitware-archive-latest.asc 2>/dev/null | \ gpg --dearmor - | tee /etc/apt/trusted.gpg.d/kitware.gpg >/dev/null && \ @@ -122,6 +125,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl1.1 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -148,6 +152,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl1.1 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -172,6 +177,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all libpq5 \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl3 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -198,6 +204,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all libpq5 \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl3 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -222,6 +229,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all libpq5 \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl3 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -248,6 +256,7 @@ RUN apt-get update && \ libpq-dev postgresql-server-dev-all libpq5 \ libsasl2-2 libsasl2-dev openssl libssl-dev libssl3 libcurl4-openssl-dev \ libyaml-dev pkg-config zlib1g-dev \ + librabbitmq-dev \ tar gzip && \ apt-get install -y --reinstall lsb-base lsb-release && \ mkdir -p "${CMAKE_HOME}" && \ @@ -282,6 +291,7 @@ ARG FLB_OUT_KAFKA=On ARG FLB_OUT_PGSQL=On ARG FLB_JEMALLOC=On ARG FLB_CHUNK_TRACE=On +ARG FLB_IN_AMQP=On ENV CFLAGS=$CFLAGS RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ @@ -296,6 +306,7 @@ RUN cmake -DCMAKE_INSTALL_PREFIX="$CMAKE_INSTALL_PREFIX" \ -DFLB_NIGHTLY_BUILD="$FLB_NIGHTLY_BUILD" \ -DFLB_JEMALLOC="${FLB_JEMALLOC}" \ -DFLB_CHUNK_TRACE="${FLB_CHUNK_TRACE}" \ + -DFLB_IN_AMQP="${FLB_IN_AMQP}" \ ../ VOLUME [ "/output" ] diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index 36fcc94ad12..d5581854b01 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -341,6 +341,7 @@ REGISTER_IN_PLUGIN("in_mqtt") REGISTER_IN_PLUGIN("in_lib") REGISTER_IN_PLUGIN("in_forward") REGISTER_IN_PLUGIN("in_random") +REGISTER_IN_PLUGIN("in_amqp") # PROCESSORS # ========== diff --git a/plugins/in_amqp/CMakeLists.txt b/plugins/in_amqp/CMakeLists.txt new file mode 100644 index 00000000000..5fbc9f8caa0 --- /dev/null +++ b/plugins/in_amqp/CMakeLists.txt @@ -0,0 +1,4 @@ +set(src + in_amqp.c) + +FLB_PLUGIN(in_amqp "${src}" rabbitmq::rabbitmq) diff --git a/plugins/in_amqp/in_amqp.c b/plugins/in_amqp/in_amqp.c new file mode 100644 index 00000000000..48a85a6be4a --- /dev/null +++ b/plugins/in_amqp/in_amqp.c @@ -0,0 +1,695 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2024 The Fluent Bit Authors + * Copyright (C) 2026 Matwey V. Kornilov + * + * 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. + */ + + +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + + +#include "in_amqp.h" +#include "fluent-bit/flb_log_event_encoder.h" + +static void in_amqp_log_reply_error(struct flb_input_instance *in, amqp_rpc_reply_t x, char const *context) { + switch (x.reply_type) { + case AMQP_RESPONSE_NORMAL: return; + + case AMQP_RESPONSE_NONE: { + flb_plg_error(in, "%s: missing RPC reply type", context); + break; + } + case AMQP_RESPONSE_LIBRARY_EXCEPTION: { + flb_plg_error(in, "%s: %s", context, amqp_error_string2(x.library_error)); + break; + } + case AMQP_RESPONSE_SERVER_EXCEPTION: { + switch (x.reply.id) { + case AMQP_CONNECTION_CLOSE_METHOD: { + amqp_connection_close_t *m = (amqp_connection_close_t *)x.reply.decoded; + + flb_plg_error(in, "%s: server connection error %hu, message: %.*s", context, m->reply_code, (int)m->reply_text.len, (char *)m->reply_text.bytes); + break; + } + case AMQP_CHANNEL_CLOSE_METHOD: { + amqp_channel_close_t *m = (amqp_channel_close_t *)x.reply.decoded; + + flb_plg_error(in, "%s: server channel error %hu, message: %.*s", context, m->reply_code, (int)m->reply_text.len, (char *)m->reply_text.bytes); + break; + } + default: + flb_plg_error(in, "%s: unknown server error, method id 0x%08X", context, x.reply.id); + break; + } + break; + } + } +} + +static int in_amqp_append_metadata_bytes(struct flb_log_event_encoder* encoder, const char* key, amqp_bytes_t bytes) { + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_CSTRING_VALUE(key), + FLB_LOG_EVENT_STRING_VALUE(bytes.bytes, bytes.len)); +} + +static int in_amqp_append_metadata_entry(struct flb_log_event_encoder* encoder, struct amqp_table_entry_t_* entry) { + struct flb_time out_time; + + amqp_bytes_t* key = &entry->key; + amqp_field_value_t* value = &entry->value; + + switch (value->kind) { + case AMQP_FIELD_KIND_BOOLEAN: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_BOOLEAN_VALUE(value->value.boolean)); + case AMQP_FIELD_KIND_I8: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_INT8_VALUE(value->value.i8)); + case AMQP_FIELD_KIND_U8: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_UINT8_VALUE(value->value.u8)); + case AMQP_FIELD_KIND_I16: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_INT16_VALUE(value->value.i16)); + case AMQP_FIELD_KIND_U16: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_UINT16_VALUE(value->value.u16)); + case AMQP_FIELD_KIND_I32: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_INT32_VALUE(value->value.i32)); + case AMQP_FIELD_KIND_U32: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_UINT32_VALUE(value->value.u32)); + case AMQP_FIELD_KIND_I64: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_INT64_VALUE(value->value.i64)); + case AMQP_FIELD_KIND_U64: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_UINT64_VALUE(value->value.u64)); + case AMQP_FIELD_KIND_F32: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_DOUBLE_VALUE(value->value.f32)); + case AMQP_FIELD_KIND_F64: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_DOUBLE_VALUE(value->value.f64)); + case AMQP_FIELD_KIND_UTF8: + /* fallthrough */ + case AMQP_FIELD_KIND_BYTES: + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_STRING_VALUE(value->value.bytes.bytes, value->value.bytes.len)); + case AMQP_FIELD_KIND_TIMESTAMP: + flb_time_set(&out_time, value->value.u64, 0); + return flb_log_event_encoder_append_metadata_values(encoder, + FLB_LOG_EVENT_STRING_VALUE(key->bytes, key->len), + FLB_LOG_EVENT_TIMESTAMP_VALUE(&out_time)); + default: /* Unsupported */ + break; + }; + + return FLB_EVENT_ENCODER_SUCCESS; +} + +static int in_amqp_handle_envelope(struct flb_amqp *ctx, amqp_envelope_t *envelope) { + amqp_message_t* message = &envelope->message; + amqp_bytes_t* body = &message->body; + amqp_basic_properties_t* properties = &message->properties; + amqp_table_t* headers = &properties->headers; + + int ret, i; + void *out_buf = NULL; + size_t out_size = 0; + struct flb_time out_time; + + flb_time_zero(&out_time); + + if (ctx->parser) { + ret = flb_parser_do(ctx->parser, body->bytes, body->len, &out_buf, &out_size, &out_time); + + if (ret < 0) { + flb_plg_trace(ctx->ins, "tried to parse '%.*s'", (int)body->len, (char*)body->bytes); + flb_plg_trace(ctx->ins, "buf_size %zu", body->len); + flb_plg_error(ctx->ins, "parser returned an error"); + + return ret; + } + } + + if (flb_time_to_nanosec(&out_time) == 0L && (properties->_flags & AMQP_BASIC_TIMESTAMP_FLAG)) { + flb_time_set(&out_time, properties->timestamp, 0); + } + + if (flb_time_to_nanosec(&out_time) == 0L) { + flb_time_get(&out_time); + } + + ret = flb_log_event_encoder_begin_record(&ctx->encoder); + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_set_timestamp(&ctx->encoder, &out_time); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = in_amqp_append_metadata_bytes(&ctx->encoder, "exchange", envelope->exchange); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = in_amqp_append_metadata_bytes(&ctx->encoder, "routing_key", envelope->routing_key); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS && (properties->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG)) { + ret = in_amqp_append_metadata_bytes(&ctx->encoder, "content_type", properties->content_type); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS && (properties->_flags & AMQP_BASIC_CONTENT_ENCODING_FLAG)) { + ret = in_amqp_append_metadata_bytes(&ctx->encoder, "content_encoding", properties->content_encoding); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS && (properties->_flags & AMQP_BASIC_CORRELATION_ID_FLAG)) { + ret = in_amqp_append_metadata_bytes(&ctx->encoder, "correlation_id", properties->correlation_id); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS && (properties->_flags & AMQP_BASIC_REPLY_TO_FLAG)) { + ret = in_amqp_append_metadata_bytes(&ctx->encoder, "reply_to", properties->reply_to); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS && (properties->_flags & AMQP_BASIC_HEADERS_FLAG)) { + ret = flb_log_event_encoder_append_metadata_cstring(&ctx->encoder, "headers"); + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_metadata_begin_map((&ctx->encoder)); + } + + for (i = 0; i < headers->num_entries && ret == FLB_EVENT_ENCODER_SUCCESS; ++i) { + ret = in_amqp_append_metadata_entry(&ctx->encoder, &headers->entries[i]); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_metadata_commit_map((&ctx->encoder)); + } + } + + if (ctx->parser) { + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_set_body_from_raw_msgpack( + &ctx->encoder, + out_buf, + out_size); + } + } else { + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_append_body_cstring( + &ctx->encoder, "amqp"); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_append_body_string( + &ctx->encoder, + body->bytes, + body->len); + } + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_commit_record(&ctx->encoder); + } + + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + flb_input_log_append(ctx->ins, NULL, 0, + ctx->encoder.output_buffer, + ctx->encoder.output_length); + + } else { + flb_plg_error(ctx->ins, "Error encoding record : %d", ret); + } + + flb_log_event_encoder_reset(&ctx->encoder); + + flb_free(out_buf); + + return ret; +} + +static int in_amqp_consumer_start(struct flb_amqp *ctx, struct flb_config *config); + +static void in_amqp_connection_destroy(struct flb_amqp *ctx); + +static int in_amqp_collect(struct flb_input_instance *in, + struct flb_config *config, + void *in_context) +{ + const struct timeval tv = {.tv_sec = 0, .tv_usec = 0}; + + struct flb_amqp* ctx = in_context; + struct flb_amqp_connection* c = &ctx->conn; + + amqp_frame_t frame; + amqp_rpc_reply_t reply; + amqp_envelope_t envelope; + int ret; + + for (;;) { + amqp_maybe_release_buffers(c->conn); + reply = amqp_consume_message(c->conn, &envelope, &tv, 0); + + if (reply.reply_type == AMQP_RESPONSE_NORMAL) { + in_amqp_handle_envelope(ctx, &envelope); + + amqp_destroy_envelope(&envelope); + + /* Proceed to the next message */ + continue; + } else if (reply.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && + reply.library_error == AMQP_STATUS_TIMEOUT) { + /* All messages have been processed */ + return 0; + } else if (reply.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && + reply.library_error == AMQP_STATUS_UNEXPECTED_STATE) { + /* + * If ret.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION, and + * ret.library_error == AMQP_STATUS_UNEXPECTED_STATE, a frame + * other than AMQP_BASIC_DELIVER_METHOD was received, the caller + * should call amqp_simple_wait_frame() to read this frame and + * take appropriate action. + */ + ret = amqp_simple_wait_frame(c->conn, &frame); + if (ret != AMQP_STATUS_OK) { + flb_plg_error(in, "An error occurred during waiting frame: %s", amqp_error_string2(ret)); + } else if (frame.frame_type == AMQP_FRAME_METHOD) { + switch (frame.payload.method.id) { + case AMQP_CHANNEL_CLOSE_METHOD: { + amqp_channel_close_t *m = (amqp_channel_close_t *)frame.payload.method.decoded; + flb_plg_warn(in, "AMQP server channel error %hu, message: %.*s", m->reply_code, (int)m->reply_text.len, (char *)m->reply_text.bytes); + break; + } + case AMQP_CONNECTION_CLOSE_METHOD: { + amqp_connection_close_t *m = (amqp_connection_close_t *)frame.payload.method.decoded; + flb_plg_warn(in, "AMQP server connection error %hu, message: %.*s", m->reply_code, (int)m->reply_text.len, (char *)m->reply_text.bytes); + break; + } + default: + flb_plg_warn(in, "An unexpected AMQP method id 0x%08X", frame.payload.method.id); + } + + /* Out of bound frame is not an error */ + continue; + } + } + + in_amqp_log_reply_error(in, reply, "An error occurred during consuming message"); + + in_amqp_connection_destroy(ctx); + + if (in_amqp_consumer_start(ctx, config) < 0) { + return -1; + } else if (c->conn == NULL) { + return 0; + } + } + + return 0; +} + +static void in_amqp_connection_destroy(struct flb_amqp *ctx) +{ + struct flb_amqp_connection* c = &ctx->conn; + + if (c->conn) { + /* Attached socket FD will be invalidated */ + if (c->coll_id >= 0) { + flb_input_collector_delete(c->coll_id, ctx->ins); + c->coll_id = -1; + } + + amqp_destroy_connection(c->conn); + c->conn = NULL; + } +} + +static int in_amqp_connection_init(struct flb_amqp *ctx, struct flb_config *config) +{ + struct flb_amqp_connection* c = &ctx->conn; + int ret; + amqp_rpc_reply_t reply; + amqp_bytes_t queue_bytes; + + c->conn = amqp_new_connection(); + if (c->conn == NULL) { + flb_plg_error(ctx->ins, "Cannot create AMQP connection"); + + return -1; + } + + c->sock = ctx->conn_info.ssl ? amqp_ssl_socket_new(c->conn) : amqp_tcp_socket_new(c->conn); + if (c->sock == NULL) { + flb_plg_error(ctx->ins, "Cannot create AMQP socket"); + goto error; + } + + ret = amqp_socket_open(c->sock, ctx->conn_info.host, ctx->conn_info.port); + if (ret != AMQP_STATUS_OK) { + flb_plg_error(ctx->ins, "Cannot open AMQP socket: %s", amqp_error_string2(ret)); + goto error; + } + + reply = amqp_login(c->conn, ctx->conn_info.vhost, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, ctx->conn_info.user, ctx->conn_info.password); + if (reply.reply_type != AMQP_RESPONSE_NORMAL) { + in_amqp_log_reply_error(ctx->ins, reply, "Cannot login to the broker"); + goto error; + } + + c->chan = 1; + if (amqp_channel_open(c->conn, c->chan) == NULL) { + in_amqp_log_reply_error(ctx->ins, amqp_get_rpc_reply(c->conn), "Cannot open AMQP channel"); + goto error; + } + + queue_bytes.len = flb_sds_len(ctx->queue_name); + queue_bytes.bytes = ctx->queue_name; + if (amqp_basic_consume(c->conn, c->chan, queue_bytes, amqp_empty_bytes, 0, 1, 1, amqp_empty_table) == NULL) { + in_amqp_log_reply_error(ctx->ins, amqp_get_rpc_reply(c->conn), "Cannot consume"); + goto error; + } + + c->coll_id = flb_input_set_collector_socket(ctx->ins, in_amqp_collect, amqp_socket_get_sockfd(c->sock), config); + if (c->coll_id < 0) { + flb_plg_error(ctx->ins, "Could not set collector for AMQP input plugin"); + goto error; + } + + ret = flb_input_collector_start(c->coll_id, ctx->ins); + if (ret < 0) { + flb_plg_error(ctx->ins, "Could not start collector for AMQP input plugin"); + goto collector_start_error; + } + + flb_plg_info(ctx->ins, "Consuming from %.*s queue", (int)queue_bytes.len, (char*)queue_bytes.bytes); + + return 0; + +collector_start_error: + flb_input_collector_delete(c->coll_id, ctx->ins); + c->coll_id = -1; +error: + in_amqp_connection_destroy(ctx); + + return -1; +} + +static int in_amqp_consumer_start(struct flb_amqp *ctx, struct flb_config *config) +{ + struct flb_amqp_connection* c = &ctx->conn; + int ret; + + if (c->conn) { + return 0; // Already connected + } + + ret = in_amqp_connection_init(ctx, config); + if (ret < 0) { + if (++ctx->retry >= ctx->reconnect_retry_limits) { + flb_plg_error(ctx->ins, "Failed to reconnect after %d attempts", ctx->retry); + + flb_input_collector_pause(ctx->retry_coll_id, ctx->ins); + + ctx->retry = 0; + + return -1; + } + + if (!flb_input_collector_running(ctx->retry_coll_id, ctx->ins)) { + flb_input_collector_resume(ctx->retry_coll_id, ctx->ins); + } + + return 0; + } + + ctx->retry = 0; + + flb_input_collector_pause(ctx->retry_coll_id, ctx->ins); + + return 0; +} + +static int in_amqp_config_destroy(struct flb_amqp *ctx) +{ + flb_log_event_encoder_destroy(&ctx->encoder); + if (ctx->retry_coll_id >= 0) { + flb_input_collector_delete(ctx->retry_coll_id, ctx->ins); + } + in_amqp_connection_destroy(ctx); + flb_free(ctx); + + return 0; +} + +static int in_amqp_reconnect(struct flb_input_instance *in, struct flb_config *config, void *in_context) +{ + struct flb_amqp* ctx = in_context; + + if (in_amqp_consumer_start(ctx, config) < 0) { + return -1; + } + + /* Read pending messages which were buffered by rabbitmq-c during + * the connection negotiation. */ + if (ctx->conn.conn) { + in_amqp_collect(in, config, ctx); + } + + return 0; +} + +/* Set plugin configuration */ +static int in_amqp_configure(struct flb_amqp *ctx, + struct flb_input_instance *in, + struct timespec *tm) +{ + int ret = -1; + + ret = flb_input_config_map_set(in, (void *) ctx); + if (ret == -1) { + return -1; + } + + if (ctx->uri) { + ret = amqp_parse_url(ctx->uri, &ctx->conn_info); + if (ret != AMQP_STATUS_OK) { + flb_plg_error(in, "Error while parsing AMQP URI: %s", amqp_error_string2(ret)); + + return -1; + } + } else { + amqp_default_connection_info(&ctx->conn_info); + } + + if (!ctx->queue_name) { + flb_plg_error(in, "AMQP queue name is not provided"); + + return -1; + } + + if (ctx->reconnect_retry_interval < 1) { + flb_plg_error(in, "reconnect.retry_interval must be >= 1"); + + return -1; + } + + return 0; +} + +/* Initialize plugin */ +static int in_amqp_init(struct flb_input_instance *in, + struct flb_config *config, void *data) +{ + int ret = -1; + struct flb_amqp *ctx = NULL; + struct timespec tm; + + /* Allocate space for the configuration */ + ctx = flb_calloc(1, sizeof(struct flb_amqp)); + if (ctx == NULL) { + return -1; + } + + ctx->parser = NULL; + ctx->ins = in; + ctx->retry_coll_id = -1; + ctx->retry = 0; + ctx->conn.conn = NULL; + ctx->conn.sock = NULL; + ctx->conn.chan = 0; + ctx->conn.coll_id = -1; + + /* Initialize head config */ + ret = in_amqp_configure(ctx, in, &tm); + if (ret < 0) { + in_amqp_config_destroy(ctx); + return -1; + } + + ret = flb_log_event_encoder_init(&ctx->encoder, FLB_LOG_EVENT_FORMAT_DEFAULT); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + flb_plg_error(in, "could not initialize event encoder"); + in_amqp_config_destroy(ctx); + + return -1; + } + + if (ctx->parser_name) { + ctx->parser = flb_parser_get(ctx->parser_name, config); + if (ctx->parser == NULL) { + flb_plg_error(in, "Requested parser '%s' not found", ctx->parser_name); + in_amqp_config_destroy(ctx); + + return -1; + } + } + + flb_input_set_context(in, ctx); + + ctx->retry_coll_id = flb_input_set_collector_time(in, &in_amqp_reconnect, ctx->reconnect_retry_interval, 0, config); + if (ctx->retry_coll_id < 0) { + flb_plg_error(in, "Cannot create reconnection collector"); + in_amqp_config_destroy(ctx); + + return -1; + } + + flb_input_collector_pause(ctx->retry_coll_id, in); + + ret = in_amqp_consumer_start(ctx, config); + if (ret < 0) { + flb_plg_error(in, "Cannot start AMQP consumer"); + in_amqp_config_destroy(ctx); + + return -1; + } + + /* Read pending messages which were buffered by rabbitmq-c during + * the connection negotiation. */ + if (ctx->conn.conn) { + in_amqp_collect(in, config, ctx); + } + + return 0; +} + +static void in_amqp_pause(void *data, struct flb_config *config) +{ + struct flb_amqp *ctx = data; + + if (ctx->conn.conn) { + flb_input_collector_pause(ctx->conn.coll_id, ctx->ins); + } else { + flb_input_collector_pause(ctx->retry_coll_id, ctx->ins); + } +} + +static void in_amqp_resume(void *data, struct flb_config *config) +{ + struct flb_amqp *ctx = data; + + if (ctx->conn.conn) { + flb_input_collector_resume(ctx->conn.coll_id, ctx->ins); + } else { + flb_input_collector_resume(ctx->retry_coll_id, ctx->ins); + } +} + +static int in_amqp_exit(void *data, struct flb_config *config) +{ + (void)config; + struct flb_amqp *ctx = data; + + if (ctx) { + in_amqp_config_destroy(ctx); + } + + return 0; +} + +/* Configuration properties map */ +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_STR, "uri", "amqp://", + 0, FLB_TRUE, offsetof(struct flb_amqp, uri), + "Specify an AMQP URI to connect the broker" + }, + { + FLB_CONFIG_MAP_STR, "queue", NULL, + 0, FLB_TRUE, offsetof(struct flb_amqp, queue_name), + "Specify an AMQP queue name to consume from" + }, + { + FLB_CONFIG_MAP_STR, "parser", NULL, + 0, FLB_TRUE, offsetof(struct flb_amqp, parser_name), + "Set a parser" + }, + { + FLB_CONFIG_MAP_INT, "reconnect.retry_limits", "5", + 0, FLB_TRUE, offsetof(struct flb_amqp, reconnect_retry_limits), + "Maximum number to retry to connect the broker" + }, + { + FLB_CONFIG_MAP_INT, "reconnect.retry_interval", "60", + 0, FLB_TRUE, offsetof(struct flb_amqp, reconnect_retry_interval), + "Retry interval to connect the broker" + }, + {0}, +}; + + +struct flb_input_plugin in_amqp_plugin = { + .name = "amqp", + .description = "AMQP input plugin", + .cb_init = in_amqp_init, + .cb_pre_run = NULL, + .cb_collect = in_amqp_collect, + .cb_flush_buf = NULL, + .config_map = config_map, + .cb_pause = in_amqp_pause, + .cb_resume = in_amqp_resume, + .cb_exit = in_amqp_exit +}; diff --git a/plugins/in_amqp/in_amqp.h b/plugins/in_amqp/in_amqp.h new file mode 100644 index 00000000000..eec2551f53d --- /dev/null +++ b/plugins/in_amqp/in_amqp.h @@ -0,0 +1,53 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2024 The Fluent Bit Authors + * Copyright (C) 2026 Matwey V. Kornilov + * + * 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_IN_AMQP_H +#define FLB_IN_AMQP_H + +#include + +#include +#include + +struct flb_amqp_connection { + amqp_connection_state_t conn; + amqp_socket_t *sock; + amqp_channel_t chan; + int coll_id; +}; + +struct flb_amqp { + flb_sds_t uri; + flb_sds_t queue_name; + flb_sds_t parser_name; + struct amqp_connection_info conn_info; + int reconnect_retry_limits; + int reconnect_retry_interval; + + struct flb_log_event_encoder encoder; + struct flb_parser *parser; + struct flb_input_instance *ins; + int retry_coll_id; + int retry; + + struct flb_amqp_connection conn; +}; + +#endif