Dragonfly2

1
/*
2
 *     Copyright 2020 The Dragonfly Authors
3
 *
4
 * Licensed under the Apache License, Version 2.0 (the "License");
5
 * you may not use this file except in compliance with the License.
6
 * You may obtain a copy of the License at
7
 *
8
 *      http://www.apache.org/licenses/LICENSE-2.0
9
 *
10
 * Unless required by applicable law or agreed to in writing, software
11
 * distributed under the License is distributed on an "AS IS" BASIS,
12
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13
 * See the License for the specific language governing permissions and
14
 * limitations under the License.
15
 */
16

17
//go:generate mockgen -destination mocks/dynconfig_mock.go -source dynconfig.go -package mocks
18

19
package config
20

21
import (
22
	"errors"
23
	"time"
24

25
	"google.golang.org/grpc/credentials"
26
	"google.golang.org/grpc/resolver"
27

28
	managerv1 "d7y.io/api/v2/pkg/apis/manager/v1"
29

30
	managerclient "d7y.io/dragonfly/v2/pkg/rpc/manager/client"
31
)
32

33
type SourceType string
34

35
const (
36
	// LocalSourceType represents read configuration from local file.
37
	LocalSourceType = "local"
38

39
	// ManagerSourceType represents pulling configuration from manager.
40
	ManagerSourceType = "manager"
41
)
42

43
// Watch dynconfig interval.
44
var watchInterval = 10 * time.Second
45

46
type DynconfigData struct {
47
	SeedPeers     []*managerv1.SeedPeer
48
	Schedulers    []*managerv1.Scheduler
49
	ObjectStorage *managerv1.ObjectStorage
50
}
51

52
type Dynconfig interface {
53
	// Get the dynamic seed peers config.
54
	GetSeedPeers() ([]*managerv1.SeedPeer, error)
55

56
	// Get the dynamic schedulers resolve addrs.
57
	GetResolveSchedulerAddrs() ([]resolver.Address, error)
58

59
	// Get the dynamic schedulers config.
60
	GetSchedulers() ([]*managerv1.Scheduler, error)
61

62
	// Get the dynamic schedulers cluster id.
63
	GetSchedulerClusterID() uint64
64

65
	// Get the dynamic object storage config.
66
	GetObjectStorage() (*managerv1.ObjectStorage, error)
67

68
	// Get the dynamic config.
69
	Get() (*DynconfigData, error)
70

71
	// Refresh refreshes dynconfig in cache.
72
	Refresh() error
73

74
	// Register allows an instance to register itself to listen/observe events.
75
	Register(Observer)
76

77
	// Deregister allows an instance to remove itself from the collection of observers/listeners.
78
	Deregister(Observer)
79

80
	// Notify publishes new events to listeners.
81
	Notify() error
82

83
	// OnNotify allows an event to be published to the dynconfig.
84
	// Used for listening changes of the local configuration.
85
	OnNotify(*DaemonOption)
86

87
	// Serve the dynconfig listening service.
88
	Serve() error
89

90
	// Stop the dynconfig listening service.
91
	Stop() error
92
}
93

94
type dynconfig struct {
95
	sourceType           SourceType
96
	managerClient        managerclient.V1
97
	cacheDir             string
98
	transportCredentials credentials.TransportCredentials
99
}
100

101
type Observer interface {
102
	// OnNotify allows an event to be published to interface implementations.
103
	OnNotify(*DynconfigData)
104
}
105

106
// DynconfigOption is a functional option for configuring the dynconfig.
107
type DynconfigOption func(d *dynconfig) error
108

109
// WithManagerClient set the manager client.
110
func WithManagerClient(c managerclient.V1) DynconfigOption {
111
	return func(d *dynconfig) error {
112
		d.managerClient = c
113
		return nil
114
	}
115
}
116

117
// WithCacheDir set the cache dir.
118
func WithCacheDir(dir string) DynconfigOption {
119
	return func(d *dynconfig) error {
120
		d.cacheDir = dir
121
		return nil
122
	}
123
}
124

125
// WithTransportCredentials returns a DialOption which configures a connection
126
// level security credentials (e.g., TLS/SSL).
127
func WithTransportCredentials(creds credentials.TransportCredentials) DynconfigOption {
128
	return func(d *dynconfig) error {
129
		d.transportCredentials = creds
130
		return nil
131
	}
132
}
133

134
// New returns a new dynconfig interface.
135
func NewDynconfig(sourceType SourceType, cfg *DaemonOption, options ...DynconfigOption) (Dynconfig, error) {
136
	d := &dynconfig{
137
		sourceType: sourceType,
138
	}
139

140
	for _, opt := range options {
141
		if err := opt(d); err != nil {
142
			return nil, err
143
		}
144
	}
145

146
	if err := d.validate(); err != nil {
147
		return nil, err
148
	}
149

150
	var (
151
		di  Dynconfig
152
		err error
153
	)
154
	switch sourceType {
155
	case ManagerSourceType:
156
		di, err = newDynconfigManager(cfg, d.managerClient, d.cacheDir, d.transportCredentials)
157
		if err != nil {
158
			return nil, err
159
		}
160
	case LocalSourceType:
161
		di, err = newDynconfigLocal(cfg, d.transportCredentials)
162
		if err != nil {
163
			return nil, err
164
		}
165
	default:
166
		return nil, errors.New("unknown source type")
167
	}
168

169
	return di, nil
170
}
171

172
// validate dynconfig parameters.
173
func (d *dynconfig) validate() error {
174
	if d.sourceType == ManagerSourceType {
175
		if d.managerClient == nil {
176
			return errors.New("manager dynconfig requires parameter ManagerClient")
177
		}
178

179
		if d.cacheDir == "" {
180
			return errors.New("manager dynconfig requires parameter CacheDir")
181
		}
182
	}
183

184
	return nil
185
}
186