refactor(providers): Create default provider that has common functions (#3241)

* Rename UnsupportedBlobRetention struct

Rename this struct to DefaultProviderImplementation in preparation for
adding other simple "default" functionality to it.

* Add other functions to default provider

Add other simple function implementations to the default provider so
that other providers can just embed this to get basic behavior.

* Cleanup existing users of default provider

* Add default provider to remaining storage types

Add the default provider to remaining storage providers and remove
functions that are now implemented by the default provider.
This commit is contained in:
ashmrtn authored and GitHub committed 2023-08-21 15:27:15 -07:00
1 parent 08c58d53b6
commit 29320a7c33
13 files changed
+39 -143

No files matched your search

+1 -17
View File
@@ -18,7 +18,7 @@ import (
type DataMap map[blob.ID][]byte
type mapStorage struct {
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
// +checklocks:mutex
data DataMap
// +checklocks:mutex
@@ -28,14 +28,6 @@ type mapStorage struct {
mutex sync.RWMutex
}
func (s *mapStorage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
return blob.Capacity{}, blob.ErrNotAVolume
}
func (s *mapStorage) IsReadOnly() bool {
return false
}
func (s *mapStorage) GetBlob(ctx context.Context, id blob.ID, offset, length int64, output blob.OutputBuffer) error {
s.mutex.RLock()
defer s.mutex.RUnlock()
@@ -166,10 +158,6 @@ func (s *mapStorage) ListBlobs(ctx context.Context, prefix blob.ID, callback fun
return nil
}
func (s *mapStorage) Close(ctx context.Context) error {
return nil
}
func (s *mapStorage) TouchBlob(ctx context.Context, blobID blob.ID, threshold time.Duration) (time.Time, error) {
s.mutex.Lock()
defer s.mutex.Unlock()
@@ -193,10 +181,6 @@ func (s *mapStorage) DisplayName() string {
return "Map"
}
func (s *mapStorage) FlushCaches(ctx context.Context) error {
return nil
}
// NewMapStorage returns an implementation of Storage backed by the contents of given map.
// Used primarily for testing.
func NewMapStorage(data DataMap, keyTime map[blob.ID]time.Time, timeNow func() time.Time) blob.Storage {
+1 -18
View File
@@ -32,6 +32,7 @@ type versionedEntries map[blob.ID][]*entry
// marker. This struct manages the retention time of each blob through the
// PutBlob options.
type objectLockingMap struct {
blob.DefaultProviderImplementation
// +checklocks:mutex
data versionedEntries
timeNow func() time.Time // +checklocksignore
@@ -69,14 +70,6 @@ func (s *objectLockingMap) getLatestForMutationLocked(id blob.ID) (*entry, error
return e, nil
}
func (s *objectLockingMap) GetCapacity(ctx context.Context) (blob.Capacity, error) {
return blob.Capacity{}, blob.ErrNotAVolume
}
func (s *objectLockingMap) IsReadOnly() bool {
return false
}
// GetBlob works the same as map-storage GetBlob except that if the latest
// version is a delete-marker then it will return ErrBlobNotFound.
func (s *objectLockingMap) GetBlob(ctx context.Context, id blob.ID, offset, length int64, output blob.OutputBuffer) error {
@@ -266,11 +259,6 @@ func (s *objectLockingMap) ListBlobs(ctx context.Context, prefix blob.ID, callba
return nil
}
// Close is a no-op for this implementation.
func (s *objectLockingMap) Close(ctx context.Context) error {
return nil
}
// TouchBlob updates the mtime of the latest version of the object. If it is a
// delete-marker or if it does not exist then this becomes a no-op. If the
// latest version has retention parameters set then they are respected.
@@ -309,11 +297,6 @@ func (s *objectLockingMap) DisplayName() string {
return "VersionedMap"
}
// FlushCaches is a no-op for this implementation.
func (s *objectLockingMap) FlushCaches(ctx context.Context) error {
return nil
}
// NewVersionedMapStorage returns an implementation of Storage backed by the
// contents of an internal in-memory map used primarily for testing.
func NewVersionedMapStorage(timeNow func() time.Time) RetentionStorage {
+1 -17
View File
@@ -28,20 +28,12 @@ const (
type azStorage struct {
Options
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
service *azblob.Client
container string
}
func (az *azStorage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
return blob.Capacity{}, blob.ErrNotAVolume
}
func (az *azStorage) IsReadOnly() bool {
return false
}
func (az *azStorage) GetBlob(ctx context.Context, b blob.ID, offset, length int64, output blob.OutputBuffer) error {
if offset < 0 {
return errors.Wrap(blob.ErrInvalidRange, "invalid offset")
@@ -234,14 +226,6 @@ func (az *azStorage) DisplayName() string {
return fmt.Sprintf("Azure: %v", az.Options.Container)
}
func (az *azStorage) Close(ctx context.Context) error {
return nil
}
func (az *azStorage) FlushCaches(ctx context.Context) error {
return nil
}
// New creates new Azure Blob Storage-backed storage with specified options:
//
// - the 'Container', 'StorageAccount' and 'StorageKey' fields are required and all other parameters are optional.
+1 -17
View File
@@ -26,20 +26,12 @@ const (
type b2Storage struct {
Options
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
cli *backblaze.B2
bucket *backblaze.Bucket
}
func (s *b2Storage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
return blob.Capacity{}, blob.ErrNotAVolume
}
func (s *b2Storage) IsReadOnly() bool {
return false
}
func (s *b2Storage) GetBlob(ctx context.Context, id blob.ID, offset, length int64, output blob.OutputBuffer) error {
fileName := s.getObjectNameString(id)
@@ -251,14 +243,6 @@ func (s *b2Storage) DisplayName() string {
return fmt.Sprintf("B2: %v", s.BucketName)
}
func (s *b2Storage) Close(ctx context.Context) error {
return nil
}
func (s *b2Storage) FlushCaches(ctx context.Context) error {
return nil
}
func (s *b2Storage) String() string {
return fmt.Sprintf("b2://%s/%s", s.BucketName, s.Prefix)
}
+1 -9
View File
@@ -34,7 +34,7 @@ const (
type fsStorage struct {
sharded.Storage
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
}
type fsImpl struct {
@@ -336,14 +336,6 @@ func (fs *fsStorage) DisplayName() string {
return fmt.Sprintf("Filesystem: %v", fs.RootPath)
}
func (fs *fsStorage) Close(ctx context.Context) error {
return nil
}
func (fs *fsStorage) FlushCaches(ctx context.Context) error {
return nil
}
// New creates new filesystem-backed storage in a specified directory.
func New(ctx context.Context, opts *Options, isCreate bool) (blob.Storage, error) {
var err error
+1 -13
View File
@@ -32,20 +32,12 @@ const (
type gcsStorage struct {
Options
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
storageClient *gcsclient.Client
bucket *gcsclient.BucketHandle
}
func (gcs *gcsStorage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
return blob.Capacity{}, blob.ErrNotAVolume
}
func (gcs *gcsStorage) IsReadOnly() bool {
return false
}
func (gcs *gcsStorage) GetBlob(ctx context.Context, b blob.ID, offset, length int64, output blob.OutputBuffer) error {
if offset < 0 {
return blob.ErrInvalidRange
@@ -213,10 +205,6 @@ func (gcs *gcsStorage) Close(ctx context.Context) error {
return errors.Wrap(gcs.storageClient.Close(), "error closing GCS storage")
}
func (gcs *gcsStorage) FlushCaches(ctx context.Context) error {
return nil
}
func tokenSourceFromCredentialsFile(ctx context.Context, fn string, scopes ...string) (oauth2.TokenSource, error) {
data, err := os.ReadFile(fn) //nolint:gosec
if err != nil {
+1 -9
View File
@@ -45,7 +45,7 @@ var log = logging.Module("gdrive")
type gdriveStorage struct {
Options
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
client *drive.FilesService
about *drive.AboutService
@@ -73,10 +73,6 @@ func (gdrive *gdriveStorage) GetCapacity(ctx context.Context) (blob.Capacity, er
}, nil
}
func (gdrive *gdriveStorage) IsReadOnly() bool {
return false
}
func (gdrive *gdriveStorage) GetBlob(ctx context.Context, b blob.ID, offset, length int64, output blob.OutputBuffer) error {
if offset < 0 {
return blob.ErrInvalidRange
@@ -330,10 +326,6 @@ func (gdrive *gdriveStorage) DisplayName() string {
return fmt.Sprintf("Google Drive: %v", gdrive.folderID)
}
func (gdrive *gdriveStorage) Close(ctx context.Context) error {
return nil
}
func (gdrive *gdriveStorage) FlushCaches(ctx context.Context) error {
gdrive.fileIDCache.Clear()
return nil
+1 -1
View File
@@ -15,7 +15,7 @@ var ErrReadonly = errors.Errorf("storage is read-only")
// readonlyStorage prevents all mutations on the underlying storage.
type readonlyStorage struct {
base blob.Storage
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
}
func (s readonlyStorage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
+1 -16
View File
@@ -30,20 +30,13 @@ const (
type s3Storage struct {
Options
blob.DefaultProviderImplementation
cli *minio.Client
storageConfig *StorageConfig
}
func (s *s3Storage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
return blob.Capacity{}, blob.ErrNotAVolume
}
func (s *s3Storage) IsReadOnly() bool {
return false
}
func (s *s3Storage) GetBlob(ctx context.Context, b blob.ID, offset, length int64, output blob.OutputBuffer) error {
return s.getBlobWithVersion(ctx, b, latestVersionID, offset, length, output)
}
@@ -300,10 +293,6 @@ func (s *s3Storage) ConnectionInfo() blob.ConnectionInfo {
}
}
func (s *s3Storage) Close(ctx context.Context) error {
return nil
}
func (s *s3Storage) String() string {
return fmt.Sprintf("s3://%v/%v", s.BucketName, s.Prefix)
}
@@ -312,10 +301,6 @@ func (s *s3Storage) DisplayName() string {
return fmt.Sprintf("S3: %v %v", s.Endpoint, s.BucketName)
}
func (s *s3Storage) FlushCaches(ctx context.Context) error {
return nil
}
func getCustomTransport(opt *Options) (*http.Transport, error) {
if opt.DoNotVerifyTLS {
//nolint:gosec
+1 -5
View File
@@ -40,7 +40,7 @@ const (
// sftpStorage implements blob.Storage on top of sftp.
type sftpStorage struct {
sharded.Storage
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
}
type sftpImpl struct {
@@ -360,10 +360,6 @@ func writeKnownHostsDataStringToTempFile(data string) (string, error) {
return tf.Name(), nil
}
func (s *sftpStorage) FlushCaches(ctx context.Context) error {
return nil
}
// getHostKeyCallback returns a HostKeyCallback that validates the connected host based on KnownHostsFile or KnownHostsData.
func getHostKeyCallback(opt *Options) (ssh.HostKeyCallback, error) {
if opt.KnownHostsData != "" {
-5
View File
@@ -44,11 +44,6 @@ type Storage struct {
parameters *Parameters
}
// IsReadOnly implements blob.Storage.
func (s *Storage) IsReadOnly() bool {
return false
}
// GetBlob implements blob.Storage.
func (s *Storage) GetBlob(ctx context.Context, blobID blob.ID, offset, length int64, output blob.OutputBuffer) error {
dirPath, filePath, err := s.GetShardedPathAndFilePath(ctx, blobID)
+28 -3
View File
@@ -138,14 +138,39 @@ type ExtendOptions struct {
RetentionPeriod time.Duration
}
// UnsupportedBlobRetention provides a default implementation for ExtendBlobRetention.
type UnsupportedBlobRetention struct{}
// DefaultProviderImplementation provides a default implementation for
// common functions that are mostly provider independent and have a sensible
// default.
//
// Storage providers should imbed this struct and override functions that they
// have different return values for.
type DefaultProviderImplementation struct{}
// ExtendBlobRetention provides a common implementation for unsupported blob retention storage.
func (s *UnsupportedBlobRetention) ExtendBlobRetention(context.Context, ID, ExtendOptions) error {
func (s DefaultProviderImplementation) ExtendBlobRetention(context.Context, ID, ExtendOptions) error {
return ErrUnsupportedObjectLock
}
// IsReadOnly complies with the Storage interface.
func (s DefaultProviderImplementation) IsReadOnly() bool {
return false
}
// Close complies with the Storage interface.
func (s DefaultProviderImplementation) Close(context.Context) error {
return nil
}
// FlushCaches complies with the Storage interface.
func (s DefaultProviderImplementation) FlushCaches(context.Context) error {
return nil
}
// GetCapacity complies with the Storage interface.
func (s DefaultProviderImplementation) GetCapacity(context.Context) (Capacity, error) {
return Capacity{}, ErrNotAVolume
}
// HasRetentionOptions returns true when blob-retention settings have been
// specified, otherwise returns false.
func (o PutOptions) HasRetentionOptions() bool {
+1 -13
View File
@@ -36,7 +36,7 @@ const (
// may be accessed using WebDAV or File interchangeably.
type davStorage struct {
sharded.Storage
blob.UnsupportedBlobRetention
blob.DefaultProviderImplementation
}
type davStorageImpl struct {
@@ -45,10 +45,6 @@ type davStorageImpl struct {
cli *gowebdav.Client
}
func (d *davStorage) GetCapacity(ctx context.Context) (blob.Capacity, error) {
return blob.Capacity{}, blob.ErrNotAVolume
}
func (d *davStorageImpl) GetBlobFromPath(ctx context.Context, dirPath, path string, offset, length int64, output blob.OutputBuffer) error {
_ = dirPath
@@ -242,14 +238,6 @@ func (d *davStorage) DisplayName() string {
return fmt.Sprintf("WebDAV: %v", o.URL)
}
func (d *davStorage) Close(ctx context.Context) error {
return nil
}
func (d *davStorage) FlushCaches(ctx context.Context) error {
return nil
}
func isRetriable(err error) bool {
var pe *os.PathError