istio

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

15
// Package memory provides an in-memory volatile config store implementation
16
package memory
17

18
import (
19
	"errors"
20
	"fmt"
21
	"sync"
22
	"time"
23

24
	"istio.io/istio/pilot/pkg/model"
25
	"istio.io/istio/pkg/config"
26
	"istio.io/istio/pkg/config/schema/collection"
27
)
28

29
var (
30
	errNotFound      = errors.New("item not found")
31
	errAlreadyExists = errors.New("item already exists")
32
	// TODO: can we make this compatible with kerror.IsConflict without imports the library?
33
	errConflict = errors.New("conflicting resource version, try again")
34
)
35

36
const ResourceVersion string = "ResourceVersion"
37

38
// Make creates an in-memory config store from a config schemas
39
// It is with validation
40
func Make(schemas collection.Schemas) model.ConfigStore {
41
	return newStore(schemas, false)
42
}
43

44
// MakeSkipValidation creates an in-memory config store from a config schemas
45
// It is without validation
46
func MakeSkipValidation(schemas collection.Schemas) model.ConfigStore {
47
	return newStore(schemas, true)
48
}
49

50
func newStore(schemas collection.Schemas, skipValidation bool) model.ConfigStore {
51
	out := store{
52
		schemas:        schemas,
53
		data:           make(map[config.GroupVersionKind]map[string]map[string]any),
54
		skipValidation: skipValidation,
55
	}
56
	for _, s := range schemas.All() {
57
		out.data[s.GroupVersionKind()] = make(map[string]map[string]any)
58
	}
59
	return &out
60
}
61

62
type store struct {
63
	schemas        collection.Schemas
64
	data           map[config.GroupVersionKind]map[string]map[string]any
65
	skipValidation bool
66
	mutex          sync.RWMutex
67
}
68

69
func (cr *store) Schemas() collection.Schemas {
70
	return cr.schemas
71
}
72

73
func (cr *store) Get(kind config.GroupVersionKind, name, namespace string) *config.Config {
74
	cr.mutex.RLock()
75
	defer cr.mutex.RUnlock()
76
	_, ok := cr.data[kind]
77
	if !ok {
78
		return nil
79
	}
80

81
	ns, exists := cr.data[kind][namespace]
82
	if !exists {
83
		return nil
84
	}
85

86
	out, exists := ns[name]
87
	if !exists {
88
		return nil
89
	}
90
	config := out.(config.Config)
91

92
	return &config
93
}
94

95
func (cr *store) List(kind config.GroupVersionKind, namespace string) []config.Config {
96
	cr.mutex.RLock()
97
	defer cr.mutex.RUnlock()
98
	data, exists := cr.data[kind]
99
	if !exists {
100
		return nil
101
	}
102

103
	var size int
104
	if namespace == "" {
105
		for _, ns := range data {
106
			size += len(ns)
107
		}
108
	} else {
109
		size = len(data[namespace])
110
	}
111

112
	out := make([]config.Config, 0, size)
113
	if namespace == "" {
114
		for _, ns := range data {
115
			for _, value := range ns {
116
				out = append(out, value.(config.Config))
117
			}
118
		}
119
	} else {
120
		ns, exists := data[namespace]
121
		if !exists {
122
			return nil
123
		}
124
		for _, value := range ns {
125
			out = append(out, value.(config.Config))
126
		}
127
	}
128
	return out
129
}
130

131
func (cr *store) Delete(kind config.GroupVersionKind, name, namespace string, resourceVersion *string) error {
132
	cr.mutex.Lock()
133
	defer cr.mutex.Unlock()
134
	data, ok := cr.data[kind]
135
	if !ok {
136
		return fmt.Errorf("unknown type %v", kind)
137
	}
138
	ns, exists := data[namespace]
139
	if !exists {
140
		return errNotFound
141
	}
142

143
	_, exists = ns[name]
144
	if !exists {
145
		return errNotFound
146
	}
147

148
	delete(ns, name)
149
	return nil
150
}
151

152
func (cr *store) Create(cfg config.Config) (string, error) {
153
	cr.mutex.Lock()
154
	defer cr.mutex.Unlock()
155
	kind := cfg.GroupVersionKind
156
	s, ok := cr.schemas.FindByGroupVersionKind(kind)
157
	if !ok {
158
		return "", fmt.Errorf("unknown type %v", kind)
159
	}
160
	if !cr.skipValidation {
161
		if _, err := s.ValidateConfig(cfg); err != nil {
162
			return "", err
163
		}
164
	}
165
	ns, exists := cr.data[kind][cfg.Namespace]
166
	if !exists {
167
		ns = map[string]any{}
168
		cr.data[kind][cfg.Namespace] = ns
169
	}
170

171
	_, exists = ns[cfg.Name]
172

173
	if !exists {
174
		tnow := time.Now()
175
		if cfg.ResourceVersion == "" {
176
			cfg.ResourceVersion = tnow.String()
177
		}
178
		// Set the creation timestamp, if not provided.
179
		if cfg.CreationTimestamp.IsZero() {
180
			cfg.CreationTimestamp = tnow
181
		}
182

183
		ns[cfg.Name] = cfg
184
		return cfg.ResourceVersion, nil
185
	}
186
	return "", errAlreadyExists
187
}
188

189
func (cr *store) Update(cfg config.Config) (string, error) {
190
	cr.mutex.Lock()
191
	defer cr.mutex.Unlock()
192
	kind := cfg.GroupVersionKind
193
	s, ok := cr.schemas.FindByGroupVersionKind(kind)
194
	if !ok {
195
		return "", fmt.Errorf("unknown type %v", kind)
196
	}
197
	if !cr.skipValidation {
198
		if _, err := s.ValidateConfig(cfg); err != nil {
199
			return "", err
200
		}
201
	}
202

203
	ns, exists := cr.data[kind][cfg.Namespace]
204
	if !exists {
205
		return "", errNotFound
206
	}
207

208
	existing, exists := ns[cfg.Name]
209
	if !exists {
210
		return "", errNotFound
211
	}
212
	if hasConflict(existing.(config.Config), cfg) {
213
		return "", errConflict
214
	}
215
	if cfg.Annotations != nil && cfg.Annotations[ResourceVersion] != "" {
216
		cfg.ResourceVersion = cfg.Annotations[ResourceVersion]
217
		delete(cfg.Annotations, ResourceVersion)
218
	} else {
219
		cfg.ResourceVersion = time.Now().String()
220
	}
221

222
	ns[cfg.Name] = cfg
223
	return cfg.ResourceVersion, nil
224
}
225

226
func (cr *store) UpdateStatus(cfg config.Config) (string, error) {
227
	return cr.Update(cfg)
228
}
229

230
func (cr *store) Patch(orig config.Config, patchFn config.PatchFunc) (string, error) {
231
	cr.mutex.Lock()
232
	defer cr.mutex.Unlock()
233

234
	gvk := orig.GroupVersionKind
235
	s, ok := cr.schemas.FindByGroupVersionKind(gvk)
236
	if !ok {
237
		return "", fmt.Errorf("unknown type %v", gvk)
238
	}
239

240
	cfg, _ := patchFn(orig)
241
	if !cr.skipValidation {
242
		if _, err := s.ValidateConfig(cfg); err != nil {
243
			return "", err
244
		}
245
	}
246

247
	_, ok = cr.data[gvk]
248
	if !ok {
249
		return "", errNotFound
250
	}
251
	ns, exists := cr.data[gvk][orig.Namespace]
252
	if !exists {
253
		return "", errNotFound
254
	}
255

256
	rev := time.Now().String()
257
	cfg.ResourceVersion = rev
258
	ns[cfg.Name] = cfg
259

260
	return rev, nil
261
}
262

263
// hasConflict checks if the two resources have a conflict, which will block Update calls
264
func hasConflict(existing, replacement config.Config) bool {
265
	if replacement.ResourceVersion == "" {
266
		// We don't care about resource version, so just always overwrite
267
		return false
268
	}
269
	// We set a resource version but its not matched, it is a conflict
270
	if replacement.ResourceVersion != existing.ResourceVersion {
271
		return true
272
	}
273
	return false
274
}
275