-
Notifications
You must be signed in to change notification settings - Fork 0
/
store.go
326 lines (275 loc) · 8.02 KB
/
store.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
324
325
326
// Copyright 2017 Istio Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package store
import (
"errors"
"fmt"
"net/url"
"sync"
"time"
"github.com/gogo/protobuf/proto"
"k8s.io/apimachinery/pkg/runtime/schema"
"istio.io/istio/pkg/log"
"istio.io/istio/pkg/mcp/creds"
"istio.io/istio/pkg/probe"
)
// ChangeType denotes the type of a change
type ChangeType int
const (
// Update - change was an update or a create to a key.
Update ChangeType = iota
// Delete - key was removed.
Delete
)
// ErrNotFound is the error to be returned when the given key does not exist in the storage.
var ErrNotFound = errors.New("not found")
// ErrWatchAlreadyExists is the error to report that the watching channel already exists.
var ErrWatchAlreadyExists = errors.New("watch already exists")
// Key represents the key to identify a resource in the store.
type Key struct {
Kind string
Namespace string
Name string
}
// BackendEvent is an event used between Backend and Store.
type BackendEvent struct {
Key
Type ChangeType
Value *BackEndResource
}
// ResourceMeta is the standard metadata associated with a resource.
type ResourceMeta struct {
Name string
Namespace string
Labels map[string]string
Annotations map[string]string
Revision string
}
// BackEndResource represents a resources with a raw spec
type BackEndResource struct {
Kind string
Metadata ResourceMeta
Spec map[string]interface{}
}
// Key returns the key of the resource in the store.
func (ber *BackEndResource) Key() Key {
return Key{
Kind: ber.Kind,
Name: ber.Metadata.Name,
Namespace: ber.Metadata.Namespace,
}
}
// Resource represents a resources with converted spec.
type Resource struct {
Metadata ResourceMeta
Spec proto.Message
}
// String is the Istio compatible string representation of the resource.
// Name.Kind.Namespace
// At the point of use Namespace can be omitted, and it is assumed to be the namespace
// of the document.
func (k Key) String() string {
return k.Name + "." + k.Kind + "." + k.Namespace
}
// Event represents an event. Used by Store.Watch.
type Event struct {
Key
Type ChangeType
// Value refers the new value in the updated event. nil if the event type is delete.
Value *Resource
}
// BackendValidator defines the interface to validte unstructured event.
type BackendValidator interface {
Validate(ev *BackendEvent) error
}
// Validator defines the interface to validate a new change.
type Validator interface {
Validate(ev *Event) error
}
// Backend defines the typeless storage backend for mixer.
type Backend interface {
Init(kinds []string) error
Stop()
// WaitForSynced blocks and awaits for the caches to be fully populated until timeout.
WaitForSynced(time.Duration) error
// Watch creates a channel to receive the events.
Watch() (<-chan BackendEvent, error)
// Get returns a resource's spec to the key.
Get(key Key) (*BackEndResource, error)
// List returns the whole mapping from key to resource specs in the store.
List() map[Key]*BackEndResource
}
// Store defines the access to the storage for mixer.
type Store interface {
Init(kinds map[string]proto.Message) error
Stop()
// WaitForSynced blocks and awaits for the caches to be fully populated until timeout.
WaitForSynced(time.Duration) error
// Watch creates a channel to receive the events. A store can conduct a single
// watch channel at the same time. Multiple calls lead to an error.
Watch() (<-chan Event, error)
// Get returns the resource to the key.
Get(key Key) (*Resource, error)
// List returns the whole mapping from key to resource specs in the store.
List() map[Key]*Resource
probe.SupportsProbe
}
// store is the implementation of Store interface.
type store struct {
kinds map[string]proto.Message
backend Backend
mu sync.Mutex
queue *eventQueue
}
func (s *store) RegisterProbe(c probe.Controller, name string) {
if e, ok := s.backend.(probe.SupportsProbe); ok {
e.RegisterProbe(c, name)
}
}
func (s *store) Stop() {
s.mu.Lock()
if s.queue != nil {
close(s.queue.closec)
}
s.queue = nil
s.mu.Unlock()
s.backend.Stop()
}
// Init initializes the connection with the storage backend. This uses "kinds"
// for the mapping from the kind's name and its structure in protobuf.
func (s *store) Init(kinds map[string]proto.Message) error {
kindNames := make([]string, 0, len(kinds))
for k := range kinds {
kindNames = append(kindNames, k)
}
if err := s.backend.Init(kindNames); err != nil {
return err
}
s.kinds = kinds
return nil
}
// WaitForSynced awaits for the backend to sync.
func (s *store) WaitForSynced(timeout time.Duration) error {
s.mu.Lock()
defer s.mu.Unlock()
return s.backend.WaitForSynced(timeout)
}
// Watch creates a channel to receive the events.
func (s *store) Watch() (<-chan Event, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.queue != nil {
return nil, ErrWatchAlreadyExists
}
ch, err := s.backend.Watch()
if err != nil {
return nil, err
}
q := newQueue(ch, s.kinds)
s.queue = q
return q.chout, nil
}
// Get returns a resource's spec to the key.
func (s *store) Get(key Key) (*Resource, error) {
obj, err := s.backend.Get(key)
if err != nil {
return nil, err
}
pbSpec, err := cloneMessage(key.Kind, s.kinds)
if err != nil {
return nil, err
}
if err = convert(key, obj.Spec, pbSpec); err != nil {
return nil, err
}
return &Resource{
Metadata: obj.Metadata,
Spec: pbSpec,
}, nil
}
// List returns the whole mapping from key to resource specs in the store.
func (s *store) List() map[Key]*Resource {
data := s.backend.List()
result := make(map[Key]*Resource, len(data))
for k, d := range data {
pbSpec, err := cloneMessage(k.Kind, s.kinds)
if err != nil {
log.Errorf("Failed to clone %s spec: %v", k, err)
continue
}
if err = convert(k, d.Spec, pbSpec); err != nil {
log.Errorf("Failed to convert %s spec: %v", k, err)
continue
}
result[k] = &Resource{
Metadata: d.Metadata,
Spec: pbSpec,
}
}
return result
}
// WithBackend creates a new Store with a certain backend. This should be used
// only by tests.
func WithBackend(b Backend) Store {
return &store{backend: b}
}
// Builder is the type of function to build a Backend.
type Builder func(u *url.URL, gv *schema.GroupVersion, credOptions *creds.Options, ck []string) (Backend, error)
// RegisterFunc is the type to register a builder for URL scheme.
type RegisterFunc func(map[string]Builder)
// Registry keeps the relationship between the URL scheme and
// the Backend implementation.
type Registry struct {
builders map[string]Builder
}
// NewRegistry creates a new Registry instance for the inventory.
func NewRegistry(inventory ...RegisterFunc) *Registry {
b := map[string]Builder{}
for _, rf := range inventory {
rf(b)
}
return &Registry{builders: b}
}
// URL types supported by the config store
const (
// example fs:///tmp/testdata/configroot
FSUrl = "fs"
)
// NewStore creates a new Store instance with the specified backend.
func (r *Registry) NewStore(
configURL string,
groupVersion *schema.GroupVersion,
credOptions *creds.Options,
criticalKinds []string) (Store, error) {
u, err := url.Parse(configURL)
if err != nil {
return nil, fmt.Errorf("invalid config URL %s %v", configURL, err)
}
var b Backend
switch u.Scheme {
case FSUrl:
b = newFsStore(u.Path)
default:
if builder, ok := r.builders[u.Scheme]; ok {
b, err = builder(u, groupVersion, credOptions, criticalKinds)
if err != nil {
return nil, err
}
}
}
if b != nil {
return &store{backend: b}, nil
}
return nil, fmt.Errorf("unknown config URL %s %v", configURL, u)
}