golang 操作ceph object storage

ceph的object storage 提供了和amazon s3兼容的接口以供客户访问。

在ceph的官网上，可以看到它提供了多种语言的访问范本，例如python的(http://docs.ceph.com/docs/master/radosgw/s3/python/)：

import boto

import boto.s3.connection

access_key = 'abc'

secret_key = 'cba'

conn = boto.connect_s3(

        aws_access_key_id = access_key,

        aws_secret_access_key = secret_key,

        host = '1.2.3.4', port=,

        is_secure=False,

        calling_format = boto.s3.connection.OrdinaryCallingFormat(),

)

#bucket = conn.create_bucket('my-first-s3-bucket')

bucket = conn.get_bucket('my-first-s3-bucket')

print "{name}".format(

            name = bucket.name,

#            created = bucket.creation_date,

            )

for obj in bucket.get_all_keys():

    print(obj.key)

key = bucket.new_key('auto')

key.set_contents_from_filename('/tmp/test.yaml')

但是找不到golang版本的，研究了很久，才在网上找到golang版本的example，链接:https://my.oschina.net/u/578775/blog/863370

package main

import (

    "bufio"

    "bytes"

    "errors"

    "fmt"

    "github.com/astaxie/beego"

    "github.com/aws/aws-sdk-go/aws"

    "github.com/aws/aws-sdk-go/aws/credentials"

    "github.com/aws/aws-sdk-go/aws/session"

    "github.com/aws/aws-sdk-go/service/s3"

    "io"

    "os"

)

type CephMgmt struct {

    host        string `ceph host`

    DoName      string `ceph doname`

    bucket_id   string `bucket_id`

    PathStyle   bool   `ceph url style, true means you can use host directly,  false means bucket_id.doname will be used`

    AccessKey   string `aws s3 aceessKey`

    SecretKey   string `aws s3 secretKey`

    block_size int64  `block_size`

}

var Ceph CephMgmt

type MyProvider struct{}

func (m *MyProvider) Retrieve() (credentials.Value, error) {

    return credentials.Value{

        AccessKeyID:     Ceph.AccessKey, //"9YGYCFH1V3QUT4B9KZD1",

        SecretAccessKey: Ceph.SecretKey, //"9hdiwSjCfyrZPKGObH8Kctdur8PiBPJu3B4zGYaZ",

    }, nil

}

func (m *MyProvider) IsExpired() bool { return false }

func (this *CephMgmt) Init() error {

    // "http://s3.devopscloud.com"

    this.host = beego.AppConfig.String("ceph::host")

    if len(this.host) <=  {

        return errors.New("ceph conf host is nil")

    }

    this.bucket_id = beego.AppConfig.String("ceph::bucket_id")

    if len(this.bucket_id) <=  {

        return errors.New("ceph conf bucket_id is nil")

    }

    this.block_size, _ = beego.AppConfig.Int64("ceph::block_size")

    if this.block_size <= ** {

        this.block_size =  *  *

    }

    return nil

}

func (this *CephMgmt) Init2(host string, bucket string, accesskey string, secretkey string) error {

    this.host = host

    this.bucket_id = bucket

    this.AccessKey = accesskey

    this.SecretKey = secretkey

    this.PathStyle = true

    this.block_size =  *  * 

    return nil

}

func (this *CephMgmt) connect() (*s3.S3, error) {

    ///////////////创建连接

    sess := session.Must(session.NewSessionWithOptions(session.Options{

        Config: aws.Config{

            Region: aws.String("default"), //Required 目前尚未分区，填写default即可

            //EndpointResolver: endpoints.ResolverFunc(s3CustResolverFn),

            Endpoint:         &this.host,

            S3ForcePathStyle: &this.PathStyle,

            Credentials:      credentials.NewCredentials(&MyProvider{}),

        },

    }))

    // Create the S3 service client with the shared session. This will

    // automatically use the S3 custom endpoint configured in the custom

    // endpoint resolver wrapping the default endpoint resolver.

    return s3.New(sess), nil

}

func (this *CephMgmt) Download(src_name, dst_name string) error {

    s3Svc, _ := this.connect()

    // Operation calls will be made to the custom endpoint.

    resp, err := s3Svc.GetObject(&s3.GetObjectInput{

        Bucket: &this.bucket_id, //Required 可认为是cpid，也可认为是对于根目录而言的第一级目录.必须先创建，在存储。

        Key:    &src_name,       //Required 文件名，中间可以带着路径，格式如：{path}/{filename}

        //Range:  aws.String("bytes=0-499"),             //not must be Required 文件范围，如果没有则是全文件

    })

    if err != nil {

        // Print the error, cast err to awserr.Error to get the Code and

        // Message from an error.

        fmt.Println("hello , i met error")

        fmt.Println(err.Error())

        return err

    }

    fmt.Println("AcceptRanges :", *resp.AcceptRanges)

    fmt.Println("ContentLength:", *resp.ContentLength)

    //fmt.Println("ContentRange :", *resp.ContentRange)

    //fmt.Println("PartsCount   :", *resp.PartsCount)

    // Pretty-print the response data.

    //fmt.Println(resp)

    //数据都在resp.Body中，可以转储到文件，具体操作参考go的操作

    out, err := os.OpenFile(dst_name, os.O_CREATE|os.O_RDWR, )

    if out == nil {

        fmt.Println("Open fail")

        return err

    }

    num, err := io.Copy(out, resp.Body)

    fmt.Printf("\n write %d err %v \n", num, err)

    return nil

}

func (this *CephMgmt) Upload(src_name, dst_name string) error {

    s3Svc, _ := this.connect()

    dst_full_name := fmt.Sprintf("/%s/%s", this.bucket_id, dst_name) //Notice: 如果要讲桶ID拼到全路径下，一定要在最前面加'/'

    //////////////////////////////////////// 存入 /////////////////////////////

    /////////////// 创建一个分片上传context

    param_init := &s3.CreateMultipartUploadInput{

        Bucket: aws.String(this.bucket_id), // Required {bucket}

        Key:    aws.String(dst_full_name),  // Required /{bucket}/{path}/{filename}

        /*

            Metadata: map[string]*string{

                "Key": aws.String("lasttime_if_need"), // Required 可以填充一些文件属性,ceph云存储不关心其中的内容

            },

        */

    }

    resp_init, err := s3Svc.CreateMultipartUpload(param_init)

    if err != nil {

        // Print the error, cast err to awserr.Error to get the Code and

        // Message from an error.

        fmt.Println("I am create upload.", err.Error())

        return err

    }

    // Pretty-print the response data.

    up_id := resp_init.UploadId

    fmt.Println(resp_init) // 需要保存该resp , resp.UploadId 是对象网关创建的标识

    ///////////////////////// 分片上传

    f, err := os.Open(src_name)

    if err != nil {

        return err

    }

    defer f.Close()

    bfRd := bufio.NewReader(f)

    buf := make([]byte, this.block_size) //TODO:一次读取多少个字节

    var part_num int64 =

    var completes []*s3.CompletedPart

    for {

        n, err := bfRd.Read(buf)

        if err == io.EOF {

            fmt.Println("read data finished")

            break

        }

        if int64(n) != this.block_size {

            data := make([]byte, n)

            data = buf[:n]

            buf = data

        }

        //////////////////////// 每次上传一个分片，每次的PartNumber都要唯一

        part_num++

        param := &s3.UploadPartInput{

            Bucket:        aws.String(this.bucket_id), // Required bucket

            Key:           aws.String(dst_name),       // Required {path}/{filename}

            PartNumber:    aws.Int64(part_num),        // Required 每次的序号唯一且递增

            UploadId:      up_id,                      // Required 创建context时返回的值

            Body:          bytes.NewReader(buf),       // Required 数据内容

            ContentLength: aws.Int64(int64(n)),        // Required 数据长度

        }

        resp2, err := s3Svc.UploadPart(param)

        if err != nil {

            fmt.Printf("Hello ,i am wrong[%s][%d][%d]\n", dst_name, part_num, n)

            return err

        }

        fmt.Println(resp2) // 需要保存该resp，因为resp.Etag 需要在通知完成上传时使用

        var c s3.CompletedPart

        c.PartNumber = aws.Int64(part_num) // Required Etag对应的PartNumber, 上一步返回的

        c.ETag = resp2.ETag                // Required 上传分片时返回的值 Etag

        completes = append(completes, &c)

    }

    /////////////////////////// 结束上传

    params := &s3.CompleteMultipartUploadInput{

        Bucket:   aws.String(this.bucket_id), // Required {bucket}

        Key:      aws.String(dst_name),       // Required {path}/{filename}

        UploadId: up_id,                      // Required 创建context时返回的值

        MultipartUpload: &s3.CompletedMultipartUpload{

            Parts: completes,

        },

        //RequestPayer: aws.String("RequestPayer"),

    }

    resp_comp, err := s3Svc.CompleteMultipartUpload(params)

    if err != nil {

        // Print the error, cast err to awserr.Error to get the Code and

        // Message from an error.

        fmt.Println(err.Error())

        return err

    }

    // Pretty-print the response data.

    fmt.Println(resp_comp)

    return nil

}

func main() {

    //既可以使用域名，也可以使用ip+端口的形式

    Ceph.Init2( /*"http://172.16.20.15"*/ "http://s3.devopscloud.com:80", "new-bucket-5de4d39a", "9YGYCFH1V3QUT4B9KZD1", "9hdiwSjCfyrZPKGObH8Kctdur8PiBPJu3B4zGYaZ")

    Ceph.Download("1.mp4", "./1.mp4")

    Ceph.Upload("1.mp4", "api/rest/2017-3-16-10-38.mp4")

}

巴特西

golang 操作ceph object storage

最新文章

热门文章