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
	"istio.io/istio/pilot/pkg/model"
19
	config2 "istio.io/istio/pkg/config"
20
	"istio.io/istio/pkg/log"
21
)
22

23
const (
24
	// BufferSize specifies the buffer size of event channel
25
	BufferSize = 100
26
)
27

28
// Handler specifies a function to apply on a Config for a given event type
29
type Handler func(config2.Config, config2.Config, model.Event)
30

31
// Monitor provides methods of manipulating changes in the config store
32
type Monitor interface {
33
	Run(<-chan struct{})
34
	AppendEventHandler(config2.GroupVersionKind, Handler)
35
	ScheduleProcessEvent(ConfigEvent)
36
}
37

38
// ConfigEvent defines the event to be processed
39
type ConfigEvent struct {
40
	config config2.Config
41
	old    config2.Config
42
	event  model.Event
43
}
44

45
type configStoreMonitor struct {
46
	store    model.ConfigStore
47
	handlers map[config2.GroupVersionKind][]Handler
48
	eventCh  chan ConfigEvent
49
	// If enabled, events will be handled synchronously
50
	sync bool
51
}
52

53
// NewMonitor returns new Monitor implementation with a default event buffer size.
54
func NewMonitor(store model.ConfigStore) Monitor {
55
	return newBufferedMonitor(store, BufferSize, false)
56
}
57

58
// NewMonitor returns new Monitor implementation which will process events synchronously
59
func NewSyncMonitor(store model.ConfigStore) Monitor {
60
	return newBufferedMonitor(store, BufferSize, true)
61
}
62

63
// NewBufferedMonitor returns new Monitor implementation with the specified event buffer size
64
func newBufferedMonitor(store model.ConfigStore, bufferSize int, sync bool) Monitor {
65
	handlers := make(map[config2.GroupVersionKind][]Handler)
66

67
	for _, s := range store.Schemas().All() {
68
		handlers[s.GroupVersionKind()] = make([]Handler, 0)
69
	}
70

71
	return &configStoreMonitor{
72
		store:    store,
73
		handlers: handlers,
74
		eventCh:  make(chan ConfigEvent, bufferSize),
75
		sync:     sync,
76
	}
77
}
78

79
func (m *configStoreMonitor) ScheduleProcessEvent(configEvent ConfigEvent) {
80
	if m.sync {
81
		m.processConfigEvent(configEvent)
82
	} else {
83
		m.eventCh <- configEvent
84
	}
85
}
86

87
func (m *configStoreMonitor) Run(stop <-chan struct{}) {
88
	if m.sync {
89
		<-stop
90
		return
91
	}
92
	for {
93
		select {
94
		case <-stop:
95
			return
96
		case ce, ok := <-m.eventCh:
97
			if ok {
98
				m.processConfigEvent(ce)
99
			}
100
		}
101
	}
102
}
103

104
func (m *configStoreMonitor) processConfigEvent(ce ConfigEvent) {
105
	if _, exists := m.handlers[ce.config.GroupVersionKind]; !exists {
106
		log.Warnf("Config GroupVersionKind %s does not exist in config store", ce.config.GroupVersionKind)
107
		return
108
	}
109
	m.applyHandlers(ce.old, ce.config, ce.event)
110
}
111

112
func (m *configStoreMonitor) AppendEventHandler(typ config2.GroupVersionKind, h Handler) {
113
	m.handlers[typ] = append(m.handlers[typ], h)
114
}
115

116
func (m *configStoreMonitor) applyHandlers(old config2.Config, config config2.Config, e model.Event) {
117
	for _, f := range m.handlers[config.GroupVersionKind] {
118
		f(old, config, e)
119
	}
120
}
121