netcdf-c/libnczarr/zs3sdk.cpp

555 lines
16 KiB
C++
Raw Normal View History

This PR adds EXPERIMENTAL support for accessing data in the cloud using a variant of the Zarr protocol and storage format. This enhancement is generically referred to as "NCZarr". The data model supported by NCZarr is netcdf-4 minus the user-defined types and the String type. In this sense it is similar to the CDF-5 data model. More detailed information about enabling and using NCZarr is described in the document NUG/nczarr.md and in a [Unidata Developer's blog entry](https://www.unidata.ucar.edu/blogs/developer/en/entry/overview-of-zarr-support-in). WARNING: this code has had limited testing, so do use this version for production work. Also, performance improvements are ongoing. Note especially the following platform matrix of successful tests: Platform | Build System | S3 support ------------------------------------ Linux+gcc | Automake | yes Linux+gcc | CMake | yes Visual Studio | CMake | no Additionally, and as a consequence of the addition of NCZarr, major changes have been made to the Filter API. NOTE: NCZarr does not yet support filters, but these changes are enablers for that support in the future. Note that it is possible (probable?) that there will be some accidental reversions if the changes here did not correctly mimic the existing filter testing. In any case, previously filter ids and parameters were of type unsigned int. In order to support the more general zarr filter model, this was all converted to char*. The old HDF5-specific, unsigned int operations are still supported but they are wrappers around the new, char* based nc_filterx_XXX functions. This entailed at least the following changes: 1. Added the files libdispatch/dfilterx.c and include/ncfilter.h 2. Some filterx utilities have been moved to libdispatch/daux.c 3. A new entry, "filter_actions" was added to the NCDispatch table and the version bumped. 4. An overly complex set of structs was created to support funnelling all of the filterx operations thru a single dispatch "filter_actions" entry. 5. Move common code to from libhdf5 to libsrc4 so that it is accessible to nczarr. Changes directly related to Zarr: 1. Modified CMakeList.txt and configure.ac to support both C and C++ -- this is in support of S3 support via the awd-sdk libraries. 2. Define a size64_t type to support nczarr. 3. More reworking of libdispatch/dinfermodel.c to support zarr and to regularize the structure of the fragments section of a URL. Changes not directly related to Zarr: 1. Make client-side filter registration be conditional, with default off. 2. Hack include/nc4internal.h to make some flags added by Ed be unique: e.g. NC_CREAT, NC_INDEF, etc. 3. cleanup include/nchttp.h and libdispatch/dhttp.c. 4. Misc. changes to support compiling under Visual Studio including: * Better testing under windows for dirent.h and opendir and closedir. 5. Misc. changes to the oc2 code to support various libcurl CURLOPT flags and to centralize error reporting. 6. By default, suppress the vlen tests that have unfixed memory leaks; add option to enable them. 7. Make part of the nc_test/test_byterange.sh test be contingent on remotetest.unidata.ucar.edu being accessible. Changes Left TO-DO: 1. fix provenance code, it is too HDF5 specific.
2020-06-29 08:02:47 +08:00
/*
* Copyright 2018, University Corporation for Atmospheric Research
* See netcdf/COPYRIGHT file for copying and redistribution conditions.
*/
#define NOOP
This PR adds EXPERIMENTAL support for accessing data in the cloud using a variant of the Zarr protocol and storage format. This enhancement is generically referred to as "NCZarr". The data model supported by NCZarr is netcdf-4 minus the user-defined types and the String type. In this sense it is similar to the CDF-5 data model. More detailed information about enabling and using NCZarr is described in the document NUG/nczarr.md and in a [Unidata Developer's blog entry](https://www.unidata.ucar.edu/blogs/developer/en/entry/overview-of-zarr-support-in). WARNING: this code has had limited testing, so do use this version for production work. Also, performance improvements are ongoing. Note especially the following platform matrix of successful tests: Platform | Build System | S3 support ------------------------------------ Linux+gcc | Automake | yes Linux+gcc | CMake | yes Visual Studio | CMake | no Additionally, and as a consequence of the addition of NCZarr, major changes have been made to the Filter API. NOTE: NCZarr does not yet support filters, but these changes are enablers for that support in the future. Note that it is possible (probable?) that there will be some accidental reversions if the changes here did not correctly mimic the existing filter testing. In any case, previously filter ids and parameters were of type unsigned int. In order to support the more general zarr filter model, this was all converted to char*. The old HDF5-specific, unsigned int operations are still supported but they are wrappers around the new, char* based nc_filterx_XXX functions. This entailed at least the following changes: 1. Added the files libdispatch/dfilterx.c and include/ncfilter.h 2. Some filterx utilities have been moved to libdispatch/daux.c 3. A new entry, "filter_actions" was added to the NCDispatch table and the version bumped. 4. An overly complex set of structs was created to support funnelling all of the filterx operations thru a single dispatch "filter_actions" entry. 5. Move common code to from libhdf5 to libsrc4 so that it is accessible to nczarr. Changes directly related to Zarr: 1. Modified CMakeList.txt and configure.ac to support both C and C++ -- this is in support of S3 support via the awd-sdk libraries. 2. Define a size64_t type to support nczarr. 3. More reworking of libdispatch/dinfermodel.c to support zarr and to regularize the structure of the fragments section of a URL. Changes not directly related to Zarr: 1. Make client-side filter registration be conditional, with default off. 2. Hack include/nc4internal.h to make some flags added by Ed be unique: e.g. NC_CREAT, NC_INDEF, etc. 3. cleanup include/nchttp.h and libdispatch/dhttp.c. 4. Misc. changes to support compiling under Visual Studio including: * Better testing under windows for dirent.h and opendir and closedir. 5. Misc. changes to the oc2 code to support various libcurl CURLOPT flags and to centralize error reporting. 6. By default, suppress the vlen tests that have unfixed memory leaks; add option to enable them. 7. Make part of the nc_test/test_byterange.sh test be contingent on remotetest.unidata.ucar.edu being accessible. Changes Left TO-DO: 1. fix provenance code, it is too HDF5 specific.
2020-06-29 08:02:47 +08:00
#include "awsincludes.h"
#include <stdlib.h>
#include <string.h>
#include <iostream>
#include <streambuf>
#include "netcdf.h"
#include "zs3sdk.h"
#define size64_t unsigned long long
static Aws::SDKOptions zs3options;
/* Forward */
static Aws::S3::Model::BucketLocationConstraint s3findregion(const char* name);
static int s3objectinfo1(const Aws::S3::Model::Object& s3_object, char** fullkeyp, size64_t* lenp);
static int s3objectsinfo(Aws::Vector<Aws::S3::Model::Object> list, size_t*, char*** keysp, size64_t** lenp);
static void freestringenvv(char** ss);
static int makes3key(const char* pathkey, const char** keyp);
#ifndef nullfree
#define nullfree(x) {if(x) {free(x);}}
#endif
void
NCZ_s3sdkinitialize(void)
{
// zs3options.loggingOptions.logLevel = Aws::Utils::Logging::LogLevel::Error;
Aws::InitAPI(zs3options);
}
void
NCZ_s3sdkfinalize(void)
{
Aws::ShutdownAPI(zs3options);
}
static char*
makeerrmsg(const Aws::Client::AWSError<Aws::S3::S3Errors> err, const char* key="")
{
char* errmsg;
size_t len;
len = strlen(err.GetExceptionName().c_str()) + strlen(err.GetMessage().c_str()) + strlen(key) + 10;
if((errmsg = (char*)malloc(len+1))==NULL)
return NULL;
snprintf(errmsg,len,"%s %s key=%s",
err.GetExceptionName().c_str(),
err.GetMessage().c_str(),
key);
return errmsg;
}
int
NCZ_s3sdkcreateconfig(const char* host, const char* region, void** configp)
{
int stat = NC_NOERR;
Aws::Client::ClientConfiguration *config = new Aws::Client::ClientConfiguration();
config->scheme = Aws::Http::Scheme::HTTPS;
config->connectTimeoutMs = 30000;
config->requestTimeoutMs = 600000;
if(region) config->region = region;
if(host) config->endpointOverride = host;
config->enableEndpointDiscovery = true;
config->followRedirects = true;
if(configp) * configp = config;
return stat;
}
int
NCZ_s3sdkcreateclient(void* config0, void** clientp)
{
Aws::Client::ClientConfiguration* config = (Aws::Client::ClientConfiguration*) config0;
Aws::S3::S3Client *s3client
= new Aws::S3::S3Client(*config,
Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::RequestDependent,
false,
Aws::S3::US_EAST_1_REGIONAL_ENDPOINT_OPTION::NOT_SET
);
if(clientp) *clientp = (void*)s3client;
return NC_NOERR;
}
int
NCZ_s3sdkbucketexists(void* s3client0, const char* bucket, int* existsp, char** errmsgp)
{
int stat = NC_NOERR;
int exists = 0;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
if(errmsgp) *errmsgp = NULL;
auto result = s3client->ListBuckets();
if(!result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(result.GetError());
stat = NC_ES3;
} else {
Aws::Vector<Aws::S3::Model::Bucket> bucket_list = result.GetResult().GetBuckets();
for(auto const &awsbucket : bucket_list) {
auto name = awsbucket.GetName();
if(name == bucket) {exists = 1; break;}
}
}
if(existsp) *existsp = exists;
return stat;
}
int
NCZ_s3sdkbucketcreate(void* s3client0, const char* region, const char* bucket, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
if(errmsgp) *errmsgp = NULL;
const Aws::S3::Model::BucketLocationConstraint &awsregion = s3findregion(region);
if(awsregion == Aws::S3::Model::BucketLocationConstraint::NOT_SET)
return NC_EURL;
/* Set up the request */
Aws::S3::Model::CreateBucketRequest create_request;
create_request.SetBucket(bucket);
if(region) {
/* Specify the region as a location constraint */
Aws::S3::Model::CreateBucketConfiguration bucket_config;
bucket_config.SetLocationConstraint(awsregion);
create_request.SetCreateBucketConfiguration(bucket_config);
}
#ifdef NOOP
/* Create the bucket */
auto create_result = s3client->CreateBucket(create_request);
if(!create_result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(create_result.GetError());
stat = NC_ES3;;
}
#else
fprintf(stderr,"create bucket: %s\n",bucket); fflush(stderr);
#endif
return stat;
}
int
NCZ_s3sdkbucketdelete(void* s3client0, void* config0, const char* region, const char* bucket, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::Client::ClientConfiguration *config = (Aws::Client::ClientConfiguration*)config0;
if(errmsgp) *errmsgp = NULL;
const Aws::S3::Model::BucketLocationConstraint &awsregion = s3findregion(region);
if(awsregion == Aws::S3::Model::BucketLocationConstraint::NOT_SET)
return NC_EURL;
/* Set up the request */
Aws::S3::Model::DeleteBucketRequest request;
request.SetBucket(bucket);
if(region) {
config->region = region; // Will this work?
}
#ifdef NOOP
/* Delete the bucket */
auto result = s3client->DeleteBucket(request);
if(!result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(result.GetError());
stat = NC_ES3;;
}
#else
fprintf(stderr,"delete bucket: %s\n",bucket); fflush(stderr);
#endif
return stat;
}
/**************************************************/
/* Object API */
/*
@return NC_NOERR if key points to a content-bearing object.
@return NC_EEMPTY if object at key has no content.
@return NC_EXXX return true error
*/
int
NCZ_s3sdkinfo(void* s3client0, const char* bucket, const char* pathkey, size64_t* lenp, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::S3::Model::HeadObjectRequest head_request;
const char* key = NULL;
if(*pathkey != '/') return NC_EINTERNAL;
/* extract the true s3 key*/
if((stat = makes3key(pathkey,&key))) return stat;
if(errmsgp) *errmsgp = NULL;
head_request.SetBucket(bucket);
head_request.SetKey(key);
auto head_outcome = s3client->HeadObject(head_request);
if(head_outcome.IsSuccess()) {
long long l = head_outcome.GetResult().GetContentLength();
if(lenp) *lenp = (size64_t)l;
} else {
/* Distinquish not-found from other errors */
switch (head_outcome.GetError().GetErrorType()) {
case Aws::S3::S3Errors::RESOURCE_NOT_FOUND:
stat = NC_EEMPTY;
break;
case Aws::S3::S3Errors::ACCESS_DENIED:
stat = NC_EACCESS;
/* fall thru */
default:
if(!stat) stat = NC_ES3;
if(errmsgp) *errmsgp = makeerrmsg(head_outcome.GetError(),key);
break;
}
}
return (stat);
}
/* Define object key
@return NC_NOERR if success
@return NC_EXXX if fail
*/
int
NCZ_s3sdkcreatekey(void* s3client0, const char* bucket, const char* pathkey, char** errmsgp)
{
int stat = NC_NOERR;
const char* key = NULL;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::S3::Model::PutObjectRequest put_request;
if(*pathkey != '/') return NC_EINTERNAL;
if((stat = makes3key(pathkey,&key))) return stat;
if(errmsgp) *errmsgp = NULL;
put_request.SetBucket(bucket);
put_request.SetKey(key);
auto put_result = s3client->PutObject(put_request);
if(!put_result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(put_result.GetError(),key);
stat = NC_ES3;
}
return (stat);
}
/*
@return NC_NOERR if success
@return NC_EXXX if fail
*/
int
NCZ_s3sdkread(void* s3client0, const char* bucket, const char* pathkey, size64_t start, size64_t count, void* content, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::S3::Model::GetObjectRequest object_request;
char range[1024];
const char* key = NULL;
if(*pathkey != '/') return NC_EINTERNAL;
if((stat = makes3key(pathkey,&key))) return stat;
object_request.SetBucket(bucket);
object_request.SetKey(key);
snprintf(range,sizeof(range),"bytes=%llu-%llu",start,(start+count)-1);
object_request.SetRange(range);
auto get_object_result = s3client->GetObject(object_request);
if(!get_object_result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(get_object_result.GetError(),key);
stat = NC_ES3;
} else {
/* Get the whole result */
Aws::IOStream &result = get_object_result.GetResultWithOwnership().GetBody();
std::string str((std::istreambuf_iterator<char>(result)),std::istreambuf_iterator<char>());
/* Verify actual result size */
size_t slen = str.size();
if(slen > count) return NC_ES3;
const char* s = str.c_str();
if(content)
memcpy(content,s,slen);
}
return (stat);
}
/*
For S3, I can see no way to do a byterange write;
so we are effectively writing the whole object
*/
int
NCZ_s3sdkwriteobject(void* s3client0, const char* bucket, const char* pathkey, size64_t count, const void* content, char** errmsgp)
{
int stat = NC_NOERR;
const char* key = NULL;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::S3::Model::PutObjectRequest put_request;
if(*pathkey != '/') return NC_EINTERNAL;
if((stat = makes3key(pathkey,&key))) return stat;
if(errmsgp) *errmsgp = NULL;
put_request.SetBucket(bucket);
put_request.SetKey(key);
put_request.SetContentLength((long long)count);
std::shared_ptr<Aws::IOStream> data = std::shared_ptr<Aws::IOStream>(new Aws::StringStream());
data->rdbuf()->pubsetbuf((char*)content,count);
put_request.SetBody(data);
auto put_result = s3client->PutObject(put_request);
if(!put_result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(put_result.GetError(),key);
stat = NC_ES3;
}
return (stat);
}
#if 0
/*
@return NC_NOERR if success
@return NC_EXXX if fail
*/
int
NCZ_s3sdkreadobject(void* s3client0, const char* bucket, const char* pathkey, size64_t* sizep, void** contentp, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::S3::Model::GetObjectRequest object_request;
size64_t size, red;
void* content = NULL;
const char* key = NULL;
if(*pathkey != '/') return NC_EINTERNAL;
if((stat = makes3key(pathkey,&key))) return stat;
if(errmsgp) *errmsgp = NULL;
object_request.SetBucket(bucket);
object_request.SetKey(key);
auto get_result = s3client->GetObject(object_request);
if(!get_result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(get_result.GetError(),key);
stat = NC_ES3;
} else {
/* Get the size */
size = (size64_t)get_result.GetResult().GetContentLength();
/* Get the whole result */
Aws::IOStream &result = get_result.GetResultWithOwnership().GetBody();
#if 0
std::string str((std::istreambuf_iterator<char>(result)),std::istreambuf_iterator<char>());
red = str.length();
if((content = malloc(red))==NULL)
{stat = NC_ENOMEM; goto done;}
memcpy(content,str.c_str(),red);
#else
red = result.rdbuf()->pubseekoff(0,std::ios::ios_base::end);
result.rdbuf()->pubseekoff(0,std::ios::ios_base::beg); /* reset for reading */
if((content = malloc(red))==NULL)
stat = NC_ENOMEM;
else
result.rdbuf()->sgetn((char*)content,red);
#endif
if(!stat) {
/* Verify actual result size */
if(red != size) {stat = NC_ES3; goto done;}
if(sizep) *sizep = red;
if(contentp) {*contentp = content; content = NULL;}
}
}
done:
nullfree(content);
return (stat);
}
#endif /*0*/
int
NCZ_s3sdkclose(void* s3client0, void* config0, const char* bucket, const char* rootkey, int deleteit, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::Client::ClientConfiguration *config = (Aws::Client::ClientConfiguration*)config0;
if(deleteit) {
/* Delete the root key; ok it if does not exist */
switch (stat = NCZ_s3sdkdeletekey(s3client0,bucket,rootkey,errmsgp)) {
case NC_NOERR: break;
case NC_EEMPTY: case NC_ENOTFOUND: stat = NC_NOERR; break;
default: break;
}
}
delete s3client;
delete config;
return (stat);
}
/*
Return a list of keys for objects "below" a specified rootkey.
In theory, the returned list should be sorted in lexical order,
but it possible that it is not.
*/
int
NCZ_s3sdkgetkeys(void* s3client0, const char* bucket, const char* rootkey0, size_t* nkeysp, char*** keysp, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
size_t nkeys = 0;
const char* rootkey = NULL;
Aws::S3::Model::ListObjectsV2Request objects_request;
if(*rootkey0 != '/') return NC_EINTERNAL;
if((stat = makes3key(rootkey0,&rootkey))) return stat;
if(errmsgp) *errmsgp = NULL;
objects_request.SetBucket(bucket);
objects_request.SetPrefix(rootkey);
auto objects_outcome = s3client->ListObjectsV2(objects_request);
if(objects_outcome.IsSuccess()) {
Aws::Vector<Aws::S3::Model::Object> object_list =
objects_outcome.GetResult().GetContents();
nkeys = (size_t)object_list.size();
if(nkeysp) *nkeysp = nkeys;
stat = s3objectsinfo(object_list,NULL,keysp,NULL);
} else {
if(errmsgp) *errmsgp = makeerrmsg(objects_outcome.GetError());
stat = NC_ES3;
}
return stat;
}
int
NCZ_s3sdkdeletekey(void* s3client0, const char* bucket, const char* pathkey, char** errmsgp)
{
int stat = NC_NOERR;
Aws::S3::S3Client* s3client = (Aws::S3::S3Client*)s3client0;
Aws::S3::Model::DeleteObjectRequest delete_request;
const char* key = NULL;
if(*pathkey != '/') return NC_EINTERNAL;
if((stat = makes3key(pathkey,&key))) return stat;
/* Delete this key object */
delete_request.SetBucket(bucket);
delete_request.SetKey(key);
auto delete_result = s3client->DeleteObject(delete_request);
if(!delete_result.IsSuccess()) {
if(errmsgp) *errmsgp = makeerrmsg(delete_result.GetError(),key);
stat = NC_ES3;
}
return stat;
}
/*
Get Info about a single object from a vector
*/
static int
s3objectinfo1(const Aws::S3::Model::Object& s3_object, char** fullkeyp, size64_t* lenp)
{
int stat = NC_NOERR;
char* cstr = NULL;
if(fullkeyp) {
const char* key = NULL;
char* cstr = NULL;
auto s = s3_object.GetKey();
key = s.c_str();
if(key[0] == '/') {
cstr = strdup(key);
} else {
size_t len = strlen(key);
cstr = (char*)malloc(len+1+1);
if(cstr != NULL) {
cstr[0] = '/';
memcpy((void*)(cstr+1),(void*)key,len);
cstr[len+1] = '\0';
}
}
if(cstr == NULL) {
stat = NC_ENOMEM;
} else if(fullkeyp) {
*fullkeyp = cstr;
cstr = NULL;
}
}
if(!stat) {
if(lenp) *lenp = (size64_t)s3_object.GetSize();
}
nullfree(cstr);
return stat;
}
/*
Get Info about a vector of objects; Keys are fixed up to start with a '/'
*/
static int
s3objectsinfo(Aws::Vector<Aws::S3::Model::Object> list, size_t* nkeysp, char*** keysp, size64_t** lenp)
{
int stat = NC_NOERR;
char** keys = NULL;
size_t nkeys;
size64_t *lengths = NULL;
int i;
nkeys = list.size();
if(nkeysp) *nkeysp = nkeys;
if((keys=(char**)calloc(sizeof(char*),(nkeys+1)))==NULL)
stat = NC_ENOMEM;
if(!stat) {
if((lengths=(size64_t*)calloc(sizeof(size64_t),(nkeys)))==NULL)
stat = NC_ENOMEM;
}
if(!stat) {
i = 0;
for (auto const &s3_object : list) {
stat = s3objectinfo1(s3_object,&keys[i],&lengths[i]);
i++;
if(stat) break;
}
}
if(!stat) {
if(keysp) {keys[nkeys] = NULL; *keysp = keys; keys = NULL;}
if(lenp) {*lenp = lengths; lengths = NULL;}
}
if(keys != NULL) freestringenvv(keys);
if(lengths != NULL) free(lengths);
return stat;
}
static Aws::S3::Model::BucketLocationConstraint
s3findregion(const char* name)
{
return Aws::S3::Model::BucketLocationConstraintMapper::GetBucketLocationConstraintForName(name);
}
struct MemBuf: std::streambuf {
MemBuf(char* base, size_t size) {
char* p(base);
this->setg(p, p, p + size);
}
};
static void
freestringenvv(char** ss)
{
char** p;
if(ss != NULL) {
for(p=ss;*p;p++)
nullfree(*p);
free(ss);
}
}
static int
makes3key(const char* pathkey, const char** keyp)
{
assert(pathkey != NULL && pathkey[0] == '/');
if(keyp) *keyp = pathkey+1;
return NC_NOERR;
}