aboutsummaryrefslogtreecommitdiff
path: root/src/zenutil
diff options
context:
space:
mode:
authorStefan Boberg <[email protected]>2026-03-21 21:43:22 +0100
committerGitHub Enterprise <[email protected]>2026-03-21 21:43:22 +0100
commit14ca5b35d0fc477ba30f10b80f937b523fd7e930 (patch)
tree8aab2acfec8be1af4bf0dffdb4badc3b64bf8385 /src/zenutil
parentfix null stats provider crash when build store is not configured (#875) (diff)
downloadzen-14ca5b35d0fc477ba30f10b80f937b523fd7e930.tar.xz
zen-14ca5b35d0fc477ba30f10b80f937b523fd7e930.zip
Interprocess pipe support (for stdout/stderr capture) (#866)
- **RAII pipe handles for child process stdout/stderr capture**: `StdoutPipeHandles` is now a proper RAII type with automatic cleanup, move semantics, and partial close support. This makes it safe to use pipes for capturing child process output without risking handle/fd leaks. - **Optional separate stderr pipe**: `CreateProcOptions` now accepts a `StderrPipe` field so callers can capture stdout and stderr independently. When null (default), stderr shares the stdout pipe as before. - **LogStreamListener with pluggable handler**: The TCP log stream listener accepts connections from remote processes and delivers parsed log lines through a `LogStreamHandler` interface, set dynamically via `SetHandler()`. This allows any client to receive log messages without depending on a specific console implementation. - **TcpLogStreamSink for zen::logging**: A logging sink that forwards log messages to a `LogStreamListener` over TCP, using the native `zen::logging::Sink` infrastructure with proper thread-safe synchronization. - **Reliable child process exit codes on Linux**: `waitpid` result handling is fixed so `ProcessHandle::GetExitCode()` returns the real exit code. `ProcessHandle::Reset()` reaps zombies directly, replacing the global `IgnoreChildSignals()` which prevented exit code collection entirely. Also fixes a TOCTOU race in `ProcessHandle::Wait()` on Linux/Mac. - **Pipe capture test suite**: Tests covering stdout/stderr capture via pipes (both shared and separate modes), RAII cleanup, move semantics, and exit code propagation using `zentest-appstub` as the child process. - **Service command integration tests**: Shell-based integration tests for `zen service` covering the full lifecycle (install, status, start, stop, uninstall) on all three platforms — Linux (systemd), macOS (launchd), and Windows (SCM via PowerShell). - **Test script reorganization**: Platform-specific test scripts moved from `scripts/test_scripts/` into `scripts/test_linux/`, `test_mac/`, and `test_windows/`.
Diffstat (limited to 'src/zenutil')
-rw-r--r--src/zenutil/config/loggingconfig.cpp2
-rw-r--r--src/zenutil/include/zenutil/config/loggingconfig.h6
-rw-r--r--src/zenutil/include/zenutil/logging.h3
-rw-r--r--src/zenutil/include/zenutil/splitconsole/logstreamlistener.h66
-rw-r--r--src/zenutil/include/zenutil/splitconsole/tcplogstreamsink.h209
-rw-r--r--src/zenutil/splitconsole/logstreamlistener.cpp270
6 files changed, 553 insertions, 3 deletions
diff --git a/src/zenutil/config/loggingconfig.cpp b/src/zenutil/config/loggingconfig.cpp
index 5092c60aa..e2db31160 100644
--- a/src/zenutil/config/loggingconfig.cpp
+++ b/src/zenutil/config/loggingconfig.cpp
@@ -29,6 +29,8 @@ ZenLoggingCmdLineOptions::AddCliOptions(cxxopts::Options& options, ZenLoggingCon
("log-critical", "Change selected loggers to level CRITICAL", cxxopts::value<std::string>(LoggingConfig.Loggers[logging::Critical]))
("log-off", "Change selected loggers to level OFF", cxxopts::value<std::string>(LoggingConfig.Loggers[logging::Off]))
("otlp-endpoint", "OpenTelemetry endpoint URI (e.g http://localhost:4318)", cxxopts::value<std::string>(LoggingConfig.OtelEndpointUri))
+ ("force-color", "Force colored log output even when stdout is not a terminal", cxxopts::value<bool>(LoggingConfig.ForceColor)->default_value("false"))
+ ("log-stream", "TCP log stream endpoint (host:port)", cxxopts::value<std::string>(LoggingConfig.LogStreamEndpoint))
;
// clang-format on
}
diff --git a/src/zenutil/include/zenutil/config/loggingconfig.h b/src/zenutil/include/zenutil/config/loggingconfig.h
index b55b2d9f7..33a5eb172 100644
--- a/src/zenutil/include/zenutil/config/loggingconfig.h
+++ b/src/zenutil/include/zenutil/config/loggingconfig.h
@@ -16,10 +16,12 @@ struct ZenLoggingConfig
{
bool NoConsoleOutput = false; // Control default use of stdout for diagnostics
bool QuietConsole = false; // Configure console logger output to level WARN
+ bool ForceColor = false; // Force colored output even when stdout is not a terminal
std::filesystem::path AbsLogFile; // Absolute path to main log file
std::string Loggers[logging::LogLevelCount];
- std::string LogId; // Id for tagging log output
- std::string OtelEndpointUri; // OpenTelemetry endpoint URI
+ std::string LogId; // Id for tagging log output
+ std::string OtelEndpointUri; // OpenTelemetry endpoint URI
+ std::string LogStreamEndpoint; // TCP log stream endpoint (host:port)
};
void ApplyLoggingOptions(cxxopts::Options& options, ZenLoggingConfig& LoggingConfig);
diff --git a/src/zenutil/include/zenutil/logging.h b/src/zenutil/include/zenutil/logging.h
index 95419c274..282ae1b9a 100644
--- a/src/zenutil/include/zenutil/logging.h
+++ b/src/zenutil/include/zenutil/logging.h
@@ -28,7 +28,8 @@ struct LoggingOptions
bool AllowAsync = true;
bool NoConsoleOutput = false;
bool QuietConsole = false;
- std::filesystem::path AbsLogFile; // Absolute path to main log file
+ bool ForceColor = false; // Force colored output even when stdout is not a terminal
+ std::filesystem::path AbsLogFile; // Absolute path to main log file
std::string LogId;
};
diff --git a/src/zenutil/include/zenutil/splitconsole/logstreamlistener.h b/src/zenutil/include/zenutil/splitconsole/logstreamlistener.h
new file mode 100644
index 000000000..06544308c
--- /dev/null
+++ b/src/zenutil/include/zenutil/splitconsole/logstreamlistener.h
@@ -0,0 +1,66 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/zencore.h>
+
+#include <cstdint>
+#include <memory>
+#include <string>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <asio/io_context.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+namespace zen {
+
+/// Interface for receiving log lines from a LogStreamListener.
+/// Clients implement this to route received log messages to their desired output.
+class LogStreamHandler
+{
+public:
+ virtual ~LogStreamHandler() = default;
+
+ virtual void AppendLogLine(std::string Line) = 0;
+};
+
+/// TCP listener that accepts connections from remote processes streaming log messages.
+/// Each message is a CbObject with fields: "text" (string), "source" (string), "level" (string, optional).
+///
+/// A LogStreamHandler can be set to receive parsed log lines. If no handler is set,
+/// received messages are silently discarded.
+///
+/// Two modes of operation:
+/// - Owned thread: pass only Port; an internal IO thread is created.
+/// - External io_context: pass an existing asio::io_context; no thread is created,
+/// the caller is responsible for running the io_context.
+class LogStreamListener
+{
+public:
+ /// Start listening with an internal IO thread.
+ explicit LogStreamListener(uint16_t Port = 0);
+
+ /// Start listening on an externally-driven io_context (no thread created).
+ LogStreamListener(asio::io_context& IoContext, uint16_t Port = 0);
+
+ ~LogStreamListener();
+
+ LogStreamListener(const LogStreamListener&) = delete;
+ LogStreamListener& operator=(const LogStreamListener&) = delete;
+
+ /// Set the handler that will receive parsed log lines. May be called at any time.
+ /// Pass nullptr to stop delivering messages.
+ void SetHandler(LogStreamHandler* Handler);
+
+ /// Returns the actual port the listener is bound to.
+ uint16_t GetPort() const;
+
+ /// Gracefully stop accepting new connections and shut down existing sessions.
+ void Shutdown();
+
+private:
+ struct Impl;
+ std::unique_ptr<Impl> m_Impl;
+};
+
+} // namespace zen
diff --git a/src/zenutil/include/zenutil/splitconsole/tcplogstreamsink.h b/src/zenutil/include/zenutil/splitconsole/tcplogstreamsink.h
new file mode 100644
index 000000000..2ab7d469e
--- /dev/null
+++ b/src/zenutil/include/zenutil/splitconsole/tcplogstreamsink.h
@@ -0,0 +1,209 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#pragma once
+
+#include <zencore/compactbinarybuilder.h>
+#include <zencore/logging.h>
+#include <zencore/logging/sink.h>
+#include <zencore/thread.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <asio.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+#include <atomic>
+#include <condition_variable>
+#include <deque>
+#include <mutex>
+#include <string>
+#include <thread>
+
+namespace zen {
+
+/// Logging sink that connects to a LogStreamListener via TCP and sends CbObject-framed log messages.
+/// Connection is lazy (on first enqueued message) and silently reconnects on failure.
+/// Messages are serialized on the caller thread and written asynchronously from a dedicated IO thread.
+/// A bounded queue drops the oldest messages on overflow to prevent unbounded memory growth.
+class TcpLogStreamSink : public logging::Sink
+{
+public:
+ TcpLogStreamSink(const std::string& Host, uint16_t Port, std::string Source, uint32_t MaxQueueSize = 4096)
+ : m_Host(Host)
+ , m_Port(Port)
+ , m_Source(std::move(Source))
+ , m_MaxQueueSize(MaxQueueSize)
+ {
+ m_IoThread = std::thread([this]() { IoThreadMain(); });
+ }
+
+ ~TcpLogStreamSink() override
+ {
+ {
+ std::lock_guard<std::mutex> Lock(m_QueueMutex);
+ m_Stopping = true;
+ m_DrainDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
+ }
+ m_QueueCv.notify_one();
+ if (m_IoThread.joinable())
+ {
+ m_IoThread.join();
+ }
+ }
+
+ void Log(const logging::LogMessage& Msg) override
+ {
+ logging::MemoryBuffer Formatted;
+ {
+ RwLock::SharedLockScope Lock(m_FormatterLock);
+ if (m_Formatter)
+ {
+ m_Formatter->Format(Msg, Formatted);
+ }
+ else
+ {
+ // Fallback: use raw payload
+ auto Payload = Msg.GetPayload();
+ Formatted.append(Payload.data(), Payload.data() + Payload.size());
+ }
+ }
+
+ std::string_view Text(Formatted.data(), Formatted.size());
+
+ // Strip trailing newlines
+ while (!Text.empty() && (Text.back() == '\n' || Text.back() == '\r'))
+ {
+ Text.remove_suffix(1);
+ }
+
+ // Build CbObject with text, source, and level fields
+ CbObjectWriter Writer;
+ Writer.AddString("text", Text);
+ Writer.AddString("source", m_Source);
+ Writer.AddString("level", ToStringView(Msg.GetLevel()));
+ CbObject Obj = Writer.Save();
+
+ // Enqueue for async write
+ {
+ std::lock_guard<std::mutex> Lock(m_QueueMutex);
+ if (m_Queue.size() >= m_MaxQueueSize)
+ {
+ m_Queue.pop_front();
+ m_DroppedMessages.fetch_add(1, std::memory_order_relaxed);
+ }
+ m_Queue.push_back(std::move(Obj));
+ }
+ m_QueueCv.notify_one();
+ }
+
+ void Flush() override
+ {
+ // Nothing to flush — writes happen asynchronously
+ }
+
+ void SetFormatter(std::unique_ptr<logging::Formatter> InFormatter) override
+ {
+ RwLock::ExclusiveLockScope Lock(m_FormatterLock);
+ m_Formatter = std::move(InFormatter);
+ }
+
+private:
+ void IoThreadMain()
+ {
+ zen::SetCurrentThreadName("TcpLogSink");
+
+ for (;;)
+ {
+ std::deque<CbObject> Batch;
+ {
+ std::unique_lock<std::mutex> Lock(m_QueueMutex);
+ m_QueueCv.wait(Lock, [this]() { return m_Stopping || !m_Queue.empty(); });
+
+ if (m_Stopping && m_Queue.empty())
+ {
+ break;
+ }
+
+ if (m_Stopping && std::chrono::steady_clock::now() >= m_DrainDeadline)
+ {
+ break;
+ }
+
+ Batch.swap(m_Queue);
+ }
+
+ uint32_t Dropped = m_DroppedMessages.exchange(0, std::memory_order_relaxed);
+ if (Dropped > 0)
+ {
+ // We could/should log here, but that could cause a feedback loop which
+ // would trigger subsequent dropped message warnings
+ }
+
+ if (!m_Connected && !Connect())
+ {
+ if (m_Stopping)
+ {
+ break; // don't retry during shutdown
+ }
+ continue; // drop batch — will retry on next batch
+ }
+
+ for (auto& Obj : Batch)
+ {
+ MemoryView View = Obj.GetView();
+ asio::error_code Ec;
+ asio::write(m_Socket, asio::buffer(View.GetData(), View.GetSize()), Ec);
+ if (Ec)
+ {
+ m_Connected = false;
+ break; // drop remaining messages in batch
+ }
+ }
+ }
+ }
+
+ bool Connect()
+ {
+ try
+ {
+ asio::ip::tcp::resolver Resolver(m_IoContext);
+ auto Endpoints = Resolver.resolve(m_Host, std::to_string(m_Port));
+ asio::connect(m_Socket, Endpoints);
+ m_Connected = true;
+ return true;
+ }
+ catch (const std::exception&)
+ {
+ // Reset the socket for next attempt
+ m_Socket = asio::ip::tcp::socket(m_IoContext);
+ m_Connected = false;
+ return false;
+ }
+ }
+
+ // IO thread state (only accessed from m_IoThread)
+ asio::io_context m_IoContext;
+ asio::ip::tcp::socket m_Socket{m_IoContext};
+ bool m_Connected = false;
+
+ // Configuration (immutable after construction)
+ std::string m_Host;
+ uint16_t m_Port;
+ std::string m_Source;
+ uint32_t m_MaxQueueSize;
+
+ // Formatter (protected by RwLock since Log is called from multiple threads)
+ RwLock m_FormatterLock;
+ std::unique_ptr<logging::Formatter> m_Formatter;
+
+ // Queue shared between Log() callers and IO thread
+ std::mutex m_QueueMutex;
+ std::condition_variable m_QueueCv;
+ std::deque<CbObject> m_Queue;
+ std::atomic<uint32_t> m_DroppedMessages{0};
+ bool m_Stopping = false;
+ std::chrono::steady_clock::time_point m_DrainDeadline;
+
+ std::thread m_IoThread;
+};
+
+} // namespace zen
diff --git a/src/zenutil/splitconsole/logstreamlistener.cpp b/src/zenutil/splitconsole/logstreamlistener.cpp
new file mode 100644
index 000000000..9f1d1a02c
--- /dev/null
+++ b/src/zenutil/splitconsole/logstreamlistener.cpp
@@ -0,0 +1,270 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include <zenutil/splitconsole/logstreamlistener.h>
+
+#include <zencore/compactbinary.h>
+#include <zencore/fmtutils.h>
+#include <zencore/logging.h>
+#include <zencore/thread.h>
+
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <asio.hpp>
+ZEN_THIRD_PARTY_INCLUDES_END
+
+#include <atomic>
+#include <vector>
+
+namespace zen {
+
+//////////////////////////////////////////////////////////////////////////
+// LogStreamSession — reads CbObject-framed messages from a single TCP connection
+
+class LogStreamSession : public std::enable_shared_from_this<LogStreamSession>
+{
+public:
+ LogStreamSession(asio::ip::tcp::socket Socket, std::atomic<LogStreamHandler*>& Handler)
+ : m_Socket(std::move(Socket))
+ , m_Handler(Handler)
+ {
+ }
+
+ void Start() { DoRead(); }
+
+private:
+ void DoRead()
+ {
+ auto Self = shared_from_this();
+ m_Socket.async_read_some(asio::buffer(m_ReadBuf.data() + m_BufferUsed, m_ReadBuf.size() - m_BufferUsed),
+ [Self](const asio::error_code& Ec, std::size_t BytesRead) {
+ if (Ec)
+ {
+ return; // connection closed or error — session ends
+ }
+ Self->m_BufferUsed += BytesRead;
+ Self->ProcessBuffer();
+ Self->DoRead();
+ });
+ }
+
+ void ProcessBuffer()
+ {
+ // Try to consume as many complete CbObject messages as possible
+ while (m_BufferUsed > 0)
+ {
+ MemoryView View = MakeMemoryView(m_ReadBuf.data(), m_BufferUsed);
+ CbFieldType Type;
+ uint64_t Size = 0;
+
+ if (!TryMeasureCompactBinary(View, Type, Size))
+ {
+ break; // need more data
+ }
+
+ if (Size > m_BufferUsed)
+ {
+ break; // need more data
+ }
+
+ // Parse the CbObject
+ CbObject Obj = CbObject(SharedBuffer::Clone(MakeMemoryView(m_ReadBuf.data(), Size)));
+
+ std::string_view Text = Obj["text"].AsString();
+ std::string_view Source = Obj["source"].AsString();
+
+ // Split multi-line messages into individual AppendLogLine calls so that
+ // each line gets its own row in the SplitConsole ring buffer.
+ while (!Text.empty())
+ {
+ std::string_view Line = Text;
+ auto Pos = Text.find('\n');
+ if (Pos != std::string_view::npos)
+ {
+ Line = Text.substr(0, Pos);
+ Text.remove_prefix(Pos + 1);
+ }
+ else
+ {
+ Text = {};
+ }
+
+ // Strip trailing CR from CRLF
+ if (!Line.empty() && Line.back() == '\r')
+ {
+ Line.remove_suffix(1);
+ }
+
+ if (Line.empty())
+ {
+ continue;
+ }
+
+ LogStreamHandler* Handler = m_Handler.load(std::memory_order_acquire);
+ if (Handler)
+ {
+ if (!Source.empty())
+ {
+ Handler->AppendLogLine(fmt::format("[{}] {}", Source, Line));
+ }
+ else
+ {
+ Handler->AppendLogLine(std::string(Line));
+ }
+ }
+ }
+
+ // Remove consumed bytes from buffer
+ std::size_t Consumed = static_cast<std::size_t>(Size);
+ std::memmove(m_ReadBuf.data(), m_ReadBuf.data() + Consumed, m_BufferUsed - Consumed);
+ m_BufferUsed -= Consumed;
+ }
+
+ // If buffer is full and we can't parse a message, the message is too large — drop connection
+ if (m_BufferUsed == m_ReadBuf.size())
+ {
+ ZEN_WARN("LogStreamSession: buffer full with no complete message, dropping connection");
+ asio::error_code Ec;
+ m_Socket.close(Ec);
+ m_BufferUsed = 0;
+ }
+ }
+
+ asio::ip::tcp::socket m_Socket;
+ std::atomic<LogStreamHandler*>& m_Handler;
+ std::array<uint8_t, 65536> m_ReadBuf{};
+ std::size_t m_BufferUsed = 0;
+};
+
+//////////////////////////////////////////////////////////////////////////
+// LogStreamListener::Impl
+
+struct LogStreamListener::Impl
+{
+ // Owned io_context mode — creates and runs its own thread
+ explicit Impl(uint16_t Port) : m_OwnedIoContext(std::make_unique<asio::io_context>()), m_Acceptor(*m_OwnedIoContext)
+ {
+ SetupAcceptor(Port);
+ m_IoThread = std::thread([this]() {
+ zen::SetCurrentThreadName("LogStreamIO");
+ m_OwnedIoContext->run();
+ });
+ }
+
+ // External io_context mode — caller drives the io_context
+ Impl(asio::io_context& IoContext, uint16_t Port) : m_Acceptor(IoContext) { SetupAcceptor(Port); }
+
+ ~Impl() { Shutdown(); }
+
+ void Shutdown()
+ {
+ if (m_Stopped.exchange(true))
+ {
+ return;
+ }
+
+ asio::error_code Ec;
+ m_Acceptor.close(Ec);
+
+ if (m_OwnedIoContext)
+ {
+ m_OwnedIoContext->stop();
+ }
+
+ if (m_IoThread.joinable())
+ {
+ m_IoThread.join();
+ }
+ }
+
+ uint16_t GetPort() const { return m_Port; }
+
+ void SetHandler(LogStreamHandler* Handler) { m_Handler.store(Handler, std::memory_order_release); }
+
+private:
+ void SetupAcceptor(uint16_t Port)
+ {
+ // Try dual-stack IPv6 first (accepts both IPv4 and IPv6), fall back to IPv4-only
+ asio::error_code Ec;
+ m_Acceptor.open(asio::ip::tcp::v6(), Ec);
+ if (!Ec)
+ {
+ m_Acceptor.set_option(asio::ip::v6_only(false), Ec);
+ if (!Ec)
+ {
+ m_Acceptor.bind(asio::ip::tcp::endpoint(asio::ip::tcp::v6(), Port), Ec);
+ }
+ }
+
+ if (Ec)
+ {
+ // Fall back to IPv4-only
+ if (m_Acceptor.is_open())
+ {
+ m_Acceptor.close();
+ }
+ m_Acceptor.open(asio::ip::tcp::v4());
+ m_Acceptor.bind(asio::ip::tcp::endpoint(asio::ip::tcp::v4(), Port));
+ }
+
+ m_Acceptor.listen();
+ m_Port = m_Acceptor.local_endpoint().port();
+ StartAccept();
+ }
+
+ void StartAccept()
+ {
+ m_Acceptor.async_accept([this](const asio::error_code& Ec, asio::ip::tcp::socket Socket) {
+ if (Ec)
+ {
+ return; // acceptor closed
+ }
+
+ auto Session = std::make_shared<LogStreamSession>(std::move(Socket), m_Handler);
+ Session->Start();
+
+ if (!m_Stopped.load())
+ {
+ StartAccept();
+ }
+ });
+ }
+
+ std::atomic<LogStreamHandler*> m_Handler{nullptr};
+ std::unique_ptr<asio::io_context> m_OwnedIoContext; // null when using external io_context
+ asio::ip::tcp::acceptor m_Acceptor;
+ std::thread m_IoThread;
+ uint16_t m_Port = 0;
+ std::atomic<bool> m_Stopped{false};
+};
+
+//////////////////////////////////////////////////////////////////////////
+// LogStreamListener
+
+LogStreamListener::LogStreamListener(uint16_t Port) : m_Impl(std::make_unique<Impl>(Port))
+{
+}
+
+LogStreamListener::LogStreamListener(asio::io_context& IoContext, uint16_t Port) : m_Impl(std::make_unique<Impl>(IoContext, Port))
+{
+}
+
+LogStreamListener::~LogStreamListener() = default;
+
+void
+LogStreamListener::SetHandler(LogStreamHandler* Handler)
+{
+ m_Impl->SetHandler(Handler);
+}
+
+uint16_t
+LogStreamListener::GetPort() const
+{
+ return m_Impl->GetPort();
+}
+
+void
+LogStreamListener::Shutdown()
+{
+ m_Impl->Shutdown();
+}
+
+} // namespace zen