-
Notifications
You must be signed in to change notification settings - Fork 3
/
aliyun.go
323 lines (280 loc) · 9.74 KB
/
aliyun.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
package aliyun
import (
"encoding/json"
"fmt"
"io"
"sort"
"time"
"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
"github.com/aliyun/alibaba-cloud-sdk-go/services/sts"
alioss "github.com/aliyun/aliyun-oss-go-sdk/oss"
"github.com/pkg/errors"
"github.com/rs/zerolog/log"
"github.com/txchat/dtalk/pkg/oss"
)
var _ oss.Oss = (*Aliyun)(nil)
type Role struct {
RoleSessionName string `json:"roleSessionName"`
RoleArn string `json:"roleArn"`
}
type Aliyun struct {
cfg *oss.Config
client *alioss.Client
}
func New(cfg *oss.Config) *Aliyun {
ali := &Aliyun{
cfg: cfg,
client: nil,
}
// 创建并维护客户端
//go ali.watchClient()
err := ali.createClient()
if err != nil {
panic(err)
}
return ali
}
// Config 返回配置信息
func (ali *Aliyun) Config() *oss.Config {
return ali.cfg
}
// AssumeRole 返回临时授权角色
func (ali *Aliyun) AssumeRole() (*oss.AssumeRoleResp, error) {
//构建一个阿里云客户端, 用于发起请求。
//构建阿里云客户端时,需要设置AccessKey ID和AccessKey Secret。
/*
regionId官方示例就是写死的
https://help.aliyun.com/document_detail/184381.htm?spm=a2c4g.11186623.2.19.25eb4bceQK29Sb#concept-1955433
*/
client, err := sts.NewClientWithAccessKey(ali.cfg.RegionId, ali.cfg.AccessKeyId, ali.cfg.AccessKeySecret)
if err != nil {
return nil, errors.WithMessage(err, "sts.NewClientWithAccessKey()")
}
//构建请求对象。
request := sts.CreateAssumeRoleRequest()
request.Scheme = "https"
var role Role
err = json.Unmarshal([]byte(ali.cfg.Role), &role)
if err != nil {
return nil, errors.WithMessagef(err, "json.Unmarshal role=%+v", role)
}
//设置参数。关于参数含义和设置方法,请参见API参考。
request.RoleArn = role.RoleArn
request.RoleSessionName = role.RoleSessionName
// 设置过期时间
request.DurationSeconds = requests.NewInteger(ali.cfg.DurationSeconds)
//发起请求,并得到响应。
response, err := client.AssumeRole(request)
if err != nil {
return nil, errors.WithMessage(err, "client.AssumeRole()")
}
return &oss.AssumeRoleResp{
RequestId: response.RequestId,
Credentials: oss.Credentials{
AccessKeySecret: response.Credentials.AccessKeySecret,
Expiration: response.Credentials.Expiration,
AccessKeyId: response.Credentials.AccessKeyId,
SecurityToken: response.Credentials.SecurityToken,
},
AssumedRoleUser: oss.AssumedRoleUser{
AssumedRoleId: response.AssumedRoleUser.AssumedRoleId,
Arn: response.AssumedRoleUser.Arn,
},
}, nil
}
// Upload 上传文件
// key string 文件名, 包括路径 "dtalk/2021-07-01/1.jpg"
// body io.Reader 文件内容.
// url string 文件资源链接.
// error it's nil if no error, otherwise it's an error object.
func (ali *Aliyun) Upload(key string, body io.Reader, size int64) (url, uri string, err error) {
bucket, err := ali.client.Bucket(ali.cfg.Bucket)
if err != nil {
return "", "", errors.WithMessage(err, "client.Bucket()")
}
// 指定存储类型为标准存储,缺省也为标准存储。
storageType := alioss.ObjectStorageClass(alioss.StorageStandard)
// 指定存储类型为归档存储。
// storageType := alioss.ObjectStorageClass(alioss.StorageArchive)
// 指定访问权限为公共读,缺省为继承bucket的权限。
objectAcl := alioss.ObjectACL(alioss.ACLPublicRead)
// 上传字符串。
err = bucket.PutObject(key, body, storageType, objectAcl)
if err != nil {
return "", "", errors.WithMessage(err, "bucket.PutObject()")
}
return ali.parseUrl(key), ali.parseUri(key), nil
}
func (ali *Aliyun) createClient() error {
client, err := alioss.New(
ali.cfg.EndPoint,
ali.cfg.AccessKeyId,
ali.cfg.AccessKeySecret,
)
if err != nil {
return errors.WithMessage(err, "alioss.New")
}
ali.client = client
return nil
}
// watchClient 定时更新临时客户端
func (ali *Aliyun) watchClient() {
// 创建一个计时器
durationTime := time.Second * time.Duration(ali.cfg.DurationSeconds)
timeTickerChan := time.Tick(durationTime)
// 更新客户端
for {
if err := ali.createTemporaryClient(); err != nil {
log.Error().Err(err).Msg("create ali temporary Client err")
} else {
log.Debug().Msg("create ali temporary Client success")
}
<-timeTickerChan
}
}
// createTemporaryClient 创建临时客户端
func (ali *Aliyun) createTemporaryClient() (err error) {
// 获得临时角色
assumeRoleResp, err := ali.AssumeRole()
if err != nil {
return errors.WithMessage(err, "ali.AssumeRole()")
}
// 创建临时客户端
ali.client, err = alioss.New(ali.cfg.EndPoint,
assumeRoleResp.Credentials.AccessKeyId,
assumeRoleResp.Credentials.AccessKeySecret,
alioss.SecurityToken(assumeRoleResp.Credentials.SecurityToken))
if err != nil {
return errors.WithMessage(err, "alioss.New()")
}
return nil
}
// parseUrl 获取对象在阿里云OSS上的完整访问URL
func (ali *Aliyun) parseUrl(key string) string {
return fmt.Sprintf("https://%s.%s/%s", ali.cfg.Bucket, ali.cfg.EndPoint, key)
}
// parseUri ...
func (ali *Aliyun) parseUri(key string) string {
return fmt.Sprintf("/%s", key)
}
// GetHost ...
func (ali *Aliyun) GetHost() string {
return fmt.Sprintf("https://%s.%s", ali.cfg.Bucket, ali.cfg.EndPoint)
}
// InitiateMultipartUpload 初始化分段上传任务
// 使用分段上传方式传输数据前,必须先通知OBS初始化一个分段上传任务。
// 该操作会返回一个OBS服务端创建的全局唯一标识(Upload ID),用于标识本次分段上传任务。
// 您可以根据这个唯一标识来发起相关的操作,如取消分段上传任务、列举分段上传任务、列举已上传的段等。
//
// key string
//
// uploadId string
// err error
func (ali *Aliyun) InitiateMultipartUpload(key string) (uploadId string, err error) {
bucket, err := ali.client.Bucket(ali.cfg.Bucket)
if err != nil {
return "", errors.WithMessage(err, "client.Bucket()")
}
imur, err := bucket.InitiateMultipartUpload(key)
if err != nil {
return "", errors.WithMessage(err, "bucket.InitiateMultipartUpload")
}
return imur.UploadID, nil
}
// UploadPart 上传段
// 初始化一个分段上传任务之后,可以根据指定的对象名和Upload ID来分段上传数据。
// 每一个上传的段都有一个标识它的号码——分段号(Part Number,范围是1~10000)。
// 对于同一个Upload ID,该分段号不但唯一标识这一段数据,也标识了这段数据在整个对象内的相对位置。
// 如果您用同一个分段号上传了新的数据,那么OBS上已有的这个段号的数据将被覆盖。
// 除了最后一段以外,其他段的大小范围是100KB~5GB;最后段大小范围是0~5GB。
// 每个段不需要按顺序上传,甚至可以在不同进程、不同机器上上传,OBS会按照分段号排序组成最终对象。
//
// key string
// uploadId string
// body io.Reader
// partNumber int32
// offset int64
// partSize int64
//
// ETag string
// err error
func (ali *Aliyun) UploadPart(key, uploadId string, body io.Reader, partNumber int32, offset,
partSize int64) (ETag string, err error) {
// TODO offset, partSize 好像是不需要的参数, 如果 body 是 io.Reader 的话
bucket, err := ali.client.Bucket(ali.cfg.Bucket)
if err != nil {
return "", errors.WithMessage(err, "client.Bucket")
}
imur := alioss.InitiateMultipartUploadResult{
Key: key,
UploadID: uploadId,
Bucket: ali.cfg.Bucket,
}
part, err := bucket.UploadPart(imur, body, partSize, int(partNumber))
if err != nil {
return "", errors.WithMessage(err, "bucket.UploadPart")
}
return part.ETag, nil
}
// CompleteMultipartUpload 合并段
// 所有分段上传完成后,需要调用合并段接口来在OBS服务端生成最终对象。
// 在执行该操作时,需要提供所有有效的分段列表(包括分段号和分段ETag值);
// OBS收到提交的分段列表后,会逐一验证每个段的有效性。当所有段验证通过后,OBS将把这些分段组合成最终的对象。
//
// key string
// uploadId string
// parts []model.Part
//
// url string
// err error
func (ali *Aliyun) CompleteMultipartUpload(key, uploadId string, parts []oss.Part) (url, uri string, err error) {
bucket, err := ali.client.Bucket(ali.cfg.Bucket)
if err != nil {
return "", "", errors.WithMessage(err, "client.Bucket")
}
imur := alioss.InitiateMultipartUploadResult{
Key: key,
UploadID: uploadId,
Bucket: ali.cfg.Bucket,
}
ossParts := make([]alioss.UploadPart, len(parts), len(parts))
sort.Sort(oss.Parts(parts))
for i := range parts {
ossParts[i] = alioss.UploadPart{
PartNumber: int(parts[i].PartNumber),
ETag: parts[i].ETag,
}
}
_, err = bucket.CompleteMultipartUpload(imur, ossParts)
if err != nil {
return "", "", errors.WithMessage(err, "bucket.CompleteMultipartUpload")
}
return ali.parseUrl(key), ali.parseUri(key), nil
}
// AbortMultipartUpload 取消分段上传任务
// 分段上传任务可以被取消,当一个分段上传任务被取消后,就不能再使用其Upload ID做任何操作,已经上传段也会被OBS删除。
// 采用分段上传方式上传对象过程中或上传对象失败后会在桶内产生段,这些段会占用您的存储空间,您可以通过取消该分段上传任务来清理掉不需要的段,节约存储空间。
//
// key string
// uploadId string
//
// err error
func (ali *Aliyun) AbortMultipartUpload(key, uploadId string) error {
bucket, err := ali.client.Bucket(ali.cfg.Bucket)
if err != nil {
return errors.WithMessage(err, "client.Bucket")
}
imur := alioss.InitiateMultipartUploadResult{
Key: key,
UploadID: uploadId,
Bucket: ali.cfg.Bucket,
}
err = bucket.AbortMultipartUpload(imur)
if err != nil {
return errors.WithMessage(err, "bucket.AbortMultipartUpload")
}
return nil
}
func (ali *Aliyun) test() {
//ali.client.
}