1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

#include "lix/libutil/c-calls.hh"
#include "lix/libutil/error.hh"
#include "lix/libutil/file-descriptor.hh"
#include "lix/libutil/file-system.hh"
#include "lix/libstore/globals.hh"
#include "lix/libstore/build/hook-instance.hh"
#include "lix/libutil/json.hh"
#include "lix/libutil/logging.hh"
#include "lix/libutil/processes.hh"
#include "lix/libutil/result.hh"
#include "lix/libutil/rpc.hh"
#include "lix/libutil/serialise.hh"
#include "lix/libutil/strings.hh"
#include "lix/libutil/logging-rpc.hh" // IWYU pragma: keep
#include "lix/libutil/types-rpc.hh" // IWYU pragma: keep
#include <fcntl.h>
#include <kj/common.h>
#include <kj/memory.h>
#include <memory>
#include <string_view>
#include <unistd.h>

namespace nix {

void HookInstance::HookLogger::emitLog(rpc::log::Event::Result::Reader r)
{
    auto type = rpc::log::from(r.getType());
    auto fields = r.getFields();
    if (!type) {
        return;
    }

    // ensure that logs from a builder using `ssh-ng://` as protocol
    // are also available to `nix log`.
    if (type == resBuildLogLine) {
        if (fields.size() > 0 && fields[0].isS()) {
            (*logSink)(fmt("%s\n", rpc::to<std::string_view>(fields[0].getS())));
        } else {
            (*logSink)("\n");
        }
    } else if (type == resSetPhase && fields.size() > 0 && fields[0].isS()) {
        // nixpkgs' stdenv produces lines in the log to signal phase changes.
        // We want to get the same lines in case of remote builds.
        // The format is:
        //   @nix { "action": "setPhase", "phase": "$curPhase" }
        const auto phase = rpc::to<std::string_view>(fields[0].getS());
        const auto logLine = JSON::object({{"action", "setPhase"}, {"phase", phase}});
        (*logSink)("@nix " + logLine.dump(-1, ' ', false, JSON::error_handler_t::replace) + "\n");
    }
}

kj::Promise<void> HookInstance::HookLogger::push(PushContext context)
{
    try {
        auto e = context.getParams().getE();
        if (logSink && e.isResult()) {
            emitLog(e.getResult());
        }
    } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions)
        printError("error in log processor: %s", e.what());
        throw; // NOLINT(lix-foreign-exceptions)
    }

    return RpcLoggerServer::push(context);
}

kj::Promise<Result<std::unique_ptr<HookInstance>>> HookInstance::create(const Activity & act)
try {
    debug("starting build hook '%s'", concatStringsSep(" ", settings.buildHook.get()));

    auto buildHookArgs = settings.buildHook.get();

    if (buildHookArgs.empty())
        throw Error("'build-hook' setting is empty");

    auto buildHook = canonPath(buildHookArgs.front());
    buildHookArgs.emplace(std::next(buildHookArgs.begin()), baseNameOf(buildHook));
    buildHookArgs.push_back(std::to_string(verbosity));

    /* Create the communication pipes. */
    auto [selfRPC, hookRPC] = SocketPair::stream();

    /* Fork the hook. */
    auto pid = runHelper(
        "run-build-hook",
        {
            .args = buildHookArgs,
            .redirections = {{.dup = STDOUT_FILENO, .from = hookRPC.get()}},
        }
    );
    KJ_DEFER({
        // kill the hook if the promise is cancelled. the hook helper creates
        // a session, so we'll kill the entire process group just to be safe.
        if (pid) {
            pid.killProcessGroup();
        }
    });

    std::map<std::string, Config::SettingInfo> settings;
    globalConfig.getSettings(settings, true);

    auto conn = AIO().lowLevelProvider.wrapUnixSocketFd(kj::AutoCloseFd(selfRPC.release()));
    auto client = std::make_unique<capnp::TwoPartyClient>(*conn, 1);
    auto rpc = client->bootstrap().castAs<rpc::build_remote::HookInstance>();

    {
        auto initReq = rpc.initRequest();
        initReq.setLogger(kj::heap<HookLogger>(act, nullptr));
        RPC_FILL(initReq, initSettings, settings);
        TRY_AWAIT_RPC(initReq.send());
    }

    co_return std::make_unique<HookInstance>(
        kj::heap(std::move(rpc)).attach(std::move(conn), std::move(client)), std::move(pid)
    );
} catch (...) {
    co_return result::current_exception();
}

HookInstance::~HookInstance()
{
    try {
        kill();
    } catch (...) {
        ignoreExceptionInDestructor();
    }
}

}