in_someip: SOME/IP input plugin

CMakeLists.txt

@@ -227,6 +227,7 @@ if(FLB_ALL)
   set(FLB_IN_NGINX_STATUS 1)
   set(FLB_IN_EXEC         1)
   set(FLB_IN_UNIX_SOCKET  1)
+  set(FLB_IN_SOMEIP       1)
 
   # Output plugins
   set(FLB_OUT_ES          1)
@@ -617,6 +618,11 @@ if(FLB_WASM)
   endif ()
 endif()
 
+# SOME/IP
+if(FLB_IN_SOMEIP)
+  add_subdirectory(${FLB_PATH_LIB_SOMEIP_C} EXCLUDE_FROM_ALL)
+endif()
+
 # AWS
 if (FLB_AWS)
   FLB_DEFINITION(FLB_HAVE_AWS)

cmake/libraries.cmake

@@ -24,3 +24,4 @@ set(FLB_PATH_LIB_SNAPPY      "lib/snappy-fef67ac")
 set(FLB_PATH_LIB_RDKAFKA     "lib/librdkafka-2.4.0")
 set(FLB_PATH_LIB_RING_BUFFER "lib/lwrb")
 set(FLB_PATH_LIB_WASM_MICRO_RUNTIME  "lib/wasm-micro-runtime-WAMR-1.3.0")
+set(FLB_PATH_LIB_SOMEIP_C    "lib/libsomeip-c")

cmake/plugins_options.cmake

@@ -47,6 +47,7 @@ DEFINE_OPTION(FLB_IN_PROMETHEUS_REMOTE_WRITE  "Enable prometheus remote write in
 DEFINE_OPTION(FLB_IN_PROMETHEUS_SCRAPE        "Enable Prometheus Scrape input plugin"        ON)
 DEFINE_OPTION(FLB_IN_RANDOM                   "Enable random input plugin"                   ON)
 DEFINE_OPTION(FLB_IN_SERIAL                   "Enable Serial input plugin"                   ON)
+DEFINE_OPTION(FLB_IN_SOMEIP                   "Enable SOME/IP input plugin"                  OFF)
 DEFINE_OPTION(FLB_IN_SPLUNK                   "Enable Splunk HTTP HEC input plugin"          ON)
 DEFINE_OPTION(FLB_IN_STATSD                   "Enable StatsD input plugin"                   ON)
 DEFINE_OPTION(FLB_IN_STDIN                    "Enable Standard input plugin"                 ON)

conf/in_someip.conf

@@ -0,0 +1,54 @@
+[SERVICE]
+    # Flush
+    # =====
+    # Set an interval of seconds before to flush records to a destination
+    Flush        5
+
+    # Daemon
+    # ======
+    # Instruct Fluent Bit to run in foreground or background mode.
+    Daemon       Off
+
+    # Log_Level
+    # =========
+    # Set the verbosity level of the service, values can be:
+    #
+    # - error
+    # - warning
+    # - info
+    # - debug
+    # - trace
+    #
+    # By default 'info' is set, that means it includes 'error' and 'warning'.
+    Log_Level    trace
+
+    # HTTP Monitoring Server
+    # ======================
+    #
+    # HTTP_Monitor: enable/disable the HTTP Server to monitor
+    #               Fluent Bit internals.
+    # HTTP_Port   : specify the TCP port of the HTTP Server
+    HTTP_Monitor Off
+    HTTP_Port    2020
+
+[INPUT]
+    Name someip
+    Tag  in.someip
+
+    # Events to subscribe to.
+    # Each event should have form:
+    #    Event <service id>,<instance id>,<event id>,<event group 1>,...
+    #
+    # Each event must have at least one event group
+    Event 4,1,32768,1
+    Event 4,1,32769,2
+
+    # RPC to send on startup
+    # Each RPC entry should have form:
+    #    RPC <service id>,<instance id>,<method id>,<Request Payload>
+    #
+    # Request payload should be base64 encoded
+    RPC   4,1,1,CgAQAw==
+[OUTPUT]
+    Name  stdout
+    Match *

dockerfiles/Dockerfile

@@ -55,6 +55,9 @@ RUN echo "deb http://deb.debian.org/debian bookworm-backports main" >> /etc/apt/
     flex \
     bison \
     libyaml-dev \
+    libboost-system-dev \
+    libboost-thread-dev \
+    libboost-filesystem-dev \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/*
 
@@ -141,6 +144,9 @@ RUN echo "deb http://deb.debian.org/debian bookworm-backports main" >> /etc/apt/
     liblzma5 \
     libyaml-0-2 \
     libcap2 \
+    libboost-system-dev \
+    libboost-thread-dev \
+    libboost-filesystem-dev \
     && \
     mkdir -p /dpkg/var/lib/dpkg/status.d/ && \
     for deb in *.deb; do \
@@ -221,6 +227,9 @@ RUN echo "deb http://deb.debian.org/debian bookworm-backports main" >> /etc/apt/
     libatomic1 \
     libgcrypt20 \
     libyaml-0-2 \
+    libboost-system-dev \
+    libboost-thread-dev \
+    libboost-filesystem-dev \
     bash gdb valgrind build-essential  \
     git bash-completion vim tmux jq \
     dnsutils iputils-ping iputils-arping iputils-tracepath iputils-clockdiff \

lib/libsomeip-c/CMakeLists.txt

@@ -0,0 +1,18 @@
+project(someipc CXX)
+
+include(FetchContent)
+FetchContent_Declare(
+    vsomeip3
+    GIT_REPOSITORY https://github.com/COVESA/vsomeip
+    GIT_TAG        0b83e24d16e1611958194e9b727136522f46556b # 3.5.1
+)
+FetchContent_MakeAvailable(vsomeip3)
+
+find_package(vsomeip3 REQUIRED)
+
+add_library(someip-c SHARED src/someip_wrapper.cc)
+target_include_directories(someip-c PUBLIC include)
+target_link_libraries(someip-c PRIVATE vsomeip3 vsomeip3-sd vsomeip3-cfg vsomeip3-e2e)
+target_link_options(someip-c PUBLIC "-Wl,--disable-new-dtags")
+
+add_subdirectory(example)

lib/libsomeip-c/example/CMakeLists.txt

@@ -0,0 +1,3 @@
+
+add_executable(someip_test_service test_service.cc)
+target_link_libraries(someip_test_service PRIVATE someip-c)

lib/libsomeip-c/example/test_service.cc

@@ -0,0 +1,138 @@
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2024 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.
+ */
+#include "someip_api.h"
+#include <iostream>
+#include <cstdlib>
+#include <cstring>
+#include <chrono>
+#include <sstream>
+#include <thread>
+
+/* Class declaration */
+class TestService {
+public:
+    bool Initialize();
+    void Teardown();
+    void HandleRequest(const struct some_ip_request* request_ptr);
+    void SendEvent(const int num);
+private:
+    uint16_t client_id_{0};
+};
+
+namespace {
+    constexpr auto NAME               = "Test Service";
+    constexpr uint16_t SERVICE_ID     = 4;
+    constexpr uint16_t INSTANCE_ID    = 1;
+    constexpr uint16_t METHOD_ID      = 1;
+    constexpr uint16_t EVENT_ID       = 0x8000U;
+    constexpr uint16_t EVENT_GROUP_ID = 1;
+
+    void RequestCallback(void* cookie, struct some_ip_request* request_ptr) {
+        if (cookie == nullptr) {
+            return;
+        }
+        auto service_pointer{static_cast<TestService*>(cookie)};
+        service_pointer->HandleRequest(request_ptr);
+    }
+}
+
+bool TestService::Initialize() {
+    auto ret = someip_initialize(NAME, &client_id_);
+    if (ret != SOMEIP_RET_SUCCESS) {
+        std::cout << "Failed to initialize SOME/IP: " << ret << std::endl;
+        return false;
+    }
+
+    /* Register Request Handler */
+    auto request_handler{[this](struct some_ip_request* request_ptr) {
+        HandleRequest(request_ptr);
+    }};
+    ret = someip_register_request_handler(client_id_, SERVICE_ID, INSTANCE_ID,
+                                          METHOD_ID, this, RequestCallback);
+
+    if (ret != SOMEIP_RET_SUCCESS) {
+        std::cout << "Failed to register request handler: " << ret << std::endl;
+        someip_shutdown(client_id_);
+        return false;
+    }
+
+    /* Offer Event */
+    ret = someip_offer_event(client_id_, SERVICE_ID, INSTANCE_ID, EVENT_ID, const_cast<uint16_t*>(&EVENT_GROUP_ID), 1);
+    if (ret != SOMEIP_RET_SUCCESS) {
+        std::cout << "Failed to Offer Event: " << ret << std::endl;
+        someip_shutdown(client_id_);
+        return false;
+    }
+
+    /* Offer Service */
+    ret = someip_offer_service(client_id_, SERVICE_ID, INSTANCE_ID);
+    if (ret != SOMEIP_RET_SUCCESS) {
+        std::cout << "Failed to Offer Service: " << ret << std::endl;
+        someip_shutdown(client_id_);
+        return false;
+    }
+
+    return true;
+}
+
+void TestService::Teardown() {
+    someip_shutdown(client_id_);
+}
+
+void TestService::HandleRequest(const struct some_ip_request* request_ptr) {
+    if (request_ptr == nullptr) {
+        return;
+    }
+    std::cout << "Received request (method = " << request_ptr->method_id << ")" << std::endl;
+    std::cout << "Payload length = " << request_ptr->payload_len << std::endl;
+
+    /* Normal service would Parse the request and perform/initiate some actions on it*/
+    /* For this example just send back a canned response */
+    auto response{"This is the response to the request"};
+    const auto ret = someip_send_response(client_id_, request_ptr->request_id.client_request_id,
+                                          const_cast<char*>(response), strlen(response));
+    if (ret != SOMEIP_RET_SUCCESS) {
+        std::cout << "Failed to send response: %d" << ret << std::endl;
+    }
+}
+
+void TestService::SendEvent(const int num) {
+    std::stringstream ss;
+    ss << "Event Number " << num;
+    const auto message = ss.str();
+
+    auto ret = someip_send_event(client_id_, SERVICE_ID, INSTANCE_ID, EVENT_ID,
+    message.data(), message.size());
+}
+
+
+int main() {
+    TestService service;
+    if (!service.Initialize()) {
+        return EXIT_FAILURE;
+    }
+
+    auto num_events{10};
+
+    for (auto i = 0; i < num_events; ++i) {
+        service.SendEvent(i);
+        std::this_thread::sleep_for(std::chrono::seconds(2));
+    }
+
+    service.Teardown();
+    return EXIT_SUCCESS;
+}