Scaffold a fluent-bit plugin

plugins/fluent-bit/CMakeLists.txt

@@ -0,0 +1,32 @@
+cmake_minimum_required(VERSION 3.19...3.24 FATAL_ERROR)
+
+project(
+  fluent-bit
+  DESCRIPTION "fluent-bit plugin for Tenzir"
+  LANGUAGES CXX)
+
+include(CTest)
+
+find_package(Tenzir REQUIRED)
+
+file(GLOB_RECURSE fluentbit_sources CONFIGURE_DEPENDS
+     "${CMAKE_CURRENT_SOURCE_DIR}/src/*.cpp")
+
+TenzirRegisterPlugin(
+  TARGET fluent-bit
+  ENTRYPOINT src/plugin.cpp
+  SOURCES ${fluentbit_sources}
+  INCLUDE_DIRECTORIES include)
+
+find_package(fluentbit QUIET)
+if (NOT fluentbit_FOUND)
+  list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
+  find_package(fluentbit QUIET REQUIRED)
+  list(POP_BACK CMAKE_MODULE_PATH)
+endif ()
+
+get_filename_component(FLUENT_BIT_LOCATION ${FLUENT_BIT_LIB} PATH)
+get_filename_component(FLUENT_BIT_LOCATION ${FLUENT_BIT_LOCATION} PATH)
+get_filename_component(FLUENT_BIT_LOCATION ${FLUENT_BIT_LOCATION} PATH)
+dependency_summary("fluent-bit" ${FLUENT_BIT_LOCATION} "Dependencies")
+target_link_libraries(fluent-bit PUBLIC fluentbit::fluentbit)

plugins/fluent-bit/cmake/Findfluentbit.cmake

@@ -0,0 +1,22 @@
+find_library(
+  FLUENT_BIT_LIB
+  NAMES fluent-bit
+  PATH_SUFFIXES fluent-bit)
+
+find_path(FLUENT_BIT_INCLUDE_DIR fluent-bit.h)
+
+include(FindPackageHandleStandardArgs)
+
+find_package_handle_standard_args(
+  fluentbit REQUIRED_VARS FLUENT_BIT_LIB FLUENT_BIT_INCLUDE_DIR)
+
+mark_as_advanced(fluentbit_FOUND FLUENT_BIT_INCLUDE_DIR FLUENT_BIT_LIB)
+
+if (NOT TARGET fluentbit::fluentbit)
+  add_library(fluentbit::fluentbit UNKNOWN IMPORTED)
+  set_target_properties(
+    fluentbit::fluentbit
+    PROPERTIES IMPORTED_LOCATION "${FLUENT_BIT_LIB}"
+               INTERFACE_INCLUDE_DIRECTORIES "${FLUENT_BIT_INCLUDE_DIR}"
+               INTERFACE_LINK_LIBRARIES "${FLUENT_BIT_LIB}")
+endif ()

plugins/fluent-bit/include/fluent-bit/fluent-bit-minimal.h

@@ -0,0 +1,83 @@
+/*  Minimal Fluent-Bit Header
+ *
+ *  This header exposes only the symbols needed to use the public
+ *  advertised API of libfluent-bit.so
+ *
+ *  This is derived from fluent-bit/flb_lib.h which is
+ *  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 FLUENT_BIT_MINIMAL_H
+#define FLUENT_BIT_MINIMAL_H
+
+#include <stddef.h>
+
+// #include <fluent-bit/flb_macros.h>
+// #include <fluent-bit/flb_config.h>
+#define FLB_EXPORT extern
+
+/* Lib engine status */
+#define FLB_LIB_ERROR -1
+#define FLB_LIB_NONE 0
+#define FLB_LIB_OK 1
+#define FLB_LIB_NO_CONFIG_MAP 2
+
+struct flb_lib_ctx;
+
+/* Used on out_lib to define a callback and further opaque data */
+struct flb_lib_out_cb {
+  int (*cb)(void* record, size_t size, void* data);
+  void* data;
+};
+
+/* For Fluent Bit library callers, we only export the following symbols */
+typedef struct flb_lib_ctx flb_ctx_t;
+
+FLB_EXPORT void flb_init_env();
+FLB_EXPORT flb_ctx_t* flb_create();
+FLB_EXPORT void flb_destroy(flb_ctx_t* ctx);
+FLB_EXPORT int flb_input(flb_ctx_t* ctx, const char* input, void* data);
+FLB_EXPORT int
+flb_output(flb_ctx_t* ctx, const char* output, struct flb_lib_out_cb* cb);
+FLB_EXPORT int flb_filter(flb_ctx_t* ctx, const char* filter, void* data);
+FLB_EXPORT int flb_input_set(flb_ctx_t* ctx, int ffd, ...);
+FLB_EXPORT int
+flb_input_property_check(flb_ctx_t* ctx, int ffd, char* key, char* val);
+FLB_EXPORT int
+flb_output_property_check(flb_ctx_t* ctx, int ffd, char* key, char* val);
+FLB_EXPORT int
+flb_filter_property_check(flb_ctx_t* ctx, int ffd, char* key, char* val);
+FLB_EXPORT int flb_output_set(flb_ctx_t* ctx, int ffd, ...);
+FLB_EXPORT int
+flb_output_set_test(flb_ctx_t* ctx, int ffd, char* test_name,
+                    void (*out_callback)(void*, int, int, void*, size_t, void*),
+                    void* out_callback_data, void* test_ctx);
+FLB_EXPORT int flb_output_set_callback(flb_ctx_t* ctx, int ffd, char* name,
+                                       void (*cb)(char*, void*, void*));
+
+FLB_EXPORT int flb_filter_set(flb_ctx_t* ctx, int ffd, ...);
+FLB_EXPORT int flb_service_set(flb_ctx_t* ctx, ...);
+FLB_EXPORT int flb_lib_free(void* data);
+FLB_EXPORT double flb_time_now();
+
+/* start stop the engine */
+FLB_EXPORT int flb_start(flb_ctx_t* ctx);
+FLB_EXPORT int flb_stop(flb_ctx_t* ctx);
+FLB_EXPORT int flb_loop(flb_ctx_t* ctx);
+
+/* data ingestion for "lib" input instance */
+FLB_EXPORT int
+flb_lib_push(flb_ctx_t* ctx, int ffd, const void* data, size_t len);
+FLB_EXPORT int flb_lib_config_file(flb_ctx_t* ctx, const char* path);
+
+#endif

plugins/fluent-bit/src/plugin.cpp

@@ -0,0 +1,126 @@
+//    _   _____   __________
+//   | | / / _ | / __/_  __/     Visibility
+//   | |/ / __ |_\ \  / /          Across
+//   |___/_/ |_/___/ /_/       Space and Time
+//
+// SPDX-FileCopyrightText: (c) 2023 The VAST Contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+#include <tenzir/argument_parser.hpp>
+#include <tenzir/chunk.hpp>
+#include <tenzir/data.hpp>
+#include <tenzir/error.hpp>
+#include <tenzir/logger.hpp>
+#include <tenzir/plugin.hpp>
+
+#include <fluent-bit/fluent-bit-minimal.h>
+
+using namespace std::chrono_literals;
+
+namespace tenzir::plugins::fluentbit {
+
+namespace {
+
+class fluentbit_loader final : public plugin_loader {
+public:
+  fluentbit_loader() = default;
+
+  auto instantiate(operator_control_plane& ctrl) const
+    -> std::optional<generator<chunk_ptr>> override {
+    auto make = []() -> generator<chunk_ptr> {
+      while (true) {
+        // TODO: Implement
+        co_yield {};
+      }
+    };
+    return make();
+  }
+
+  auto to_string() const -> std::string override {
+    auto result = name();
+    return result;
+  }
+
+  auto name() const -> std::string override {
+    return "fluent-bit";
+  }
+
+  auto default_parser() const -> std::string override {
+    return "json";
+  }
+
+  friend auto inspect(auto& f, fluentbit_loader& x) -> bool {
+    return f.object(x).pretty_name("fluent-bit_loader").fields();
+  }
+};
+
+class fluentbit_saver final : public plugin_saver {
+public:
+  fluentbit_saver() = default;
+
+  auto instantiate(operator_control_plane& ctrl, std::optional<printer_info>)
+    -> caf::expected<std::function<void(chunk_ptr)>> override {
+    return [](chunk_ptr chunk) mutable {
+      if (!chunk || chunk->size() == 0)
+        return;
+      // TODO: Implement
+    };
+  }
+
+  auto name() const -> std::string override {
+    return "fluent-bit";
+  }
+
+  auto default_printer() const -> std::string override {
+    return "json";
+  }
+
+  auto is_joining() const -> bool override {
+    return true;
+  }
+
+  friend auto inspect(auto& f, fluentbit_saver& x) -> bool {
+    return f.object(x).pretty_name("fluent-bit_saver").fields();
+  }
+};
+
+class plugin final : public virtual loader_plugin<fluentbit_loader>,
+                     public virtual saver_plugin<fluentbit_saver> {
+public:
+  auto initialize(const record& config, const record& /* global_config */)
+    -> caf::error override {
+    config_ = config;
+    return caf::none;
+  }
+
+  auto parse_loader(parser_interface& p) const
+    -> std::unique_ptr<plugin_loader> override {
+    auto parser = argument_parser{
+      name(),
+      fmt::format("https://docs.tenzir.com/docs/next/connectors/{}", name())};
+    // TODO: Implement
+    return std::make_unique<fluentbit_loader>();
+  }
+
+  auto parse_saver(parser_interface& p) const
+    -> std::unique_ptr<plugin_saver> override {
+    auto parser = argument_parser{
+      name(),
+      fmt::format("https://docs.tenzir.com/docs/next/connectors/{}", name())};
+    // TODO: Implement
+    return std::make_unique<fluentbit_saver>();
+  }
+
+  auto name() const -> std::string override {
+    return "fluent-bit";
+  }
+
+private:
+  record config_;
+};
+
+} // namespace
+
+} // namespace tenzir::plugins::fluentbit
+
+TENZIR_REGISTER_PLUGIN(tenzir::plugins::fluentbit::plugin)