526互联

go ceph s3文件管理

发布时间 2023-05-24 22:21:38作者: 牛奔

导入依赖

go get gopkg.in/amz.v1/aws
go get gopkg.in/amz.v1/s3

创建用户

在初始化连接之前，我们需要创建一个用户得到accessKey和secretKey，新增用户的指令如下：

docker exec ceph-rgw radosgw-admin user create --uid="test" --display-name="test user"

封装使用

package main

import (
	"gopkg.in/amz.v1/aws"
	"gopkg.in/amz.v1/s3"
	"io/ioutil"
	"log"
)

var cephConn *s3.S3

// GetCephConnection : 获取ceph连接
func GetCephConnection() *s3.S3 {
	if cephConn != nil {
		return cephConn
	}
	// 1. 初始化ceph的一些信息
	auth := aws.Auth{
		AccessKey: "N4GHKXCJNT4S8M0LFSV0",
		SecretKey: "ZrDN1DhWNWo2Hw3JxGk0cmlBfkThu1aXUuzNGQ2g",
	}

	curRegion := aws.Region{
		Name:                 "default",
		EC2Endpoint:          "http://127.0.0.1:7480",
		S3Endpoint:           "http://127.0.0.1:7480",
		S3BucketEndpoint:     "",
		S3LocationConstraint: false,
		S3LowercaseBucket:    false,
		Sign:                 aws.SignV2,
	}

	// 2. 创建S3类型的连接
	return s3.New(auth, curRegion)
}

// GetCephBucket : 获取指定的bucket对象
func GetCephBucket(bucket string) *s3.Bucket {
	conn := GetCephConnection()
	return conn.Bucket(bucket)
}

// 将本地文件上传到ceph的一个bucket中
func put2Bucket(bucket *s3.Bucket, localPath, cephPath string) (*s3.Bucket, error) {
	err := bucket.PutBucket(s3.PublicRead)
	if err != nil {
		log.Fatal(err.Error())
		return nil, err
	}

	bytes, err := ioutil.ReadFile(localPath)
	if err != nil {
		log.Fatal(err.Error())
		return nil, err
	}

	err = bucket.Put(cephPath, bytes, "octet-stream", s3.PublicRead)
	return bucket, err
}

// 从ceph下载文件
func downloadFromCeph(bucket *s3.Bucket, localPath, cephPath string) error {
	data, err := bucket.Get(cephPath)
	if err != nil {
		log.Fatal(err.Error())
		return err
	}
	return ioutil.WriteFile(localPath, data, 0666)
}

// 删除指定的文件
func delCephData(bucket *s3.Bucket, cephPath string) error {
	err := bucket.Del(cephPath)
	if err != nil {
		log.Fatal(err.Error())
	}
	return err
}

// 删除桶，删除桶时要保证桶内文件已经被删除
func delBucket(bucket *s3.Bucket) error {
	err := bucket.DelBucket()
	if err != nil {
		log.Fatal(err.Error())
	}
	return err
}

// 批量获取文件信息
func getBatchFromCeph(bucket *s3.Bucket, prefixCephPath string) []string {
	maxBatch := 100

	// bucket.List() 返回桶内objects的信息，默认1000条
	resultListResp, err := bucket.List(prefixCephPath, "", "", maxBatch)
	if err != nil {
		log.Fatal(err.Error())
		return nil
	}

	keyList := make([]string, 0)
	for _, key := range resultListResp.Contents {
		keyList = append(keyList, key.Key)
	}

	return keyList
}

测试

package main

import (
	"fmt"
	"time"
)

func main() {
	bucketName := "bucket_test"
	filename := "/niuben/ceph_store/1308540.jpg"
	cephPath := "/static/default/bucket_test/V1/" + "1308540.jpg"

	// 获取指定桶
	bucket := GetCephBucket(bucketName)

	// 上传
	bucket, err := put2Bucket(bucket, filename, cephPath)
	if err != nil {
		return
	}

	// 下载
	localPath := "/niuben/ceph_store/1308540_ceph.jpg"
	err = downloadFromCeph(bucket, localPath, cephPath)
	if err != nil {
		return
	}

	// 获得url
	url := bucket.SignedURL(cephPath, time.Now().Add(time.Hour))
	fmt.Println(url)

	// 批量查找
	prefixCephpath := "static/default/bucket_test/V1"
	lists := getBatchFromCeph(bucket, prefixCephpath)
	for _, list := range lists {
		fmt.Println(list)
	}

	// 删除数据
	delCephData(bucket, cephPath)

	// 删除桶
	delBucket(bucket)

}

测试时，可以 go run store.go main.go 运行结果