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
16

17
import (
18
	"fmt"
19

20
	"k8s.io/apimachinery/pkg/types"
21

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

28
// Controller is an implementation of ConfigStoreController.
29
type Controller struct {
30
	monitor     Monitor
31
	configStore model.ConfigStore
32
	hasSynced   func() bool
33

34
	// If meshConfig.DiscoverySelectors are specified, the namespacesFilter tracks the namespaces this controller watches.
35
	namespacesFilter func(obj interface{}) bool
36
}
37

38
// NewController return an implementation of ConfigStoreController
39
// This is a client-side monitor that dispatches events as the changes are being
40
// made on the client.
41
func NewController(cs model.ConfigStore) *Controller {
42
	out := &Controller{
43
		configStore: cs,
44
		monitor:     NewMonitor(cs),
45
	}
46
	return out
47
}
48

49
// NewSyncController return an implementation of model.ConfigStoreController which processes events synchronously
50
func NewSyncController(cs model.ConfigStore) *Controller {
51
	out := &Controller{
52
		configStore: cs,
53
		monitor:     NewSyncMonitor(cs),
54
	}
55

56
	return out
57
}
58

59
func (c *Controller) RegisterHasSyncedHandler(cb func() bool) {
60
	c.hasSynced = cb
61
}
62

63
func (c *Controller) RegisterEventHandler(kind config.GroupVersionKind, f model.EventHandler) {
64
	c.monitor.AppendEventHandler(kind, f)
65
}
66

67
// HasSynced return whether store has synced
68
// It can be controlled externally (such as by the data source),
69
// otherwise it'll always consider synced.
70
func (c *Controller) HasSynced() bool {
71
	if c.hasSynced != nil {
72
		return c.hasSynced()
73
	}
74
	return true
75
}
76

77
func (c *Controller) Run(stop <-chan struct{}) {
78
	c.monitor.Run(stop)
79
}
80

81
func (c *Controller) Schemas() collection.Schemas {
82
	return c.configStore.Schemas()
83
}
84

85
func (c *Controller) Get(kind config.GroupVersionKind, key, namespace string) *config.Config {
86
	if c.namespacesFilter != nil && !c.namespacesFilter(namespace) {
87
		return nil
88
	}
89
	return c.configStore.Get(kind, key, namespace)
90
}
91

92
func (c *Controller) Create(config config.Config) (revision string, err error) {
93
	if revision, err = c.configStore.Create(config); err == nil {
94
		c.monitor.ScheduleProcessEvent(ConfigEvent{
95
			config: config,
96
			event:  model.EventAdd,
97
		})
98
	}
99
	return
100
}
101

102
func (c *Controller) Update(config config.Config) (newRevision string, err error) {
103
	oldconfig := c.configStore.Get(config.GroupVersionKind, config.Name, config.Namespace)
104
	if newRevision, err = c.configStore.Update(config); err == nil {
105
		c.monitor.ScheduleProcessEvent(ConfigEvent{
106
			old:    *oldconfig,
107
			config: config,
108
			event:  model.EventUpdate,
109
		})
110
	}
111
	return
112
}
113

114
func (c *Controller) UpdateStatus(config config.Config) (newRevision string, err error) {
115
	oldconfig := c.configStore.Get(config.GroupVersionKind, config.Name, config.Namespace)
116
	if newRevision, err = c.configStore.UpdateStatus(config); err == nil {
117
		c.monitor.ScheduleProcessEvent(ConfigEvent{
118
			old:    *oldconfig,
119
			config: config,
120
			event:  model.EventUpdate,
121
		})
122
	}
123
	return
124
}
125

126
func (c *Controller) Patch(orig config.Config, patchFn config.PatchFunc) (newRevision string, err error) {
127
	cfg, typ := patchFn(orig.DeepCopy())
128
	switch typ {
129
	case types.MergePatchType:
130
	case types.JSONPatchType:
131
	default:
132
		return "", fmt.Errorf("unsupported merge type: %s", typ)
133
	}
134
	if newRevision, err = c.configStore.Patch(cfg, patchFn); err == nil {
135
		c.monitor.ScheduleProcessEvent(ConfigEvent{
136
			old:    orig,
137
			config: cfg,
138
			event:  model.EventUpdate,
139
		})
140
	}
141
	return
142
}
143

144
func (c *Controller) Delete(kind config.GroupVersionKind, key, namespace string, resourceVersion *string) error {
145
	if config := c.Get(kind, key, namespace); config != nil {
146
		if err := c.configStore.Delete(kind, key, namespace, resourceVersion); err != nil {
147
			return err
148
		}
149
		c.monitor.ScheduleProcessEvent(ConfigEvent{
150
			config: *config,
151
			event:  model.EventDelete,
152
		})
153
		return nil
154
	}
155
	return fmt.Errorf("delete: config %v/%v/%v does not exist", kind, namespace, key)
156
}
157

158
func (c *Controller) List(kind config.GroupVersionKind, namespace string) []config.Config {
159
	configs := c.configStore.List(kind, namespace)
160
	if c.namespacesFilter != nil {
161
		return slices.Filter(configs, func(config config.Config) bool {
162
			return c.namespacesFilter(config)
163
		})
164
	}
165
	return configs
166
}
167