From d155d8015578c43953e4a9d1867e49c0b71534d7 Mon Sep 17 00:00:00 2001 From: Eelco Dolstra Date: Thu, 21 Apr 2016 16:02:48 +0200 Subject: [PATCH] Move S3BinaryCacheStore from Hydra This allows running arbitrary Nix commands against an S3 binary cache. To do: make this a compile time option to prevent a dependency on aws-sdk-cpp. --- release.nix | 9 +- src/libstore/local.mk | 2 +- src/libstore/s3-binary-cache-store.cc | 218 ++++++++++++++++++++++++++ src/libstore/s3-binary-cache-store.hh | 34 ++++ 4 files changed, 261 insertions(+), 2 deletions(-) create mode 100644 src/libstore/s3-binary-cache-store.cc create mode 100644 src/libstore/s3-binary-cache-store.hh diff --git a/release.nix b/release.nix index b81571e2d..9d9923afc 100644 --- a/release.nix +++ b/release.nix @@ -71,7 +71,14 @@ let src = tarball; buildInputs = - [ curl perl bzip2 xz openssl pkgconfig sqlite boehmgc ] + [ curl perl bzip2 xz openssl pkgconfig sqlite boehmgc + + (aws-sdk-cpp.override { + apis = ["s3"]; + customMemoryManagement = false; + }) + + ] ++ lib.optional stdenv.isLinux libsodium; configureFlags = '' diff --git a/src/libstore/local.mk b/src/libstore/local.mk index 9a01596c3..15fa91b5c 100644 --- a/src/libstore/local.mk +++ b/src/libstore/local.mk @@ -8,7 +8,7 @@ libstore_SOURCES := $(wildcard $(d)/*.cc) libstore_LIBS = libutil libformat -libstore_LDFLAGS = $(SQLITE3_LIBS) -lbz2 $(LIBCURL_LIBS) $(SODIUM_LIBS) +libstore_LDFLAGS = $(SQLITE3_LIBS) -lbz2 $(LIBCURL_LIBS) $(SODIUM_LIBS) -laws-cpp-sdk-s3 -laws-cpp-sdk-core ifeq ($(OS), SunOS) libstore_LDFLAGS += -lsocket diff --git a/src/libstore/s3-binary-cache-store.cc b/src/libstore/s3-binary-cache-store.cc new file mode 100644 index 000000000..9ac79cf41 --- /dev/null +++ b/src/libstore/s3-binary-cache-store.cc @@ -0,0 +1,218 @@ +#include "s3-binary-cache-store.hh" +#include "nar-info.hh" +#include "nar-info-disk-cache.hh" +#include "globals.hh" + +#include +#include +#include +#include +#include +#include +#include + +namespace nix { + +struct S3Error : public Error +{ + Aws::S3::S3Errors err; + S3Error(Aws::S3::S3Errors err, const FormatOrString & fs) + : Error(fs), err(err) { }; +}; + +/* Helper: given an Outcome, return R in case of success, or + throw an exception in case of an error. */ +template +R && checkAws(const FormatOrString & fs, Aws::Utils::Outcome && outcome) +{ + if (!outcome.IsSuccess()) + throw S3Error( + outcome.GetError().GetErrorType(), + fs.s + ": " + outcome.GetError().GetMessage()); + return outcome.GetResultWithOwnership(); +} + +struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore +{ + std::string bucketName; + + ref config; + ref client; + + Stats stats; + + S3BinaryCacheStoreImpl(std::shared_ptr localStore, + const Path & secretKeyFile, const std::string & bucketName) + : S3BinaryCacheStore(localStore, secretKeyFile) + , bucketName(bucketName) + , config(makeConfig()) + , client(make_ref(*config)) + { + diskCache = getNarInfoDiskCache(); + } + + std::string getUri() + { + return "s3://" + bucketName; + } + + ref makeConfig() + { + auto res = make_ref(); + res->region = Aws::Region::US_EAST_1; // FIXME: make configurable + res->requestTimeoutMs = 600 * 1000; + return res; + } + + void init() + { + if (!diskCache->cacheExists(getUri())) { + + /* 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 checking bucket ‘%s’: %s") % bucketName % res.GetError().GetMessage()); + + checkAws(format("AWS error creating bucket ‘%s’") % bucketName, + client->CreateBucket( + Aws::S3::Model::CreateBucketRequest() + .WithBucket(bucketName) + .WithCreateBucketConfiguration( + Aws::S3::Model::CreateBucketConfiguration() + /* .WithLocationConstraint( + Aws::S3::Model::BucketLocationConstraint::US) */ ))); + } + + BinaryCacheStore::init(); + + diskCache->createCache(getUri()); + } + } + + const Stats & getS3Stats() + { + return stats; + } + + /* This is a specialisation of isValidPath() that optimistically + fetches the .narinfo file, rather than first checking for its + existence via a HEAD request. Since .narinfos are small, doing + a GET is unlikely to be slower than HEAD. */ + bool isValidPathUncached(const Path & storePath) + { + try { + queryPathInfo(storePath); + return true; + } catch (InvalidPath & e) { + return false; + } + } + + bool fileExists(const std::string & path) + { + stats.head++; + + 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 fetching ‘%s’: %s") % path % error.GetMessage()); + } + + return true; + } + + void upsertFile(const std::string & path, const std::string & data) + { + auto request = + Aws::S3::Model::PutObjectRequest() + .WithBucket(bucketName) + .WithKey(path); + + auto stream = std::make_shared(data); + + request.SetBody(stream); + + stats.put++; + stats.putBytes += data.size(); + + auto now1 = std::chrono::steady_clock::now(); + + auto result = checkAws(format("AWS error uploading ‘%s’") % path, + client->PutObject(request)); + + auto now2 = std::chrono::steady_clock::now(); + + auto duration = std::chrono::duration_cast(now2 - now1).count(); + + printMsg(lvlInfo, format("uploaded ‘s3://%1%/%2%’ (%3% bytes) in %4% ms") + % bucketName % path % data.size() % duration); + + stats.putTimeMs += duration; + } + + std::shared_ptr getFile(const std::string & path) + { + printMsg(lvlDebug, format("fetching ‘s3://%1%/%2%’...") % bucketName % path); + + auto request = + Aws::S3::Model::GetObjectRequest() + .WithBucket(bucketName) + .WithKey(path); + + request.SetResponseStreamFactory([&]() { + return Aws::New("STRINGSTREAM"); + }); + + stats.get++; + + try { + + auto now1 = std::chrono::steady_clock::now(); + + auto result = checkAws(format("AWS error fetching ‘%s’") % path, + client->GetObject(request)); + + auto now2 = std::chrono::steady_clock::now(); + + auto res = dynamic_cast(result.GetBody()).str(); + + auto duration = std::chrono::duration_cast(now2 - now1).count(); + + printMsg(lvlTalkative, format("downloaded ‘s3://%1%/%2%’ (%3% bytes) in %4% ms") + % bucketName % path % res.size() % duration); + + stats.getBytes += res.size(); + stats.getTimeMs += duration; + + return std::make_shared(res); + + } catch (S3Error & e) { + if (e.err == Aws::S3::S3Errors::NO_SUCH_KEY) return 0; + throw; + } + } + +}; + +static RegisterStoreImplementation regStore([](const std::string & uri) -> std::shared_ptr { + if (std::string(uri, 0, 5) != "s3://") return 0; + auto store = std::make_shared(std::shared_ptr(0), + settings.get("binary-cache-secret-key-file", string("")), + std::string(uri, 5)); + store->init(); + return store; +}); + +} diff --git a/src/libstore/s3-binary-cache-store.hh b/src/libstore/s3-binary-cache-store.hh new file mode 100644 index 000000000..0425f6bb9 --- /dev/null +++ b/src/libstore/s3-binary-cache-store.hh @@ -0,0 +1,34 @@ +#pragma once + +#include "binary-cache-store.hh" + +#include + +namespace nix { + +class S3BinaryCacheStore : public BinaryCacheStore +{ +protected: + + S3BinaryCacheStore(std::shared_ptr localStore, + const Path & secretKeyFile) + : BinaryCacheStore(localStore, secretKeyFile) + { } + +public: + + struct Stats + { + std::atomic put{0}; + std::atomic putBytes{0}; + std::atomic putTimeMs{0}; + std::atomic get{0}; + std::atomic getBytes{0}; + std::atomic getTimeMs{0}; + std::atomic head{0}; + }; + + const Stats & getS3Stats(); +}; + +}