k8s-csi-s3/pkg/s3/s3-client.go

168 lines
4 KiB
Go
Raw Normal View History

2018-07-14 08:48:22 +00:00
package s3
import (
"bytes"
2021-01-15 03:19:59 +00:00
"context"
"encoding/json"
2018-07-14 08:48:22 +00:00
"fmt"
"io"
2018-07-14 08:48:22 +00:00
"net/url"
"github.com/golang/glog"
2021-01-15 03:19:59 +00:00
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
2018-07-14 08:48:22 +00:00
)
const (
metadataName = ".metadata.json"
defaultFsPrefix = "csi-fs"
2018-07-14 08:48:22 +00:00
)
type s3Client struct {
2018-07-16 18:24:54 +00:00
cfg *Config
2018-07-14 08:48:22 +00:00
minio *minio.Client
2021-01-15 03:19:59 +00:00
ctx context.Context
2018-07-14 08:48:22 +00:00
}
type bucket struct {
Name string
Mounter string
FSPath string
CapacityBytes int64
CreatedByCsi bool
}
2018-07-16 18:24:54 +00:00
func newS3Client(cfg *Config) (*s3Client, error) {
2018-07-14 08:48:22 +00:00
var client = &s3Client{}
2018-07-16 18:24:54 +00:00
client.cfg = cfg
u, err := url.Parse(client.cfg.Endpoint)
2018-07-14 08:48:22 +00:00
if err != nil {
return nil, err
}
ssl := u.Scheme == "https"
endpoint := u.Hostname()
if u.Port() != "" {
endpoint = u.Hostname() + ":" + u.Port()
}
2021-01-15 03:19:59 +00:00
minioClient, err := minio.New(endpoint, &minio.Options{
Creds: credentials.NewStaticV4(client.cfg.AccessKeyID, client.cfg.SecretAccessKey, client.cfg.Region),
Secure: ssl,
})
2018-07-14 08:48:22 +00:00
if err != nil {
return nil, err
}
client.minio = minioClient
2021-01-15 03:19:59 +00:00
client.ctx = context.Background()
2018-07-14 08:48:22 +00:00
return client, nil
}
func newS3ClientFromSecrets(secrets map[string]string) (*s3Client, error) {
return newS3Client(&Config{
AccessKeyID: secrets["accessKeyID"],
SecretAccessKey: secrets["secretAccessKey"],
Region: secrets["region"],
Endpoint: secrets["endpoint"],
// Mounter is set in the volume preferences, not secrets
Mounter: "",
})
}
2018-07-14 08:48:22 +00:00
func (client *s3Client) bucketExists(bucketName string) (bool, error) {
2021-01-15 03:19:59 +00:00
return client.minio.BucketExists(client.ctx, bucketName)
2018-07-14 08:48:22 +00:00
}
func (client *s3Client) createBucket(bucketName string) error {
2021-01-15 03:19:59 +00:00
return client.minio.MakeBucket(client.ctx, bucketName, minio.MakeBucketOptions{Region: client.cfg.Region})
2018-07-14 08:48:22 +00:00
}
func (client *s3Client) createPrefix(bucketName string, prefix string) error {
2021-01-15 03:19:59 +00:00
_, err := client.minio.PutObject(client.ctx, bucketName, prefix+"/", bytes.NewReader([]byte("")), 0, minio.PutObjectOptions{})
if err != nil {
return err
}
return nil
}
2018-07-14 08:48:22 +00:00
func (client *s3Client) removeBucket(bucketName string) error {
if err := client.emptyBucket(bucketName); err != nil {
return err
}
2021-01-15 03:19:59 +00:00
return client.minio.RemoveBucket(client.ctx, bucketName)
2018-07-14 08:48:22 +00:00
}
func (client *s3Client) emptyBucket(bucketName string) error {
2021-01-15 03:19:59 +00:00
objectsCh := make(chan minio.ObjectInfo)
2018-07-14 08:48:22 +00:00
var listErr error
go func() {
defer close(objectsCh)
doneCh := make(chan struct{})
defer close(doneCh)
2021-01-15 03:19:59 +00:00
for object := range client.minio.ListObjects(
client.ctx,
bucketName,
minio.ListObjectsOptions{Prefix: "", Recursive: true}) {
2018-07-14 08:48:22 +00:00
if object.Err != nil {
listErr = object.Err
return
}
2021-01-15 03:19:59 +00:00
objectsCh <- object
2018-07-14 08:48:22 +00:00
}
}()
if listErr != nil {
glog.Error("Error listing objects", listErr)
return listErr
}
select {
default:
2021-01-15 03:19:59 +00:00
opts := minio.RemoveObjectsOptions{
GovernanceBypass: true,
}
errorCh := client.minio.RemoveObjects(client.ctx, bucketName, objectsCh, opts)
2018-07-14 08:48:22 +00:00
for e := range errorCh {
glog.Errorf("Failed to remove object %s, error: %s", e.ObjectName, e.Err)
}
if len(errorCh) != 0 {
return fmt.Errorf("Failed to remove all objects of bucket %s", bucketName)
}
}
2019-05-14 19:01:54 +00:00
// ensure our prefix is also removed
return client.minio.RemoveObject(client.ctx, bucketName, defaultFsPrefix, minio.RemoveObjectOptions{})
2018-07-14 08:48:22 +00:00
}
func (client *s3Client) setBucket(bucket *bucket) error {
b := new(bytes.Buffer)
json.NewEncoder(b).Encode(bucket)
opts := minio.PutObjectOptions{ContentType: "application/json"}
2021-01-15 03:19:59 +00:00
_, err := client.minio.PutObject(client.ctx, bucket.Name, metadataName, b, int64(b.Len()), opts)
return err
}
func (client *s3Client) getBucket(bucketName string) (*bucket, error) {
opts := minio.GetObjectOptions{}
2021-01-15 03:19:59 +00:00
obj, err := client.minio.GetObject(client.ctx, bucketName, metadataName, opts)
if err != nil {
return &bucket{}, err
}
objInfo, err := obj.Stat()
if err != nil {
return &bucket{}, err
}
b := make([]byte, objInfo.Size)
_, err = obj.Read(b)
if err != nil && err != io.EOF {
return &bucket{}, err
}
var meta bucket
err = json.Unmarshal(b, &meta)
return &meta, err
}