Skip to content

Commit 25c1a1d

Browse files
DuodenumLDuodenumL
committed
support setting tls configs in "SetNode"
1 parent 76ac53f commit 25c1a1d

File tree

14 files changed

+1101
-963
lines changed

14 files changed

+1101
-963
lines changed

‎cluster/calcium/node.go

+11-1
Original file line numberDiff line numberDiff line change
@@ -122,13 +122,23 @@ func (c *Calcium) SetNode(ctx context.Context, opts *types.SetNodeOptions) (*typ
122122
if opts.Endpoint != "" {
123123
n.Endpoint = opts.Endpoint
124124
}
125+
// update ca / cert / key
126+
if opts.Ca != "" {
127+
n.Ca = opts.Ca
128+
}
129+
if opts.Cert != "" {
130+
n.Cert = opts.Cert
131+
}
132+
if opts.Key != "" {
133+
n.Key = opts.Key
134+
}
125135
// update key value
126136
if len(opts.Labels) != 0 {
127137
n.Labels = opts.Labels
128138
}
129139
// update numa
130140
if len(opts.NUMA) != 0 {
131-
n.NUMA = types.NUMA(opts.NUMA)
141+
n.NUMA = opts.NUMA
132142
}
133143
// update numa memory
134144
for numaNode, memoryDelta := range opts.DeltaNUMAMemory {

‎cluster/calcium/node_test.go

+6
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,12 @@ func TestSetNode(t *testing.T) {
196196
n, err = c.SetNode(ctx, setOpts)
197197
assert.NoError(t, err)
198198
assert.Equal(t, n.Endpoint, "tcp://10.10.10.10:2379")
199+
// set ca / cert / key
200+
setOpts.Ca = "hh"
201+
setOpts.Cert = "hh"
202+
setOpts.Key = "hh"
203+
n, err = c.SetNode(ctx, setOpts)
204+
assert.NoError(t, err)
199205
// set numa
200206
setOpts.NUMA = types.NUMA{"100": "node1"}
201207
n, err = c.SetNode(ctx, setOpts)

‎core.yaml.sample

+1
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ bind: ":5001"
33
statsd: "127.0.0.1:8125"
44
profile: ":12346"
55
global_timeout: 300s
6+
connection_timeout: 10s
67
lock_timeout: 30s
78
cert_path: "/etc/eru/tls"
89
sentry_dsn: "https://examplePublicKey@o0.ingest.sentry.io/0"

‎engine/factory/factory.go

+41-3
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"github.com/projecteru2/core/engine/mocks/fakeengine"
1212
"github.com/projecteru2/core/engine/systemd"
1313
"github.com/projecteru2/core/engine/virt"
14+
"github.com/projecteru2/core/log"
1415
"github.com/projecteru2/core/types"
1516
"github.com/projecteru2/core/utils"
1617
)
@@ -29,15 +30,45 @@ var (
2930
engineCache = utils.NewEngineCache(12*time.Hour, 10*time.Minute)
3031
)
3132

33+
func getEngineCacheKey(endpoint, ca, cert, key string) string {
34+
return fmt.Sprintf("%v:%v:%v:%v", endpoint, ca, cert, key)
35+
}
36+
37+
func validateEngine(ctx context.Context, engine engine.API, timeout time.Duration) (err error) {
38+
utils.WithTimeout(ctx, timeout, func(ctx context.Context) {
39+
_, err = engine.Info(ctx)
40+
})
41+
return err
42+
}
43+
44+
// GetEngineFromCache .
45+
func GetEngineFromCache(ctx context.Context, config types.Config, endpoint, ca, cert, key string) engine.API {
46+
client := engineCache.Get(getEngineCacheKey(endpoint, ca, cert, key))
47+
if client == nil {
48+
return nil
49+
}
50+
if err := validateEngine(ctx, client, config.ConnectionTimeout); err != nil {
51+
log.Errorf(ctx, "[GetEngineFromCache] engine of %v is unavailable, will be removed from cache, err: %v", endpoint, err)
52+
RemoveEngineFromCache(endpoint, ca, cert, key)
53+
return nil
54+
}
55+
return client
56+
}
57+
58+
// RemoveEngineFromCache .
59+
func RemoveEngineFromCache(endpoint, ca, cert, key string) {
60+
engineCache.Delete(getEngineCacheKey(endpoint, ca, cert, key))
61+
}
62+
3263
// GetEngine get engine
3364
func GetEngine(ctx context.Context, config types.Config, nodename, endpoint, ca, cert, key string) (client engine.API, err error) {
34-
if client = engineCache.Get(endpoint); client != nil {
65+
if client = GetEngineFromCache(ctx, config, endpoint, ca, cert, key); client != nil {
3566
return
3667
}
3768

3869
defer func() {
3970
if err == nil && client != nil {
40-
engineCache.Set(endpoint, client)
71+
engineCache.Set(getEngineCacheKey(endpoint, ca, cert, key), client)
4172
}
4273
}()
4374

@@ -49,7 +80,14 @@ func GetEngine(ctx context.Context, config types.Config, nodename, endpoint, ca,
4980
if !ok {
5081
return nil, types.ErrNotSupport
5182
}
52-
return e(ctx, config, nodename, endpoint, ca, cert, key)
83+
if client, err = e(ctx, config, nodename, endpoint, ca, cert, key); err != nil {
84+
return nil, err
85+
}
86+
if err = validateEngine(ctx, client, config.ConnectionTimeout); err != nil {
87+
log.Errorf(ctx, "[GetEngine] engine of %v is unavailable, err: %v", endpoint, err)
88+
return nil, err
89+
}
90+
return client, nil
5391
}
5492

5593
func getEnginePrefix(endpoint string) (string, error) {

‎rpc/gen/core.pb.go

+960-933
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎rpc/gen/core.proto

+3
Original file line numberDiff line numberDiff line change
@@ -181,6 +181,9 @@ message SetNodeOptions {
181181
bool workloads_down = 10;
182182
string endpoint = 11;
183183
TriOpt bypass_opt = 12;
184+
string ca = 13;
185+
string cert = 14;
186+
string key = 15;
184187
}
185188

186189
message SetNodeStatusOptions {

‎rpc/transform.go

+3
Original file line numberDiff line numberDiff line change
@@ -171,6 +171,9 @@ func toCoreSetNodeOptions(b *pb.SetNodeOptions) (*types.SetNodeOptions, error) {
171171
NUMA: b.Numa,
172172
Labels: b.Labels,
173173
BypassOpt: types.TriOptions(b.BypassOpt),
174+
Ca: b.Ca,
175+
Cert: b.Cert,
176+
Key: b.Key,
174177
}
175178
for cpuID, cpuShare := range b.DeltaCpu {
176179
r.DeltaCPU[cpuID] = int64(cpuShare)

‎store/etcdv3/node.go

+33-9
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,11 @@ import (
77
"path/filepath"
88
"strconv"
99
"strings"
10+
"sync"
11+
12+
"github.com/pkg/errors"
13+
"go.etcd.io/etcd/api/v3/mvccpb"
14+
clientv3 "go.etcd.io/etcd/client/v3"
1015

1116
"github.com/projecteru2/core/engine"
1217
enginefactory "github.com/projecteru2/core/engine/factory"
@@ -15,10 +20,6 @@ import (
1520
"github.com/projecteru2/core/store"
1621
"github.com/projecteru2/core/types"
1722
"github.com/projecteru2/core/utils"
18-
19-
"github.com/pkg/errors"
20-
"go.etcd.io/etcd/api/v3/mvccpb"
21-
clientv3 "go.etcd.io/etcd/client/v3"
2223
)
2324

2425
// AddNode save it to etcd
@@ -145,6 +146,11 @@ func (m *Mercury) GetNodesByPod(ctx context.Context, podname string, labels map[
145146
// UpdateNodes .
146147
func (m *Mercury) UpdateNodes(ctx context.Context, nodes ...*types.Node) error {
147148
data := map[string]string{}
149+
addIfNotEmpty := func(key, value string) {
150+
if value != "" {
151+
data[key] = value
152+
}
153+
}
148154
for _, node := range nodes {
149155
bytes, err := json.Marshal(node)
150156
if err != nil {
@@ -153,6 +159,10 @@ func (m *Mercury) UpdateNodes(ctx context.Context, nodes ...*types.Node) error {
153159
d := string(bytes)
154160
data[fmt.Sprintf(nodeInfoKey, node.Name)] = d
155161
data[fmt.Sprintf(nodePodKey, node.Podname, node.Name)] = d
162+
addIfNotEmpty(fmt.Sprintf(nodeCaKey, node.Name), node.Ca)
163+
addIfNotEmpty(fmt.Sprintf(nodeCertKey, node.Name), node.Cert)
164+
addIfNotEmpty(fmt.Sprintf(nodeKeyKey, node.Name), node.Key)
165+
enginefactory.RemoveEngineFromCache(node.Endpoint, node.Ca, node.Cert, node.Key)
156166
}
157167

158168
resp, err := m.BatchUpdate(ctx, data)
@@ -180,13 +190,19 @@ func (m *Mercury) UpdateNodeResource(ctx context.Context, node *types.Node, reso
180190
}
181191

182192
func (m *Mercury) makeClient(ctx context.Context, node *types.Node) (client engine.API, err error) {
193+
// try to get from cache without ca/cert/key
194+
if client = enginefactory.GetEngineFromCache(ctx, m.config, node.Endpoint, "", "", ""); client != nil {
195+
return client, nil
196+
}
197+
183198
keyFormats := []string{nodeCaKey, nodeCertKey, nodeKeyKey}
184199
data := []string{"", "", ""}
185200
for i := 0; i < 3; i++ {
186201
ev, err := m.GetOne(ctx, fmt.Sprintf(keyFormats[i], node.Name))
187202
if err != nil {
188203
if !errors.Is(err, types.ErrBadCount) {
189204
log.Warnf(ctx, "[makeClient] Get key failed %v", err)
205+
return nil, err
190206
}
191207
continue
192208
}
@@ -282,13 +298,21 @@ func (m *Mercury) doGetNodes(ctx context.Context, kvs []*mvccpb.KeyValue, labels
282298
return nil, err
283299
}
284300
node.Init()
285-
if (!node.IsDown() || all) && utils.FilterWorkload(node.Labels, labels) {
286-
if node.Engine, err = m.makeClient(ctx, node); err != nil {
287-
return
301+
nodes = append(nodes, node)
302+
}
303+
wg := &sync.WaitGroup{}
304+
wg.Add(len(nodes))
305+
for _, node := range nodes {
306+
go func(node *types.Node) {
307+
defer wg.Done()
308+
if (!node.IsDown() || all) && utils.FilterWorkload(node.Labels, labels) {
309+
if node.Engine, err = m.makeClient(ctx, node); err != nil {
310+
return
311+
}
288312
}
289-
nodes = append(nodes, node)
290-
}
313+
}(node)
291314
}
315+
wg.Wait()
292316
return nodes, nil
293317
}
294318

‎store/etcdv3/node_test.go

+6-2
Original file line numberDiff line numberDiff line change
@@ -201,8 +201,12 @@ func TestUpdateNode(t *testing.T) {
201201
assert.Equal(t, node.Name, "test")
202202
fakeNode := &types.Node{
203203
NodeMeta: types.NodeMeta{
204-
Name: "nil",
205-
Podname: "wtf",
204+
Name: "nil",
205+
Podname: "wtf",
206+
Endpoint: "mock://hh",
207+
Ca: "hh",
208+
Cert: "hh",
209+
Key: "hh",
206210
},
207211
}
208212
assert.Error(t, m.UpdateNodes(ctx, fakeNode))

‎store/redis/node.go

+14-4
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,11 @@ func (r *Rediaron) GetNodesByPod(ctx context.Context, podname string, labels map
144144
// UpdateNodes .
145145
func (r *Rediaron) UpdateNodes(ctx context.Context, nodes ...*types.Node) error {
146146
data := map[string]string{}
147+
addIfNotEmpty := func(key, value string) {
148+
if value != "" {
149+
data[key] = value
150+
}
151+
}
147152
for _, node := range nodes {
148153
bytes, err := json.Marshal(node)
149154
if err != nil {
@@ -152,6 +157,9 @@ func (r *Rediaron) UpdateNodes(ctx context.Context, nodes ...*types.Node) error
152157
d := string(bytes)
153158
data[fmt.Sprintf(nodeInfoKey, node.Name)] = d
154159
data[fmt.Sprintf(nodePodKey, node.Podname, node.Name)] = d
160+
addIfNotEmpty(fmt.Sprintf(nodeCaKey, node.Name), node.Ca)
161+
addIfNotEmpty(fmt.Sprintf(nodeCertKey, node.Name), node.Cert)
162+
addIfNotEmpty(fmt.Sprintf(nodeKeyKey, node.Name), node.Key)
155163
}
156164
return errors.WithStack(r.BatchUpdate(ctx, data))
157165
}
@@ -170,17 +178,19 @@ func (r *Rediaron) UpdateNodeResource(ctx context.Context, node *types.Node, res
170178
return r.UpdateNodes(ctx, node)
171179
}
172180

173-
func (r *Rediaron) makeClient(ctx context.Context, node *types.Node) (engine.API, error) {
174-
// try get client, if nil, create a new one
175-
var client engine.API
176-
var err error
181+
func (r *Rediaron) makeClient(ctx context.Context, node *types.Node) (client engine.API, err error) {
182+
// try to get from cache without ca/cert/key
183+
if client = enginefactory.GetEngineFromCache(ctx, r.config, node.Endpoint, "", "", ""); client != nil {
184+
return client, nil
185+
}
177186
keyFormats := []string{nodeCaKey, nodeCertKey, nodeKeyKey}
178187
data := []string{"", "", ""}
179188
for i := 0; i < 3; i++ {
180189
v, err := r.GetOne(ctx, fmt.Sprintf(keyFormats[i], node.Name))
181190
if err != nil {
182191
if !isRedisNoKeyError(err) {
183192
log.Warnf(ctx, "[makeClient] Get key failed %v", err)
193+
return nil, err
184194
}
185195
continue
186196
}

‎store/redis/node_test.go

+6-2
Original file line numberDiff line numberDiff line change
@@ -194,8 +194,12 @@ func (s *RediaronTestSuite) TestUpdateNode() {
194194
s.Equal(node.Name, "test")
195195
fakeNode := &types.Node{
196196
NodeMeta: types.NodeMeta{
197-
Name: "nil",
198-
Podname: "wtf",
197+
Name: "nil",
198+
Podname: "wtf",
199+
Endpoint: "mock://hh",
200+
Ca: "hh",
201+
Cert: "hh",
202+
Key: "hh",
199203
},
200204
}
201205
s.Error(s.rediaron.UpdateNodes(ctx, fakeNode))

‎types/config.go

+10-9
Original file line numberDiff line numberDiff line change
@@ -17,15 +17,16 @@ const (
1717

1818
// Config holds eru-core config
1919
type Config struct {
20-
LogLevel string `yaml:"log_level" required:"true" default:"INFO"`
21-
Bind string `yaml:"bind" required:"true" default:"5001"` // HTTP API address
22-
LockTimeout time.Duration `yaml:"lock_timeout" required:"true" default:"30s"` // timeout for lock (ttl)
23-
GlobalTimeout time.Duration `yaml:"global_timeout" required:"true" default:"300s"` // timeout for remove, run_and_wait and build, in second
24-
Statsd string `yaml:"statsd"` // statsd host and port
25-
Profile string `yaml:"profile"` // profile ip:port
26-
CertPath string `yaml:"cert_path"` // docker cert files path
27-
MaxConcurrency int64 `yaml:"max_concurrency" default:"20"` // concurrently call single runtime in the same time
28-
Store string `yaml:"store" default:"etcd"` // store type
20+
LogLevel string `yaml:"log_level" required:"true" default:"INFO"`
21+
Bind string `yaml:"bind" required:"true" default:"5001"` // HTTP API address
22+
LockTimeout time.Duration `yaml:"lock_timeout" required:"true" default:"30s"` // timeout for lock (ttl)
23+
GlobalTimeout time.Duration `yaml:"global_timeout" required:"true" default:"300s"` // timeout for remove, run_and_wait and build, in second
24+
ConnectionTimeout time.Duration `yaml:"connection_timeout" default:"10s"` // timeout for connections
25+
Statsd string `yaml:"statsd"` // statsd host and port
26+
Profile string `yaml:"profile"` // profile ip:port
27+
CertPath string `yaml:"cert_path"` // docker cert files path
28+
MaxConcurrency int64 `yaml:"max_concurrency" default:"20"` // concurrently call single runtime in the same time
29+
Store string `yaml:"store" default:"etcd"` // store type
2930

3031
Auth AuthConfig `yaml:"auth"` // grpc auth
3132
GRPCConfig GRPCConfig `yaml:"grpc"` // grpc config

‎types/node.go

+4
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,10 @@ type NodeMeta struct {
4242
InitStorageCap int64 `json:"init_storage_cap"`
4343
InitNUMAMemory NUMAMemory `json:"init_numa_memory"`
4444
InitVolume VolumeMap `json:"init_volume"`
45+
46+
Ca string `json:"-"`
47+
Cert string `json:"-"`
48+
Key string `json:"-"`
4549
}
4650

4751
// DeepCopy returns a deepcopy of nodemeta

‎types/options.go

+3
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,9 @@ type SetNodeOptions struct {
248248
NUMA map[string]string
249249
Labels map[string]string
250250
BypassOpt TriOptions
251+
Ca string
252+
Cert string
253+
Key string
251254
}
252255

253256
// Validate checks options

0 commit comments

Comments
 (0)