/
watcher.go
153 lines (141 loc) · 3.61 KB
/
watcher.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
// Copyright (c) 2019 Tigera, Inc. All rights reserved.
package calico
import (
"context"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/watch"
k8swatch "k8s.io/apimachinery/pkg/watch"
"k8s.io/apiserver/pkg/storage"
"github.com/projectcalico/libcalico-go/lib/options"
cwatch "github.com/projectcalico/libcalico-go/lib/watch"
)
// watchChan implements watch.Interface.
type watchChan struct {
resultChan chan watch.Event
pred storage.SelectionPredicate
watcher cwatch.Interface
ctx context.Context
cancel context.CancelFunc
}
func (rs *resourceStore) watchResource(ctx context.Context, resourceVersion string,
p storage.SelectionPredicate, name, namespace string) (k8swatch.Interface, error) {
opts := options.ListOptions{Name: name, Namespace: namespace, ResourceVersion: resourceVersion}
ctx, cancel := context.WithCancel(ctx)
lWatch, err := rs.watch(ctx, rs.client, opts)
if err != nil {
return nil, err
}
wc := &watchChan{
resultChan: make(chan watch.Event),
pred: p,
watcher: lWatch,
ctx: ctx,
cancel: cancel,
}
go wc.run()
return wc, nil
}
func (wc *watchChan) convertEvent(ce cwatch.Event) (res *watch.Event) {
switch ce.Type {
case cwatch.Added:
aapiObject := convertToAAPI(ce.Object)
if aapiObject == nil || !wc.filter(aapiObject) {
return nil
}
res = &watch.Event{
Type: watch.Added,
Object: aapiObject,
}
case cwatch.Deleted:
aapiObject := convertToAAPI(ce.Previous)
if aapiObject == nil || !wc.filter(aapiObject) {
return nil
}
res = &watch.Event{
Type: watch.Deleted,
Object: aapiObject,
}
case cwatch.Modified:
aapiObject := convertToAAPI(ce.Object)
if aapiObject == nil {
return nil
}
if wc.acceptAll() {
res = &watch.Event{
Type: watch.Modified,
Object: aapiObject,
}
return res
}
oldAapiObject := convertToAAPI(ce.Previous)
curObjPasses := wc.filter(aapiObject)
oldObjPasses := wc.filter(oldAapiObject)
switch {
case curObjPasses && oldObjPasses:
res = &watch.Event{
Type: watch.Modified,
Object: aapiObject,
}
case curObjPasses && !oldObjPasses:
res = &watch.Event{
Type: watch.Added,
Object: aapiObject,
}
case !curObjPasses && oldObjPasses:
res = &watch.Event{
Type: watch.Deleted,
Object: oldAapiObject,
}
}
case cwatch.Error:
select {
case <-wc.ctx.Done():
// Any error received after we have cancelled this watcher should be ignored.
return nil
default:
// Fall through if we have not cancelled this watcher.
}
var msg string
if ce.Error != nil {
msg = ce.Error.Error()
}
res = &watch.Event{
Type: watch.Error,
Object: &metav1.Status{
Reason: metav1.StatusReasonInternalError,
Message: msg,
},
}
}
return res
}
func (wc *watchChan) run() {
for e := range wc.watcher.ResultChan() {
we := wc.convertEvent(e)
if we != nil {
wc.resultChan <- *we
if we.Type == watch.Error {
// We use wc.ctx to reap all goroutines. Under whatever condition, we should stop them all.
// It's fine to double cancel.
wc.cancel()
}
}
}
close(wc.resultChan)
}
// filter returns whether a result should be filtered in (true) or filtered out (false).
func (wc *watchChan) filter(obj runtime.Object) bool {
matches, err := wc.pred.Matches(obj)
return matches && err == nil
}
// acceptAll returns true if all results should be filtered in.
func (wc *watchChan) acceptAll() bool {
return wc.pred.Empty()
}
func (wc *watchChan) Stop() {
wc.cancel()
}
func (wc *watchChan) ResultChan() <-chan watch.Event {
return wc.resultChan
}