Cloudpods/backend/pkg/mcclient/modules/compute/mod_buckets.go

// Copyright 2019 Yunion
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package compute

import (
	"context"
	"fmt"
	"io"
	"math/rand"
	"net/http"
	"strconv"
	"time"

	"yunion.io/x/cloudmux/pkg/cloudprovider"
	_ "yunion.io/x/cloudmux/pkg/multicloud/objectstore/provider"
	"yunion.io/x/pkg/errors"
	"yunion.io/x/pkg/util/httputils"

	api "yunion.io/x/onecloud/pkg/apis/compute"
	"yunion.io/x/onecloud/pkg/mcclient"
	"yunion.io/x/onecloud/pkg/mcclient/modulebase"
	"yunion.io/x/onecloud/pkg/mcclient/modules"
)

type SBucketManager struct {
	modulebase.ResourceManager
}

func (manager *SBucketManager) Upload(s *mcclient.ClientSession, bucketId string, key string, body io.Reader, contLength int64, storageClass string, acl string, meta http.Header) error {
	method := httputils.POST
	path := fmt.Sprintf("/%s/%s/upload", manager.URLPath(), bucketId)
	headers := cloudprovider.MetaToHttpHeader(cloudprovider.META_HEADER_PREFIX, meta)
	headers.Set(api.BUCKET_UPLOAD_OBJECT_KEY_HEADER, key)

	if contLength > 0 {
		headers.Set("Content-Length", strconv.FormatInt(contLength, 10))
	}

	if len(storageClass) > 0 {
		headers.Set(api.BUCKET_UPLOAD_OBJECT_STORAGECLASS_HEADER, storageClass)
	}

	if len(acl) > 0 {
		headers.Set(api.BUCKET_UPLOAD_OBJECT_ACL_HEADER, acl)
	}

	resp, err := modulebase.RawRequest(manager.ResourceManager, s, method, path, headers, body)
	if err != nil {
		return errors.Wrap(err, "rawRequest")
	}

	_, _, err = s.ParseJSONResponse("", resp, err)
	if err != nil {
		return err
	}

	return nil
}

var (
	Buckets SBucketManager
)

func init() {
	Buckets = SBucketManager{
		modules.NewComputeManager("bucket", "buckets",
			[]string{"ID", "Name", "Storage_Class",
				"Status", "location", "acl",
				"region", "manager_id", "public_scope",
			},
			[]string{}),
	}

	modules.RegisterCompute(&Buckets)
}

func GetIBucket(ctx context.Context, sess *mcclient.ClientSession, bucketDetails *api.BucketDetails) (cloudprovider.ICloudBucket, error) {
	provider, err := Cloudproviders.GetProvider(ctx, sess, bucketDetails.ManagerId)
	if err != nil {
		return nil, errors.Wrap(err, "computemodules.Cloudproviders.GetProvider")
	}

	iregion, err := func() (cloudprovider.ICloudRegion, error) {
		if provider.GetFactory().IsOnPremise() {
			return provider.GetOnPremiseIRegion()
		} else {
			return provider.GetIRegionById(bucketDetails.RegionExternalId)
		}
	}()
	if err != nil {
		return nil, errors.Wrap(err, "GetIRegion")
	}

	bucket, err := iregion.GetIBucketById(bucketDetails.ExternalId)
	if err != nil {
		return nil, errors.Wrap(err, "iregion.GetIBucketById")
	}

	return bucket, nil
}

type nullWriter struct{}

func (w *nullWriter) WriteAt(p []byte, off int64) (n int, err error) {
	return len(p), nil
}

var (
	randBuffer [1024]byte
)

func init() {
	for i := range randBuffer {
		randBuffer[i] = byte(rand.Intn(256))
	}
}

type randReader struct {
	offset    int64
	sizeBytes int64
}

func newRandReader(sizeBytes int64) io.Reader {
	return &randReader{offset: 0, sizeBytes: sizeBytes}
}

func (r *randReader) Read(p []byte) (n int, err error) {
	if r.offset >= r.sizeBytes {
		return 0, io.EOF
	}
	offset := 0
	for offset < len(p) && r.offset+int64(offset) < r.sizeBytes {
		readLen := len(randBuffer)
		if readLen > len(p)-offset {
			readLen = len(p) - offset
		}
		if readLen > int(r.sizeBytes-r.offset) {
			readLen = int(r.sizeBytes-r.offset) - offset
		}
		copy(p[offset:], randBuffer[:readLen])
		offset += readLen
	}
	r.offset += int64(offset)
	return offset, nil
}

func ProbeBucketStats(ctx context.Context, bucket cloudprovider.ICloudBucket, testKey string, sizeBytes int64) (*api.BucketProbeResult, error) {
	result := &api.BucketProbeResult{}
	start := time.Now()

	// force upload object in one shot
	err := cloudprovider.UploadObject(ctx, bucket, testKey, sizeBytes*2, newRandReader(sizeBytes), sizeBytes, cloudprovider.ACLPrivate, "", nil, false)
	if err != nil {
		return nil, errors.Wrap(err, "cloudprovider.UploadObject")
	}

	result.UploadTime = time.Since(start)

	_, err = cloudprovider.DownloadObjectParallel(ctx, bucket, testKey, nil, &nullWriter{}, 0, 0, false, 1)
	if err != nil {
		return nil, errors.Wrap(err, "cloudprovider.DownloadObjectParallel")
	}

	result.DownloadTime = time.Since(start) - result.UploadTime

	err = bucket.DeleteObject(ctx, testKey)
	if err != nil {
		return nil, errors.Wrap(err, "bucket.DeleteObject")
	}

	result.DeleteTime = time.Since(start) - result.DownloadTime

	return result, nil
}