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
package kube
import (
"context"
"sync"
"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/tools/cache"
)
type Event[T any] struct {
Type watch.EventType
Object *T
}
func Watch[T any](ctx context.Context, r Resource, opts metav1.ListOptions) (<-chan Event[T], func(), error) {
ri, err := r.iface()
if err != nil {
return nil, nil, err
}
w, err := ri.Watch(ctx, opts)
if err != nil {
return nil, nil, err
}
out := make(chan Event[T])
forwarder := supervise.New(ctx)
var once sync.Once
stop := func() {
once.Do(func() {
w.Stop()
_ = forwarder.Stop()
})
}
forwarder.Go("kube: watch", func(ctx context.Context) error {
defer close(out)
for ev := range w.ResultChan() {
e := Event[T]{Type: ev.Type}
if u, ok := ev.Object.(*unstructured.Unstructured); ok {
v, err := Decode[T](u.Object)
if err != nil {
continue
}
e.Object = v
} else if ev.Type != watch.Error {
continue
}
select {
case out <- e:
case <-ctx.Done():
return nil
}
}
return nil
})
return out, stop, nil
}
type Cache[T any] struct {
store cache.Store
ctrl cache.Controller
}
func NewCache[T any](r Resource, opts metav1.ListOptions, changed func()) (*Cache[T], error) {
if _, err := r.iface(); err != nil {
return nil, err
}
kick := func() {
if changed != nil {
changed()
}
}
store, ctrl := cache.NewInformerWithOptions(cache.InformerOptions{ListerWatcher: r.ListWatch(opts), ObjectType: &unstructured.Unstructured{},
Handler: cache.ResourceEventHandlerFuncs{
AddFunc: func(any) { kick() },
UpdateFunc: func(any, any) { kick() },
DeleteFunc: func(any) { kick() },
}})
return &Cache[T]{store: store, ctrl: ctrl}, nil
}
func (c *Cache[T]) Run(ctx context.Context) { c.ctrl.RunWithContext(ctx) }
func (c *Cache[T]) HasSynced() bool { return c.ctrl.HasSynced() }
func (c *Cache[T]) WaitSynced(ctx context.Context) bool {
return c.ctrl.HasSynced() || cache.WaitForCacheSync(ctx.Done(), c.ctrl.HasSynced)
}
func (c *Cache[T]) List() []T {
var out []T
for _, item := range c.store.List() {
u, ok := item.(*unstructured.Unstructured)
if !ok {
continue
}
v, err := Decode[T](u.Object)
if err != nil {
continue
}
out = append(out, *v)
}
return out
}
