#include <S3Archive.h>

Inheritance diagram for S3Archive:

Collaboration diagram for S3Archive:

Public Member Functions
	S3Archive (const std::string &url, const bool plain_text)

	S3Archive (const std::string &url, const std::string &s3_access_key, const std::string &s3_secret_key, const std::string &s3_region, const std::string &s3_endpoint, const bool plain_text)

	~S3Archive () override

void	init_for_read () override

const std::vector< std::string > &	get_objkeys ()

const std::string	land (const std::string &objkey, std::exception_ptr &teptr, const bool for_detection)

void	vacuum (const std::string &objkey)

size_t	get_total_file_size () const

Public Member Functions inherited from Archive
	Archive (const std::string url, const bool plain_text)

virtual	~Archive ()

virtual std::string	archive_error (int err)

virtual bool	read_next_header ()

virtual bool	read_data_block (const void *buff, size_t size, int64_t *offset)

virtual int64_t	get_position_compressed () const

virtual int	open ()

virtual int	close ()

virtual ssize_t	read (const void **buff)

const std::string	url_part (const int i)

std::string	entryName ()

Private Attributes
std::string	s3_access_key

std::string	s3_secret_key

std::string	s3_region

std::string	s3_endpoint

std::string	s3_temp_dir

std::string	bucket_name

std::string	prefix_name

std::vector< std::string >	objkeys

std::map< const std::string, const std::string >	file_paths

size_t	total_file_size {0}

Additional Inherited Members
Static Public Member Functions inherited from Archive
static ssize_t	read (struct archive a, void client_data, const void **buff)

static int	open (struct archive a, void client_data)

static int	close (struct archive a, void client_data)

static void	parse_url (const std::string url, std::map< int, std::string > &url_parts)

Protected Attributes inherited from Archive
std::string	url

std::map< int, std::string >	url_parts

archive *	ar = 0

archive_entry *	entry

bool	plain_text

Detailed Description

Definition at line 38 of file S3Archive.h.

Constructor & Destructor Documentation

S3Archive::S3Archive	(	const std::string &	url,
		const bool	plain_text
	)

inline

Definition at line 40 of file S3Archive.h.

References s3_access_key, s3_endpoint, s3_region, and s3_secret_key.

                                                          : Archive(url, plain_text) {
 // init aws api should be singleton because because
 // it's bad to call Aws::InitAPI and Aws::ShutdownAPI
 // multiple times.
 #ifdef HAVE_AWS_S3
     {
       std::unique_lock<std::mutex> lck(awsapi_mtx);
       if (0 == awsapi_count++) {
         Aws::InitAPI(awsapi_options);
       }
     }
 #endif  // HAVE_AWS_S3
 
     // these envs are on server side so are global settings
     // which make few senses in case of private s3 resources
     char* env;
     if (0 != (env = getenv("AWS_REGION"))) {
       s3_region = env;
     }
     if (0 != (env = getenv("AWS_ACCESS_KEY_ID"))) {
       s3_access_key = env;
     }
     if (0 != (env = getenv("AWS_SECRET_ACCESS_KEY"))) {
       s3_secret_key = env;
     }
     if (0 != (env = getenv("AWS_ENDPOINT"))) {
       s3_endpoint = env;
     }
   }

S3Archive::S3Archive	(	const std::string &	url,
		const std::string &	s3_access_key,
		const std::string &	s3_secret_key,
		const std::string &	s3_region,
		const std::string &	s3_endpoint,
		const bool	plain_text
	)

inline

Definition at line 70 of file S3Archive.h.

References s3_access_key, s3_endpoint, s3_region, s3_secret_key, and s3_temp_dir.

       : S3Archive(url, plain_text) {
     this->s3_access_key = s3_access_key;
     this->s3_secret_key = s3_secret_key;
     this->s3_region = s3_region;
     this->s3_endpoint = s3_endpoint;
 
     // this must be local to omnisci_server not client
     // or posix dir path accessible to omnisci_server
     auto env_s3_temp_dir = getenv("TMPDIR");
     s3_temp_dir = env_s3_temp_dir ? env_s3_temp_dir : "/tmp";
   }

S3Archive::~S3Archive ( )

inlineoverride

Definition at line 88 of file S3Archive.h.

                         {
 #ifdef HAVE_AWS_S3
     for (auto& thread : threads) {
       if (thread.joinable()) {
         thread.join();
       }
     }
     std::unique_lock<std::mutex> lck(awsapi_mtx);
     if (0 == --awsapi_count) {
       Aws::ShutdownAPI(awsapi_options);
     }
 #endif  // HAVE_AWS_S3
   }

Member Function Documentation

const std::vector<std::string>& S3Archive::get_objkeys ( )

inline

Definition at line 103 of file S3Archive.h.

References objkeys.

103 { return objkeys; }

S3Archive::objkeys

std::vector< std::string > objkeys

Definition: S3Archive.h:138

size_t S3Archive::get_total_file_size ( ) const

inline

Definition at line 119 of file S3Archive.h.

References total_file_size.

119 { return total_file_size; }

S3Archive::total_file_size

size_t total_file_size

Definition: S3Archive.h:140

void S3Archive::init_for_read ( )

overridevirtual

Reimplemented from Archive.

Definition at line 35 of file S3Archive.cpp.

References bucket_name, logger::INFO, LOG, objkeys, prefix_name, s3_access_key, s3_endpoint, s3_region, s3_secret_key, s3_temp_dir, total_file_size, Archive::url, and Archive::url_part().

                               {
   boost::filesystem::create_directories(s3_temp_dir);
   if (!boost::filesystem::is_directory(s3_temp_dir)) {
     throw std::runtime_error("failed to create s3_temp_dir directory '" + s3_temp_dir +
                              "'");
   }
 
   try {
     bucket_name = url_part(4);
     prefix_name = url_part(5);
 
     // a prefix '/obj/' should become 'obj/'
     // a prefix '/obj'  should become 'obj'
     if (prefix_name.size() && '/' == prefix_name.front()) {
       prefix_name = prefix_name.substr(1);
     }
 
     Aws::S3::Model::ListObjectsV2Request objects_request;
     objects_request.WithBucket(bucket_name);
     objects_request.WithPrefix(prefix_name);
     objects_request.SetMaxKeys(1 << 20);
 
     // for a daemon like omnisci_server it seems improper to set s3 credentials
     // via AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY env's because that way
     // credentials are configured *globally* while different users with private
     // s3 resources may need separate credentials to access.in that case, use
     // WITH s3_access_key/s3_secret_key parameters.
     Aws::Client::ClientConfiguration s3_config;
     s3_config.region = s3_region.size() ? s3_region : Aws::Region::US_EAST_1;
     s3_config.endpointOverride = s3_endpoint;
 
     /*
        Fix a wrong ca path established at building libcurl on Centos being carried to
        Ubuntu. To fix the issue, this is this sequence of locating ca file: 1) if
        `SSL_CERT_DIR` or `SSL_CERT_FILE` is set, set it to S3 ClientConfiguration. 2) if
        none ^ is set, omnisci_server searches a list of known ca file paths. 3) if 2)
        finds nothing, it is users' call to set correct SSL_CERT_DIR or SSL_CERT_FILE. S3
        c++ sdk: "we only want to override the default path if someone has explicitly told
        us to."
      */
     std::list<std::string> v_known_ca_paths({
         "/etc/ssl/certs/ca-certificates.crt",
         "/etc/pki/tls/certs/ca-bundle.crt",
         "/usr/share/ssl/certs/ca-bundle.crt",
         "/usr/local/share/certs/ca-root.crt",
         "/etc/ssl/cert.pem",
         "/etc/ssl/ca-bundle.pem",
     });
     char* env;
     if (nullptr != (env = getenv("SSL_CERT_DIR"))) {
       s3_config.caPath = env;
     }
     if (nullptr != (env = getenv("SSL_CERT_FILE"))) {
       v_known_ca_paths.push_front(env);
     }
     for (const auto& known_ca_path : v_known_ca_paths) {
       if (boost::filesystem::exists(known_ca_path)) {
         s3_config.caFile = known_ca_path;
         break;
       }
     }
 
     if (!s3_access_key.empty() && !s3_secret_key.empty()) {
       s3_client.reset(new Aws::S3::S3Client(
           Aws::Auth::AWSCredentials(s3_access_key, s3_secret_key), s3_config));
     } else {
       s3_client.reset(new Aws::S3::S3Client(
           std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>(), s3_config));
     }
     while (true) {
       auto list_objects_outcome = s3_client->ListObjectsV2(objects_request);
       if (list_objects_outcome.IsSuccess()) {
         // pass only object keys to next stage, which may be Importer::import_parquet,
         // Importer::import_compressed or else, depending on copy_params (eg. .is_parquet)
         auto object_list = list_objects_outcome.GetResult().GetContents();
         if (0 == object_list.size()) {
           if (objkeys.empty()) {
             throw std::runtime_error("no object was found with s3 url '" + url + "'");
           }
         }
 
         LOG(INFO) << "Found " << (objkeys.empty() ? "" : "another ") << object_list.size()
                   << " objects with url '" + url + "':";
         for (auto const& obj : object_list) {
           std::string objkey = obj.GetKey().c_str();
           LOG(INFO) << "\t" << objkey << " (size = " << obj.GetSize() << " bytes)";
           total_file_size += obj.GetSize();
           // skip _SUCCESS and keys with trailing / or basename with heading '.'
           boost::filesystem::path path{objkey};
           if (0 == obj.GetSize()) {
             continue;
           }
           if ('/' == objkey.back()) {
             continue;
           }
           if ('.' == path.filename().string().front()) {
             continue;
           }
           objkeys.push_back(objkey);
         }
       } else {
         // could not ListObject
         // could be the object is there but we do not have listObject Privilege
         // We can treat it as a specific object, so should try to parse it and pass to
         // getObject as a singleton
         // Null prefix in urls such like 's3://bucket/' should be ignored.
         if (objkeys.empty()) {
           if (!prefix_name.empty()) {
             objkeys.push_back(prefix_name);
           } else {
             throw std::runtime_error("failed to list objects of s3 url '" + url + "': " +
                                      list_objects_outcome.GetError().GetExceptionName() +
                                      ": " + list_objects_outcome.GetError().GetMessage());
           }
         }
       }
       // continue to read next 1000 files
       if (list_objects_outcome.GetResult().GetIsTruncated()) {
         objects_request.SetContinuationToken(
             list_objects_outcome.GetResult().GetNextContinuationToken());
       } else {
         break;
       }
     }
   } catch (...) {
     throw;
   }
 }

Here is the call graph for this function:

const std::string S3Archive::land	(	const std::string &	objkey,
		std::exception_ptr &	teptr,
		const bool	for_detection
	)

inline

Definition at line 110 of file S3Archive.h.

                                                    {
     throw std::runtime_error("AWS S3 support not available");
   }

void S3Archive::vacuum ( const std::string & objkey )

inline

Definition at line 115 of file S3Archive.h.

                                        {
     throw std::runtime_error("AWS S3 support not available");
   }

Member Data Documentation

std::string S3Archive::bucket_name

private

Definition at line 136 of file S3Archive.h.

Referenced by init_for_read().

std::map<const std::string, const std::string> S3Archive::file_paths

private

Definition at line 139 of file S3Archive.h.

std::vector<std::string> S3Archive::objkeys

private

Definition at line 138 of file S3Archive.h.

Referenced by get_objkeys(), and init_for_read().

std::string S3Archive::prefix_name

private

Definition at line 137 of file S3Archive.h.

Referenced by init_for_read().

std::string S3Archive::s3_access_key

private

Definition at line 130 of file S3Archive.h.

Referenced by init_for_read(), omnisci.thrift.ttypes.TCopyParams::read(), S3Archive(), and omnisci.thrift.ttypes.TCopyParams::write().

std::string S3Archive::s3_endpoint

private

Definition at line 133 of file S3Archive.h.

Referenced by init_for_read(), omnisci.thrift.ttypes.TCopyParams::read(), S3Archive(), and omnisci.thrift.ttypes.TCopyParams::write().

std::string S3Archive::s3_region

private

Definition at line 132 of file S3Archive.h.

Referenced by init_for_read(), omnisci.thrift.ttypes.TCopyParams::read(), S3Archive(), and omnisci.thrift.ttypes.TCopyParams::write().

std::string S3Archive::s3_secret_key

private

Definition at line 131 of file S3Archive.h.

Referenced by init_for_read(), omnisci.thrift.ttypes.TCopyParams::read(), S3Archive(), and omnisci.thrift.ttypes.TCopyParams::write().

std::string S3Archive::s3_temp_dir

private

Definition at line 134 of file S3Archive.h.

Referenced by init_for_read(), and S3Archive().

size_t S3Archive::total_file_size {0}

private

Definition at line 140 of file S3Archive.h.

Referenced by get_total_file_size(), and init_for_read().

The documentation for this class was generated from the following files:

/home/jenkins-slave/workspace/core-os-doxygen/Archive/S3Archive.h
/home/jenkins-slave/workspace/core-os-doxygen/Archive/S3Archive.cpp

Public Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation