Add an S3-backed binary cache store

release.nix

@@ -159,6 +159,10 @@ rec {
           guile # optional, for Guile + Guix support
           perlDeps perl
           postgresql92 # for running the tests
+          (aws-sdk-cpp.override {
+            apis = ["s3"];
+            customMemoryManagement = false;
+          })
         ];
 
       hydraPath = lib.makeSearchPath "bin" (

src/hydra-queue-runner/Makefile.am

@@ -4,7 +4,8 @@ hydra_queue_runner_SOURCES = hydra-queue-runner.cc queue-monitor.cc dispatcher.c
  builder.cc build-result.cc build-remote.cc  \
  build-result.hh counter.hh pool.hh sync.hh token-server.hh state.hh db.hh \
  binary-cache-store.hh binary-cache-store.cc \
- local-binary-cache-store.hh local-binary-cache-store.cc
+ local-binary-cache-store.hh local-binary-cache-store.cc \
+ s3-binary-cache-store.hh s3-binary-cache-store.cc
 hydra_queue_runner_LDADD = $(NIX_LIBS) -lpqxx
 
-AM_CXXFLAGS = $(NIX_CFLAGS) -Wall
+AM_CXXFLAGS = $(NIX_CFLAGS) -Wall -laws-cpp-sdk-s3

src/hydra-queue-runner/binary-cache-store.cc

@@ -7,6 +7,8 @@
 #include "nar-info.hh"
 #include "worker-protocol.hh"
 
+#include <chrono>
+
 namespace nix {
 
 BinaryCacheStore::BinaryCacheStore(ref<Store> localStore,
@@ -50,15 +52,19 @@ void BinaryCacheStore::addToCache(const ValidPathInfo & info,
     if (info.narHash.type != htUnknown && info.narHash != narInfo.narHash)
         throw Error(format("refusing to copy corrupted path ‘%1%’ to binary cache") % info.path);
 
-    printMsg(lvlTalkative, format("copying path ‘%1%’ (%2% bytes) to binary cache")
-        % info.path % info.narSize);
-
     /* Compress the NAR. */
     narInfo.compression = "xz";
+    auto now1 = std::chrono::steady_clock::now();
     string narXz = compressXZ(nar);
+    auto now2 = std::chrono::steady_clock::now();
     narInfo.fileHash = hashString(htSHA256, narXz);
     narInfo.fileSize = narXz.size();
 
+    printMsg(lvlTalkative, format("copying path ‘%1%’ (%2% bytes, compressed %3$.1f%% in %4% ms) to binary cache")
+        % info.path % info.narSize
+        % ((1.0 - (double) narXz.size() / nar.size()) * 100.0)
+        % std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count());
+
     /* Atomically write the NAR file. */
     narInfo.url = "nar/" + printHash32(narInfo.fileHash) + ".nar.xz";
     if (!fileExists(narInfo.url)) upsertFile(narInfo.url, narXz);

src/hydra-queue-runner/hydra-queue-runner.cc

@@ -8,6 +8,7 @@
 #include "state.hh"
 #include "build-result.hh"
 #include "local-binary-cache-store.hh"
+#include "s3-binary-cache-store.hh"
 
 #include "shared.hh"
 #include "globals.hh"
@@ -33,10 +34,16 @@ ref<Store> State::getLocalStore()
 
 ref<Store> State::getDestStore()
 {
+#if 0
     auto store = make_ref<LocalBinaryCacheStore>(getLocalStore(),
-        "/tmp/binary-cache",
         "/home/eelco/Misc/Keys/test.nixos.org/secret",
-        "/home/eelco/Misc/Keys/test.nixos.org/public");
+        "/home/eelco/Misc/Keys/test.nixos.org/public",
+        "/tmp/binary-cache");
+#endif
+    auto store = make_ref<S3BinaryCacheStore>(getLocalStore(),
+        "/home/eelco/Misc/Keys/test.nixos.org/secret",
+        "/home/eelco/Misc/Keys/test.nixos.org/public",
+        "nix-test-cache-3");
     store->init();
     return store;
 }

src/hydra-queue-runner/local-binary-cache-store.cc

@@ -3,7 +3,8 @@
 namespace nix {
 
 LocalBinaryCacheStore::LocalBinaryCacheStore(ref<Store> localStore,
-    const Path & binaryCacheDir,  const Path & secretKeyFile, const Path & publicKeyFile)
+    const Path & secretKeyFile, const Path & publicKeyFile,
+    const Path & binaryCacheDir)
     : BinaryCacheStore(localStore, secretKeyFile, publicKeyFile)
     , binaryCacheDir(binaryCacheDir)
 {

src/hydra-queue-runner/local-binary-cache-store.hh

@@ -12,8 +12,9 @@ private:
 
 public:
 
-    LocalBinaryCacheStore(ref<Store> localStore, const Path & binaryCacheDir,
+    LocalBinaryCacheStore(ref<Store> localStore,
-        const Path & secretKeyFile, const Path & publicKeyFile);
+        const Path & secretKeyFile, const Path & publicKeyFile,
+        const Path & binaryCacheDir);
 
     void init() override;
 

src/hydra-queue-runner/s3-binary-cache-store.cc

@@ -0,0 +1,134 @@
+#include "s3-binary-cache-store.hh"
+
+#include <aws/core/client/ClientConfiguration.h>
+#include <aws/s3/S3Client.h>
+#include <aws/s3/model/CreateBucketRequest.h>
+#include <aws/s3/model/GetBucketLocationRequest.h>
+#include <aws/s3/model/GetObjectRequest.h>
+#include <aws/s3/model/HeadObjectRequest.h>
+#include <aws/s3/model/PutObjectRequest.h>
+
+namespace nix {
+
+/* Helper: given an Outcome<R, E>, return R in case of success, or
+   throw an exception in case of an error. */
+template<typename R, typename E>
+R && checkAws(Aws::Utils::Outcome<R, E> && outcome)
+{
+    if (!outcome.IsSuccess())
+        throw Error(format("AWS error: %1%") % outcome.GetError().GetMessage());
+    return outcome.GetResultWithOwnership();
+}
+
+S3BinaryCacheStore::S3BinaryCacheStore(ref<Store> localStore,
+    const Path & secretKeyFile, const Path & publicKeyFile,
+    const std::string & bucketName)
+    : BinaryCacheStore(localStore, secretKeyFile, publicKeyFile)
+    , bucketName(bucketName)
+    , config(makeConfig())
+    , client(make_ref<Aws::S3::S3Client>(*config))
+{
+}
+
+ref<Aws::Client::ClientConfiguration> S3BinaryCacheStore::makeConfig()
+{
+    auto res = make_ref<Aws::Client::ClientConfiguration>();
+    res->region = Aws::Region::EU_WEST_1;
+    res->requestTimeoutMs = 600 * 1000;
+    return res;
+}
+
+void S3BinaryCacheStore::init()
+{
+    /* Create the bucket if it doesn't already exists. */
+    // FIXME: HeadBucket would be more appropriate, but doesn't return
+    // an easily parsed 404 message.
+    auto res = client->GetBucketLocation(
+        Aws::S3::Model::GetBucketLocationRequest().WithBucket(bucketName));
+
+    if (!res.IsSuccess()) {
+        if (res.GetError().GetErrorType() != Aws::S3::S3Errors::NO_SUCH_BUCKET)
+            throw Error(format("AWS error: %1%") % res.GetError().GetMessage());
+
+        checkAws(client->CreateBucket(
+            Aws::S3::Model::CreateBucketRequest()
+            .WithBucket(bucketName)
+            .WithCreateBucketConfiguration(
+                Aws::S3::Model::CreateBucketConfiguration()
+                .WithLocationConstraint(
+                    Aws::S3::Model::BucketLocationConstraint::eu_west_1))));
+    }
+
+    BinaryCacheStore::init();
+}
+
+bool S3BinaryCacheStore::fileExists(const std::string & path)
+{
+    auto res = client->HeadObject(
+        Aws::S3::Model::HeadObjectRequest()
+        .WithBucket(bucketName)
+        .WithKey(path));
+
+    if (!res.IsSuccess()) {
+        auto & error = res.GetError();
+        if (error.GetErrorType() == Aws::S3::S3Errors::UNKNOWN // FIXME
+            && error.GetMessage().find("404") != std::string::npos)
+            return false;
+        throw Error(format("AWS error: %1%") % error.GetMessage());
+    }
+
+    return true;
+}
+
+void S3BinaryCacheStore::upsertFile(const std::string & path, const std::string & data)
+{
+    auto request =
+        Aws::S3::Model::PutObjectRequest()
+        .WithBucket(bucketName)
+        .WithKey(path);
+
+    auto stream = std::make_shared<std::stringstream>(data);
+
+    request.SetBody(stream);
+
+    auto now1 = std::chrono::steady_clock::now();
+
+    auto result = checkAws(client->PutObject(request));
+
+    auto now2 = std::chrono::steady_clock::now();
+
+    printMsg(lvlError, format("uploaded ‘s3://%1%/%2%’ (%3% bytes) in %4% ms")
+        % bucketName % path
+        % data.size()
+        % std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count());
+}
+
+std::string S3BinaryCacheStore::getFile(const std::string & path)
+{
+    auto request =
+        Aws::S3::Model::GetObjectRequest()
+        .WithBucket(bucketName)
+        .WithKey(path);
+
+    request.SetResponseStreamFactory([&]() {
+        return Aws::New<std::stringstream>("STRINGSTREAM");
+    });
+
+    auto now1 = std::chrono::steady_clock::now();
+
+    auto result = checkAws(client->GetObject(request));
+
+    auto now2 = std::chrono::steady_clock::now();
+
+    auto res = dynamic_cast<std::stringstream &>(result.GetBody()).str();
+
+    printMsg(lvlError, format("downloaded ‘s3://%1%/%2%’ (%3%) in %4% ms")
+        % bucketName % path
+        % res.size()
+        % std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count());
+
+    return res;
+}
+
+}
+

src/hydra-queue-runner/s3-binary-cache-store.hh

@@ -0,0 +1,41 @@
+#pragma once
+
+#include "binary-cache-store.hh"
+
+namespace Aws { namespace Client { class ClientConfiguration; } }
+namespace Aws { namespace S3 { class S3Client; } }
+
+namespace nix {
+
+class S3BinaryCacheStore : public BinaryCacheStore
+{
+private:
+
+    std::string bucketName;
+
+    ref<Aws::Client::ClientConfiguration> config;
+    ref<Aws::S3::S3Client> client;
+
+public:
+
+    S3BinaryCacheStore(ref<Store> localStore,
+        const Path & secretKeyFile, const Path & publicKeyFile,
+        const std::string & bucketName);
+
+    void init() override;
+
+private:
+
+    ref<Aws::Client::ClientConfiguration> makeConfig();
+
+protected:
+
+    bool fileExists(const std::string & path) override;
+
+    void upsertFile(const std::string & path, const std::string & data) override;
+
+    std::string getFile(const std::string & path) override;
+
+};
+
+}