diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc index 37b2f48cb..584a19a95 100644 --- a/lix/libstore/filetransfer.cc +++ b/lix/libstore/filetransfer.cc @@ -375,7 +375,7 @@ struct curlFileTransfer : public FileTransfer { try { if (act.progress(dlnow, dltotal) == Logger::BufferState::NeedsFlush) { - act.getLogger().waitForSpace(); + act.getLogger().waitForSpace(); // NOLINT(lix-never-async) } } catch (nix::Interrupted &) { } @@ -409,7 +409,7 @@ struct curlFileTransfer : public FileTransfer else if (code == CURLE_OK && successfulStatuses.count(httpStatus)) { if (act.progress(bodySize, bodySize) == Logger::BufferState::NeedsFlush) { - act.getLogger().waitForSpace(); + act.getLogger().waitForSpace(); // NOLINT(lix-never-async) } auto state = downloadState.lock(); state->done = true; diff --git a/lix/libutil/logging-rpc.cc b/lix/libutil/logging-rpc.cc new file mode 100644 index 000000000..46a9e4c10 --- /dev/null +++ b/lix/libutil/logging-rpc.cc @@ -0,0 +1,324 @@ +#include "logging-rpc.hh" +#include "types-rpc.hh" // IWYU pragma: keep +#include "sync.hh" +#include "types.hh" +#include +#include + +namespace nix { +namespace { +struct RpcLogger : Logger +{ + struct Log + { + Verbosity level; + std::string msg; + }; + struct LogEI + { + ErrorInfo ei; + }; + struct StartActivity + { + Verbosity level; + uint64_t id; + ActivityType type; + std::string text; + uint64_t parent; + Logger::Fields fields; + }; + struct StopActivity + { + uint64_t id; + }; + struct ActivityResult + { + uint64_t id; + ResultType type; + Logger::Fields fields; + }; + + using Event = std::variant; + + struct Buffer + { + std::vector items; + /// *very* rough approximation of how much memory our buffer uses. this + /// should not be an exact byte count to keep accounting simple, but it + /// should still be roughly representative of reality. a small constant + /// error factor during average execution is acceptable, expected even. + size_t sizeEstimate; + + /// non-null if a remote log operation failed. we'll rethrow it blindly + /// every time the buffers are flushed, but not during message enqueue. + /// this hopefully avoids crashing due to recursive exception handling. + std::exception_ptr failure; + + auto take() + { + if (failure) { + std::rethrow_exception(failure); + } + sizeEstimate = 0; + return std::move(items); + } + }; + + rpc::log::LogStream::Client remote; + Sync buffer; + Sync>>, AsyncMutex> flushReq; + kj::Promise flusher; + + RpcLogger(rpc::log::LogStream::Client remote) + : remote(remote) + , flusher(flushLoop().eagerlyEvaluate([](kj::Exception && e) { + if (e.getType() == kj::Exception::Type::DISCONNECTED) { + std::cerr << "peer disconnected, exiting with haste\n"; + std::exit(90); + } else { + std::cerr << "log flusher failed catastrophically!\n"; + // rethrow the exception and terminate to cause a core dump and stack trace + try { + kj::throwFatalException(std::move(e)); + } catch (...) { + std::terminate(); + } + } + })) + { + } + + ~RpcLogger() noexcept = default; + + static size_t fieldSize(const Logger::Fields & fields) + { + size_t size = 0; + for (auto & f : fields) { + size += sizeof(f) + f.s.size(); + } + return size; + } + + BufferState push(size_t extraSize, Event e) + { + auto buffer = this->buffer.lock(); + if (buffer->failure) { + return BufferState::HasSpace; + } + buffer->sizeEstimate += sizeof(e) + extraSize; + buffer->items.emplace_back(std::move(e)); + return buffer->sizeEstimate >= 1024 * 1024 ? BufferState::NeedsFlush + : BufferState::HasSpace; + } + + BufferState log(Verbosity lvl, std::string_view s) override + { + return push(s.size(), Log{lvl, std::string(s)}); + } + + BufferState logEI(const ErrorInfo & ei) override + { + // size is just a guess. errors are usually rare and small + return push(1024, LogEI{ei}); + } + + BufferState startActivityImpl( + ActivityId act, + Verbosity lvl, + ActivityType type, + const std::string & s, + const Fields & fields, + ActivityId parent + ) override + { + return push(fieldSize(fields), StartActivity{lvl, act, type, s, parent, fields}); + } + + BufferState stopActivityImpl(ActivityId act) override + { + return push(0, StopActivity{act}); + } + + BufferState resultImpl(ActivityId act, ResultType type, const Fields & fields) override + { + return push(fieldSize(fields), ActivityResult{act, type, fields}); + } + + kj::Promise> flush() override + try { + auto pfp = kj::newPromiseAndCrossThreadFulfiller(); + (co_await flushReq.lock())->emplace_back(std::move(pfp.fulfiller)); + flushReq.notify(); + co_await pfp.promise; + co_return result::success(); + } catch (...) { + co_return result::current_exception(); + } + + static void fillEventArg(rpc::log::Event::Builder arg, const Event & e) + { + overloaded handlers{ + [&](const Log & l) { + arg.initLog().setLevel(rpc::Verbosity(l.level)); + RPC_FILL(arg.getLog(), setMsg, l.msg); + }, + [&](const LogEI & l) { + auto ei = arg.initLogEI(); + RPC_FILL(ei, initInfo, l.ei); + }, + [&](const StartActivity & s) { + auto sa = arg.initStartActivity(); + sa.setLevel(rpc::Verbosity(s.level)); + sa.setId(s.id); + sa.setType(rpc::log::to(s.type)); + RPC_FILL(sa, setText, s.text); + sa.setParent(s.parent); + RPC_FILL(sa, initFields, s.fields); + }, + [&](const StopActivity & s) { arg.initStopActivity().setId(s.id); }, + [&](const ActivityResult & r) { + auto ar = arg.initResult(); + ar.setId(r.id); + ar.setType(rpc::log::to(r.type)); + RPC_FILL(ar, initFields, r.fields); + }, + }; + std::visit(handlers, e); + } + + kj::Promise flushLoop() + { + auto req = co_await flushReq.lock(); + std::optional failure; + + while (true) { + kj::Own> f; + if (req->empty()) { + co_await req.waitFor(100 * kj::MILLISECONDS); + } + if (!req->empty()) { + f = std::move(req->front()); + req->pop_front(); + } + + try { + auto buffer = this->buffer.lock()->take(); + for (auto & e : buffer) { + auto req = remote.pushRequest(); + fillEventArg(req.initE(), e); + co_await req.send(); + } + co_await remote.synchronizeRequest().send(); + if (f) { + f->fulfill(); + } + } catch (...) { + this->buffer.lock()->failure = std::current_exception(); + failure.emplace(kj::getCaughtExceptionAsKj()); + if (f) { + f->reject(auto(*failure)); + } + break; + } + } + + while (true) { + while (req->empty()) { + co_await req.wait(); + } + req->front()->reject(auto(*failure)); + req->pop_front(); + } + } + + void waitForSpace(NeverAsync) override + { + struct Fulfiller final : kj::PromiseFulfiller + { + std::shared_ptr flag; + + Fulfiller(std::shared_ptr flag) : flag(flag) {} + + void fulfill(kj::_::Void && value = {}) override + { + flag->clear(); + flag->notify_all(); + } + + void reject(kj::Exception && exception) override + { + fulfill(); + } + + bool isWaiting() override + { + return flag->test(); + } + }; + + auto flag = std::make_shared(true); + + flushReq.lockSync()->emplace_back(kj::heap(flag)); + flushReq.notify(); + flag->wait(true); + } +}; +} +} + +namespace nix::rpc::log { +Logger * makeRpcLoggerClient(LogStream::Client remote) +{ + return new RpcLogger(remote); +} + +kj::Promise RpcLoggerServer::push(PushContext context) +try { + auto state = Logger::BufferState::HasSpace; + + auto e = context.getParams().getE(); + if (e.isStartActivity()) { + auto args = e.getStartActivity(); + activities.emplace( + args.getId(), + parent.addChild( + nix::Verbosity(args.getLevel()), + from(args.getType()).value_or(actUnknown), + rpc::to(args.getText()), + rpc::to(args.getFields()) + ) + ); + } else if (e.isStopActivity()) { + activities.erase(e.getStopActivity().getId()); + } else if (e.isResult()) { + auto args = e.getResult(); + if (auto type = rpc::log::from(args.getType())) { + if (auto act = get(activities, args.getId())) { + state = act->result(*type, rpc::to(args.getFields())); + } + } else { + debug("got unintellegible result message %s", args.toString().flatten().cStr()); + } + } else if (e.isLog()) { + state = logger->log( + nix::Verbosity(e.getLog().getLevel()), rpc::to(e.getLog().getMsg()) + ); + } else if (e.isLogEI()) { + state = logger->logEI(from(e.getLogEI().getInfo())); + } else { + debug("got unintellegible log message %s", e.toString().flatten().cStr()); + } + + if (state == Logger::BufferState::NeedsFlush) { + TRY_AWAIT(parent.getLogger().flush()); + } +} catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) + printError("error in log processor: %s", e.what()); + throw; // NOLINT(lix-foreign-exceptions) +} + +kj::Promise RpcLoggerServer::synchronize(SynchronizeContext context) +{ + return kj::READY_NOW; +} +} diff --git a/lix/libutil/logging-rpc.hh b/lix/libutil/logging-rpc.hh new file mode 100644 index 000000000..e8de0bc32 --- /dev/null +++ b/lix/libutil/logging-rpc.hh @@ -0,0 +1,137 @@ +#pragma once +///@file + +#include "logging.capnp.h" +#include "logging.hh" +#include "rpc.hh" // IWYU pragma: keep +#include + +namespace nix::rpc::log { +inline std::optional from(ActivityType at) +{ + // clang-format off + switch (at) { + case ActivityType::UNKNOWN: return actUnknown; + case ActivityType::COPY_PATH: return actCopyPath; + case ActivityType::FILE_TRANSFER: return actFileTransfer; + case ActivityType::REALISE: return actRealise; + case ActivityType::COPY_PATHS: return actCopyPaths; + case ActivityType::BUILDS: return actBuilds; + case ActivityType::BUILD: return actBuild; + case ActivityType::OPTIMISE_STORE: return actOptimiseStore; + case ActivityType::VERIFY_PATHS: return actVerifyPaths; + case ActivityType::SUBSTITUTE: return actSubstitute; + case ActivityType::QUERY_PATH_INFO: return actQueryPathInfo; + case ActivityType::POST_BUILD_HOOK: return actPostBuildHook; + case ActivityType::BUILD_WAITING: return actBuildWaiting; + default: return std::nullopt; + } + // clang-format on +} + +inline ActivityType to(const nix::ActivityType & at) +{ + // clang-format off + switch (at) { + case actUnknown: return log::ActivityType::UNKNOWN; + case actCopyPath: return log::ActivityType::COPY_PATH; + case actFileTransfer: return log::ActivityType::FILE_TRANSFER; + case actRealise: return log::ActivityType::REALISE; + case actCopyPaths: return log::ActivityType::COPY_PATHS; + case actBuilds: return log::ActivityType::BUILDS; + case actBuild: return log::ActivityType::BUILD; + case actOptimiseStore: return log::ActivityType::OPTIMISE_STORE; + case actVerifyPaths: return log::ActivityType::VERIFY_PATHS; + case actSubstitute: return log::ActivityType::SUBSTITUTE; + case actQueryPathInfo: return log::ActivityType::QUERY_PATH_INFO; + case actPostBuildHook: return log::ActivityType::POST_BUILD_HOOK; + case actBuildWaiting: return log::ActivityType::BUILD_WAITING; + } + // clang-format on +} + +inline std::optional from(ResultType rt) +{ + // clang-format off + switch (rt) { + case ResultType::FILE_LINKED: return resFileLinked; + case ResultType::BUILD_LOG_LINE: return resBuildLogLine; + case ResultType::UNTRUSTED_PATH: return resUntrustedPath; + case ResultType::CORRUPTED_PATH: return resCorruptedPath; + case ResultType::SET_PHASE: return resSetPhase; + case ResultType::PROGRESS: return resProgress; + case ResultType::SET_EXPECTED: return resSetExpected; + case ResultType::POST_BUILD_LOG_LINE: return resPostBuildLogLine; + default: return std::nullopt; + } + // clang-format on +} + +inline ResultType to(const nix::ResultType & rt) +{ + // clang-format off + switch (rt) { + case resFileLinked: return ResultType::FILE_LINKED; + case resBuildLogLine: return ResultType::BUILD_LOG_LINE; + case resUntrustedPath: return ResultType::UNTRUSTED_PATH; + case resCorruptedPath: return ResultType::CORRUPTED_PATH; + case resSetPhase: return ResultType::SET_PHASE; + case resProgress: return ResultType::PROGRESS; + case resSetExpected: return ResultType::SET_EXPECTED; + case resPostBuildLogLine: return ResultType::POST_BUILD_LOG_LINE; + } + // clang-format on +} +} + +namespace nix::rpc { +template<> +struct Convert +{ + static nix::Logger::Field convert(log::Event::Field::Reader r, auto &&...) + { + if (r.isI()) { + return {r.getI()}; + } else { + return {to(r.getS())}; + } + } +}; + +template<> +struct Fill +{ + static void fill(log::Event::Field::Builder fb, const Logger::Field & e, auto &&...) + { + if (e.type == Logger::Field::tInt) { + fb.setI(e.i); + } else { + LIX_RPC_FILL(fb, setS, e.s); + } + } +}; +} + +namespace nix::rpc::log { +/** + * create an RPC-backed logger. this logger will flush its contents periodically + * (current fixed to a 100ms interval) or once the log buffer fills up enough to + * warrant immediate traffic (currently fixed to approximately 1 MiB of buffered + * log traffic). *any* error caught during rpc calls will terminate the process. + */ +Logger * makeRpcLoggerClient(LogStream::Client remote); + +class RpcLoggerServer : public LogStream::Server +{ +private: + const Activity & parent; + std::map activities; + +public: + RpcLoggerServer(const Activity & parent) : parent(parent) {} + virtual ~RpcLoggerServer() noexcept(false) = default; + + kj::Promise push(PushContext context) override; + kj::Promise synchronize(SynchronizeContext context) override; +}; +} diff --git a/lix/libutil/logging.capnp b/lix/libutil/logging.capnp new file mode 100644 index 000000000..ba44d5187 --- /dev/null +++ b/lix/libutil/logging.capnp @@ -0,0 +1,93 @@ +@0xff79d868797f140a; + +# IMPORTANT NOTICE +# +# these definitions are EXPERIMENTAL and come with NO stability guarantees + +using Cxx = import "/capnp/c++.capnp"; +$Cxx.namespace("nix::rpc::log"); +$Cxx.allowCancellation; + +using Types = import "/lix/libutil/types.capnp"; + +enum ActivityType { + unknown @0; + copyPath @1; + fileTransfer @2; + realise @3; + copyPaths @4; + builds @5; + build @6; + optimiseStore @7; + verifyPaths @8; + substitute @9; + queryPathInfo @10; + postBuildHook @11; + buildWaiting @12; +} + +enum ResultType { + fileLinked @0; + buildLogLine @1; + untrustedPath @2; + corruptedPath @3; + setPhase @4; + progress @5; + setExpected @6; + postBuildLogLine @7; +} + +struct Event { + struct Field { + union { + s @0 :Types.String; + i @1 :UInt64; + } + } + + union { + log :group { + level @0 :Types.Verbosity; + msg @1 :Types.String; + } + + logEI :group { + info @2 :Types.Error; + } + + startActivity :group { + level @3 :Types.Verbosity; + id @4 :UInt64; + type @5 :ActivityType; + text @6 :Types.String; + parent @7 :UInt64; + fields @8 :List(Field); + } + + stopActivity :group { + id @9 :UInt64; + } + + result :group { + id @10 :UInt64; + type @11 :ResultType; + fields @12 :List(Field); + } + } +} + +# NOTE: these streams do not return any result. loggers are expected to +# be infallible since once the logger fails the only thing we can still +# do is panic. reporting any status is impossible, and writing to other +# streams (like stdout/stderr) should be reserved for emergencies only. +# +# we also purposely do not model the c++ api to the loggers here. since +# the rpc definitions are still experimental and used only within *one* +# defined version of lix on a single machine we can get away with this. + +interface LogStream { + push @0 (e :Event) -> stream; + + # flush rpc streams and synchronize capnp flow control/error state + synchronize @1 (); +} diff --git a/lix/libutil/logging.hh b/lix/libutil/logging.hh index 198b6a933..b64c3769f 100644 --- a/lix/libutil/logging.hh +++ b/lix/libutil/logging.hh @@ -167,7 +167,7 @@ public: return {result::success()}; } - virtual void waitForSpace() {} + virtual void waitForSpace(NeverAsync = {}) {} protected: virtual BufferState startActivityImpl( diff --git a/lix/libutil/meson.build b/lix/libutil/meson.build index 8ca4eded3..7b4b033bd 100644 --- a/lix/libutil/meson.build +++ b/lix/libutil/meson.build @@ -25,6 +25,7 @@ libutil_sources = files( 'hilite.cc', 'io-buffer.cc', 'json-utils.cc', + 'logging-rpc.cc', 'logging.cc', 'monitor-fd.cc', 'mount.cc', @@ -102,6 +103,7 @@ libutil_headers = files( 'json-fwd.hh', 'json.hh', 'linear-map.hh', + 'logging-rpc.hh', 'logging.hh', 'lru-cache.hh', 'manually-drop.hh', @@ -296,33 +298,35 @@ libutil_settings_headers += custom_target( libutil_rpc_headers = [] libutil_rpc_sources = [] -libutil_rpc = [] -libutil_rpc += custom_target( - command : [ - capnpc_wrapper, - '--language=c++', - '--src-prefix=@CURRENT_SOURCE_DIR@', - '--outdir=@OUTDIR@', - '--depfile=@DEPFILE@', - '-I@SOURCE_ROOT@', - '@INPUT@', - ], - input : files( - # keep-sorted start - 'types.capnp', - # keep-sorted end - ), - output : [ - '@PLAINNAME@.h', - '@PLAINNAME@.c++', - ], - install : true, - install_dir : [includedir / 'lix/libutil', false], - depfile : '@PLAINNAME@.d', +libutil_rpc_schemas = files( + # keep-sorted start + 'logging.capnp', + 'types.capnp', + # keep-sorted end ) -foreach rpc : libutil_rpc +foreach rpc : libutil_rpc_schemas + rpc = custom_target( + command : [ + capnpc_wrapper, + '--language=c++', + '--src-prefix=@CURRENT_SOURCE_DIR@', + '--outdir=@OUTDIR@', + '--depfile=@DEPFILE@', + '-I@SOURCE_ROOT@', + '@INPUT@', + ], + input : rpc, + output : [ + '@PLAINNAME@.h', + '@PLAINNAME@.c++', + ], + install : true, + install_dir : [includedir / 'lix/libutil', false], + depfile : '@PLAINNAME@.d', + ) + libutil_rpc_headers += rpc[0] libutil_rpc_sources += rpc[1] endforeach @@ -332,6 +336,7 @@ dependencies = [ brotli, cpuid, kj, + capnp_rpc, libarchive, nlohmann_json, openssl, @@ -410,6 +415,7 @@ liblixutil = declare_dependency( # Everything has to link to kj if it uses libutil internally (ensured by # lix-base pkg-config externally) kj, + capnp_rpc, libarchive, nlohmann_json, libatomic, diff --git a/lix/libutil/types.capnp b/lix/libutil/types.capnp index ef82cea76..54181a51e 100644 --- a/lix/libutil/types.capnp +++ b/lix/libutil/types.capnp @@ -3,6 +3,9 @@ using Cxx = import "/capnp/c++.capnp"; $Cxx.namespace("nix::rpc"); +# many of our strings must be nul-safe :( +using String = Data; + enum Verbosity { error @0; warn @1;