forked from xiaojiaoyu100/aliyun-acm
-
Notifications
You must be signed in to change notification settings - Fork 0
/
diamond.go
228 lines (207 loc) · 4.9 KB
/
diamond.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
package aliacm
import (
"fmt"
"github.com/xiaojiaoyu100/aliyun-acm/v2/config"
"github.com/xiaojiaoyu100/aliyun-acm/v2/info"
"github.com/xiaojiaoyu100/aliyun-acm/v2/observer"
"github.com/xiaojiaoyu100/curlew"
"math/rand"
"time"
"context"
"github.com/sirupsen/logrus"
"github.com/xiaojiaoyu100/cast"
)
const (
// PublicAddr 公网(测试)
PublicAddr = "acm.aliyun.com"
// HZAddr 华东 1(杭州)
HZAddr = "addr-hz-internal.edas.aliyun.com"
// QDAddr 华北 1(青岛)
QDAddr = "addr-qd-internal.edas.aliyun.com"
// SHAddr 华东 2(上海)
SHAddr = "addr-sh-internal.edas.aliyun.com"
// BJAddr 华北 2(北京)
BJAddr = "addr-bj-internal.edas.aliyun.com"
// SZAddr 华南 1(深圳)
SZAddr = "addr-sz-internal.edas.aliyun.com"
// HKAddr 香港
HKAddr = "addr-hk-internal.edas.aliyuncs.com"
// SingaporeAddr 新加坡
SingaporeAddr = "addr-singapore-internal.edas.aliyun.com"
// ApAddr 澳大利亚(悉尼)
ApAddr = "addr-ap-southeast-2-internal.edas.aliyun.com"
// USWest1Addr 美国(硅谷)
USWest1Addr = "addr-us-west-1-internal.acm.aliyun.com"
// USEast1Addr 美国(弗吉尼亚)
USEast1Addr = "addr-us-east-1-internal.acm.aliyun.com"
// ShanghaiFinance1Addr 华东 2(上海)金融云
ShanghaiFinance1Addr = "addr-cn-shanghai-finance-1-internal.edas.aliyun.com"
)
// Option 参数设置
type Option struct {
addr string
tenant string
accessKey string
secretKey string
}
// Diamond 提供了操作阿里云ACM的能力
type Diamond struct {
option Option
c *cast.Cast
errHook Hook
r *rand.Rand
oo []*observer.Observer
filter map[*observer.Observer]struct{}
all map[info.Info]*config.Config
dispatcher *curlew.Dispatcher
}
// New 产生Diamond实例
func New(addr, tenant, accessKey, secretKey string, setters ...Setter) (*Diamond, error) {
option := Option{
addr: addr,
tenant: tenant,
accessKey: accessKey,
secretKey: secretKey,
}
c, err := cast.New(
cast.WithRetry(2),
cast.WithHTTPClientTimeout(60*time.Second),
cast.WithExponentialBackoffDecorrelatedJitterStrategy(
time.Millisecond*200,
time.Millisecond*500,
),
cast.WithLogLevel(logrus.WarnLevel),
)
if err != nil {
return nil, err
}
s := rand.NewSource(time.Now().UnixNano())
r := rand.New(s)
var monitor = func(err error) {
}
dispatcher, err := curlew.New(
curlew.WithMaxWorkerNum(100),
curlew.WithMonitor(monitor))
if err != nil {
return nil, err
}
d := &Diamond{
option: option,
c: c,
r: r,
filter: make(map[*observer.Observer]struct{}),
all: make(map[info.Info]*config.Config),
dispatcher: dispatcher,
}
for _, setter := range setters {
if err := setter(d); err != nil {
return nil, err
}
}
return d, nil
}
func randomIntInRange(min, max int) int {
return rand.Intn(max-min) + min
}
// Register registers an observer list.
func (d *Diamond) Register(oo ...*observer.Observer) int64 {
start := time.Now()
for _, o := range oo {
_, ok := d.filter[o]
if ok {
continue
}
d.filter[o] = struct{}{}
d.oo = append(d.oo, o)
}
for _, o := range oo {
for _, i := range o.Info() {
d.all[i] = &config.Config{}
}
}
for i := range d.all {
req := &GetConfigRequest{
Tenant: d.option.tenant,
Group: i.Group,
DataID: i.DataID,
}
b, err := d.GetConfig(req)
d.hang(i)
if err != nil {
d.checkErr(i, err)
continue
}
conf := &config.Config{
Content: b,
ContentMD5: Md5(string(b)),
Pulled: true,
}
d.all[i] = conf
for _, o := range d.oo {
o.UpdateInfo(i, conf)
}
}
d.notify()
return time.Now().Sub(start).Milliseconds()
}
func (d *Diamond) notify() {
for _, o := range d.oo {
if o.Ready() {
j := curlew.NewJob()
j.Arg = o
j.Fn = func(ctx context.Context, arg interface{}) error {
o := arg.(*observer.Observer)
o.Handle()
return nil
}
d.dispatcher.Submit(j)
}
}
}
func (d *Diamond) hang(i info.Info) {
go func() {
for {
// 防止网络差的时候没挂住
time.Sleep(time.Duration(randomIntInRange(20, 100)) * time.Millisecond)
content, newContentMD5, err := d.LongPull(i, d.all[i].ContentMD5)
d.checkErr(i, err)
// 防止MD5被重置,重新请求
if newContentMD5 == "" {
continue
}
conf := &config.Config{
Content: content,
ContentMD5: newContentMD5,
Pulled: true,
}
d.all[i] = conf
for _, o := range d.oo {
o.UpdateInfo(i, conf)
}
d.notify()
}
}()
}
// SetHook 用于提醒关键错误
func (d *Diamond) SetHook(h Hook) {
d.errHook = h
}
func (d *Diamond) checkErr(i info.Info, err error) {
if shouldIgnore(err) {
return
}
if d.errHook == nil {
return
}
d.errHook(i, err)
}
// LoadingProgress shows the current load progress roughly.
func (d *Diamond) LoadingProgress() string {
var ret = 0
for _, conf := range d.all {
if conf.Pulled {
ret++
}
}
return fmt.Sprintf("%d/%d", ret, len(d.all))
}