Add JSON logging to Fluentd TCP socket for Bedrock

BedrockServer.cpp

@@ -1299,6 +1299,19 @@ BedrockServer::BedrockServer(const SData& args_)
         SSyslogFunc = &SSyslogSocketDirect;
     }
 
+    // Log destination: rsyslog, fluentd, or both. Default is rsyslog.
+    // Fluentd defaults: 127.0.0.1:24224, tag=bedrock. Override with -fluentdHost, -fluentdPort, -fluentdTag
+    string logDestination = args.isSet("-logDestination") ? args["-logDestination"] : "rsyslog";
+    if (logDestination == "fluentd" || logDestination == "both") {
+        string host = args.isSet("-fluentdHost") ? args["-fluentdHost"] : "127.0.0.1";
+        int port = args.isSet("-fluentdPort") ? SToInt(args["-fluentdPort"]) : 24224;
+        string tag = args.isSet("-fluentdTag") ? args["-fluentdTag"] : "bedrock";
+        SFluentdInitialize(host, port, tag);
+    }
+    if (logDestination == "fluentd") {
+        SSyslogFunc = &SSyslogNoop;
+    }
+
     // Check for commands that will be forced to use QUORUM write consistency.
     if (args.isSet("-synchronousCommands")) {
         list<string> syncCommands;

libstuff/libstuff.cpp

@@ -280,6 +280,87 @@ void SSyslogSocketDirect(int priority, const char* format, ...)
     }
 }
 
+void SSyslogNoop(int priority, const char* format, ...)
+{
+}
+
+static int SFluentdSocketFD = -1;
+static mutex SFluentdSocketMutex;
+static string SFluentdHost;
+static int SFluentdPort = 0;
+static string SFluentdTag;
+static atomic<bool> SFluentdConfigured{false};
+
+// Lock parameter enforces mutex is held before calling this function.
+static bool SFluentdConnect(const lock_guard<mutex>&)
+{
+    SFluentdSocketFD = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+    if (SFluentdSocketFD == -1) {
+        return false;
+    }
+
+    struct sockaddr_in addr;
+    memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(SFluentdPort);
+    inet_pton(AF_INET, SFluentdHost.c_str(), &addr.sin_addr);
+
+    if (connect(SFluentdSocketFD, (struct sockaddr*) &addr, sizeof(addr)) == -1) {
+        close(SFluentdSocketFD);
+        SFluentdSocketFD = -1;
+        return false;
+    }
+
+    return true;
+}
+
+void SFluentdInitialize(const string& host, int port, const string& tag)
+{
+    lock_guard<mutex> lock(SFluentdSocketMutex);
+    SFluentdHost = host;
+    SFluentdPort = port;
+    SFluentdTag = tag;
+    SFluentdConnect(lock);
+    SFluentdConfigured.store(true);
+}
+
+void SFluentdLog(int priority, const string& message, const STable& params)
+{
+    if (!SFluentdConfigured.load()) {
+        return;
+    }
+
+    // Build JSON before acquiring lock to avoid doing heavy stuff in the critical section
+    STable record;
+    record["timestamp"] = to_string(time(nullptr));
+    record["priority"] = to_string(priority);
+    record["thread_name"] = SThreadLogName;
+    record["thread_prefix"] = SThreadLogPrefix;
+    record["process"] = SProcessName;
+    record["message"] = message;
+
+    for (const auto& [key, value] : params) {
+        record[key] = value;
+    }
+
+    string json = "[\"" + SFluentdTag + "\"," + to_string(time(nullptr)) + "," + SComposeJSONObject(record) + "]\n";
+
+    lock_guard<mutex> lock(SFluentdSocketMutex);
+
+    // Reconnect if needed
+    if (SFluentdSocketFD == -1) {
+        if (!SFluentdConnect(lock)) {
+            return;
+        }
+    }
+
+    // Try to send the log over TCP. Close the socket on failure. It'll try to reconnect on next attempt
+    if (send(SFluentdSocketFD, json.c_str(), json.size(), MSG_NOSIGNAL) == -1) {
+        close(SFluentdSocketFD);
+        SFluentdSocketFD = -1;
+    }
+}
+
 /////////////////////////////////////////////////////////////////////////////
 // Math stuff
 /////////////////////////////////////////////////////////////////////////////

libstuff/libstuff.h

@@ -255,9 +255,23 @@ void SWhitelistLogParams(const set<string>& params);
 // This is a drop-in replacement for syslog that directly logs to `/run/systemd/journal/syslog` bypassing journald.
 void SSyslogSocketDirect(int priority, const char* format, ...);
 
-// Atomic pointer to the syslog function that we'll actually use. Easy to change to `syslog` or `SSyslogSocketDirect`.
+// No-op function to disable rsyslog logging.
+void SSyslogNoop(int priority, const char* format, ...);
+
+// Atomic pointer to the syslog function that we'll actually use.
+// Can be set to `syslog`, `SSyslogSocketDirect`, or `SSyslogNoop`.
 extern atomic<void (*)(int priority, const char* format, ...)> SSyslogFunc;
 
+// --------------------------------------------------------------------------
+// Fluentd JSON logging stuff
+// --------------------------------------------------------------------------
+// Initialize Fluentd TCP socket connection. Call once at startup.
+void SFluentdInitialize(const string& host, int port, const string& tag);
+
+// Log a message to Fluentd in JSON format. Thread-safe.
+// No-op if Fluentd is not initialized. Handles connection failures gracefully.
+void SFluentdLog(int priority, const string& message, const STable& params = {});
+
 string addLogParams(string&& message, const STable& params = {});
 
 // **NOTE: rsyslog default max line size is 8k bytes. We split on 7k byte boundaries in order to fit the syslog line prefix and the expanded \r\n to #015#012
@@ -267,11 +281,13 @@ string addLogParams(string&& message, const STable& params = {});
             if (_g_SLogMask & (1 << (_PRI_))) {                                     \
                 ostringstream __out;                                                \
                 __out << _MSG_;                                                     \
-                const string s = addLogParams(__out.str(), ## __VA_ARGS__);          \
+                const string __rawMsg = __out.str();                                \
+                const string s = addLogParams(string(__rawMsg), ## __VA_ARGS__);    \
                 const string prefix = SWHEREAMI;                                    \
                 for (size_t i = 0; i < s.size(); i += 7168) {                       \
                     (*SSyslogFunc)(_PRI_, "%s", (prefix + s.substr(i, 7168)).c_str()); \
                 }                                                                   \
+                SFluentdLog(_PRI_, prefix + __rawMsg, ## __VA_ARGS__);              \
             }                                                                       \
         } while (false)