s3

Documentation

Index

• Constants
• Variables
• func GenerateInventory(ctx context.Context, logger logging.Logger, manifestURL string, ...) (block.Inventory, error)
• func NewAdapter(s3 s3iface.S3API, opts ...func(a *Adapter)) block.Adapter
• func ReadAllWithTimeout(r io.Reader, buf []byte, timeout time.Duration) (n int, err error)
• func WithContext(ctx context.Context) func(a *Adapter)
• func WithHTTPClient(c *http.Client) func(a *Adapter)
• func WithStreamingChunkSize(sz int) func(a *Adapter)
• func WithStreamingChunkTimeout(d time.Duration) func(a *Adapter)
• func WithTranslator(t block.UploadIDTranslator) func(a *Adapter)
• type Adapter
  • func (s *Adapter) AbortMultiPartUpload(obj block.ObjectPointer, uploadID string) error
  • func (s *Adapter) CompleteMultiPartUpload(obj block.ObjectPointer, uploadID string, ...) (*string, int64, error)
  • func (s *Adapter) CreateMultiPartUpload(obj block.ObjectPointer, r *http.Request, opts block.CreateMultiPartUploadOpts) (string, error)
  • func (s *Adapter) GenerateInventory(ctx context.Context, logger logging.Logger, manifestURL string) (block.Inventory, error)
  • func (s *Adapter) Get(obj block.ObjectPointer, _ int64) (io.ReadCloser, error)
  • func (s *Adapter) GetProperties(obj block.ObjectPointer) (block.Properties, error)
  • func (s *Adapter) GetRange(obj block.ObjectPointer, startPosition int64, endPosition int64) (io.ReadCloser, error)
  • func (s *Adapter) Put(obj block.ObjectPointer, sizeBytes int64, reader io.Reader, opts block.PutOpts) error
  • func (s *Adapter) PutWithoutStream(obj block.ObjectPointer, sizeBytes int64, reader io.Reader, opts block.PutOpts) error
  • func (s *Adapter) Remove(obj block.ObjectPointer) error
  • func (s *Adapter) UploadPart(obj block.ObjectPointer, sizeBytes int64, reader io.Reader, uploadID string, ...) (string, error)
  • func (s *Adapter) ValidateConfiguration(storageNamespace string) error
  • func (s *Adapter) WithContext(ctx context.Context) block.Adapter
• type CloseFunc
• type Inventory
  • func (inv *Inventory) InventoryURL() string
  • func (inv *Inventory) Iterator() block.InventoryIterator
  • func (inv *Inventory) SourceName() string
• type InventoryIterator
  • func NewInventoryIterator(inv *Inventory) *InventoryIterator
  • func (it *InventoryIterator) Err() error
  • func (it *InventoryIterator) Get() *block.InventoryObject
  • func (it *InventoryIterator) Next() bool
• type ParquetInventoryObject
  • func (o *ParquetInventoryObject) GetPhysicalAddress() string
• type ParquetReader
• type StreamingReader
  • func (s *StreamingReader) GetLastChunk() []byte
  • func (s *StreamingReader) Read(p []byte) (int, error)

Constants

const (
	BlockstoreType = "s3"

	DefaultStreamingChunkSize    = 2 << 19         // 1MiB by default per chunk
	DefaultStreamingChunkTimeout = time.Second * 1 // if we haven't read DefaultStreamingChunkSize by this duration, write whatever we have as a chunk

	ExpireObjectS3Tag = "lakefs_expire_object"
)

const (
	StreamingSha256          = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"
	StreamingContentEncoding = "aws-chunked"
)

const DefaultReadBatchSize = 100000

Variables

var ErrMissingETag = errors.New("missing ETag")

var ErrReaderTimeout = errors.New("reader timeout")

Functions

func GenerateInventory

func GenerateInventory(ctx context.Context, logger logging.Logger, manifestURL string, s3 s3iface.S3API, getParquetReader parquetReaderGetter) (block.Inventory, error)

func NewAdapter

func NewAdapter(s3 s3iface.S3API, opts ...func(a *Adapter)) block.Adapter

func ReadAllWithTimeout

func ReadAllWithTimeout(r io.Reader, buf []byte, timeout time.Duration) (n int, err error)

ReadAllWithTimeout is taken from io.ReadAtLeast and adapted to support a timeout

func WithContext

func WithContext(ctx context.Context) func(a *Adapter)

func WithHTTPClient

func WithHTTPClient(c *http.Client) func(a *Adapter)

func WithStreamingChunkSize

func WithStreamingChunkSize(sz int) func(a *Adapter)

func WithStreamingChunkTimeout

func WithStreamingChunkTimeout(d time.Duration) func(a *Adapter)

func WithTranslator

func WithTranslator(t block.UploadIDTranslator) func(a *Adapter)

Types

type Adapter

type Adapter struct {
	// contains filtered or unexported fields
}

func (*Adapter) AbortMultiPartUpload

func (s *Adapter) AbortMultiPartUpload(obj block.ObjectPointer, uploadID string) error

func (*Adapter) CompleteMultiPartUpload

func (s *Adapter) CompleteMultiPartUpload(obj block.ObjectPointer, uploadID string, multipartList *block.MultipartUploadCompletion) (*string, int64, error)

func (*Adapter) CreateMultiPartUpload

func (s *Adapter) CreateMultiPartUpload(obj block.ObjectPointer, r *http.Request, opts block.CreateMultiPartUploadOpts) (string, error)

func (*Adapter) GenerateInventory

func (s *Adapter) GenerateInventory(ctx context.Context, logger logging.Logger, manifestURL string) (block.Inventory, error)

func (*Adapter) Get

func (s *Adapter) Get(obj block.ObjectPointer, _ int64) (io.ReadCloser, error)

func (*Adapter) GetProperties

func (s *Adapter) GetProperties(obj block.ObjectPointer) (block.Properties, error)

func (*Adapter) GetRange

func (s *Adapter) GetRange(obj block.ObjectPointer, startPosition int64, endPosition int64) (io.ReadCloser, error)

func (*Adapter) Put

func (s *Adapter) Put(obj block.ObjectPointer, sizeBytes int64, reader io.Reader, opts block.PutOpts) error

func (*Adapter) PutWithoutStream

func (s *Adapter) PutWithoutStream(obj block.ObjectPointer, sizeBytes int64, reader io.Reader, opts block.PutOpts) error

work around, because put failed with trying to create symlinks

func (*Adapter) Remove

func (s *Adapter) Remove(obj block.ObjectPointer) error

func (*Adapter) UploadPart

func (s *Adapter) UploadPart(obj block.ObjectPointer, sizeBytes int64, reader io.Reader, uploadID string, partNumber int64) (string, error)

func (*Adapter) ValidateConfiguration

func (s *Adapter) ValidateConfiguration(storageNamespace string) error

ValidateConfiguration on an S3 adapter checks for a usable bucket lifecycle policy: the storageNamespace bucket should expire objects marked with ExpireObjectS3Tag (with _some_ duration, even if nonzero).

func (*Adapter) WithContext

func (s *Adapter) WithContext(ctx context.Context) block.Adapter

type CloseFunc

type CloseFunc func() error

type Inventory

type Inventory struct {
	S3       s3iface.S3API
	Manifest *manifest
	// contains filtered or unexported fields
}

func (*Inventory) InventoryURL

func (inv *Inventory) InventoryURL() string

func (*Inventory) Iterator

func (inv *Inventory) Iterator() block.InventoryIterator

func (*Inventory) SourceName

func (inv *Inventory) SourceName() string

type InventoryIterator

type InventoryIterator struct {
	*Inventory
	ReadBatchSize int
	// contains filtered or unexported fields
}

func NewInventoryIterator

func NewInventoryIterator(inv *Inventory) *InventoryIterator

func (*InventoryIterator) Err

func (it *InventoryIterator) Err() error

func (*InventoryIterator) Get

func (it *InventoryIterator) Get() *block.InventoryObject

func (*InventoryIterator) Next

func (it *InventoryIterator) Next() bool

type ParquetInventoryObject

type ParquetInventoryObject struct {
	Bucket         string  `parquet:"name=bucket, type=UTF8"`
	Key            string  `parquet:"name=key, type=UTF8"`
	IsLatest       *bool   `parquet:"name=is_latest, type=BOOLEAN"`
	IsDeleteMarker *bool   `parquet:"name=is_delete_marker, type=BOOLEAN"`
	Size           *int64  `parquet:"name=size, type=INT_64"`
	LastModified   *int64  `parquet:"name=last_modified_date, type=TIMESTAMP_MILLIS"`
	Checksum       *string `parquet:"name=e_tag, type=UTF8"`
}

func (*ParquetInventoryObject) GetPhysicalAddress

func (o *ParquetInventoryObject) GetPhysicalAddress() string

type ParquetReader

type ParquetReader interface {
	Read(dstInterface interface{}) error
	GetNumRows() int64
	SkipRows(int64) error
}

type StreamingReader

type StreamingReader struct {
	Reader       io.Reader
	Size         int
	StreamSigner *v4.StreamSigner
	Time         time.Time
	ChunkSize    int
	ChunkTimeout time.Duration
	// contains filtered or unexported fields
}

func (*StreamingReader) GetLastChunk

func (s *StreamingReader) GetLastChunk() []byte

func (*StreamingReader) Read

func (s *StreamingReader) Read(p []byte) (int, error)