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
package marketintents
import (
"context"
"fmt"
apierrors "k8s.io/apimachinery/pkg/api/errors"
apimeta "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
exposurev1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/exposure/v1alpha1"
"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
market "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market"
ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
)
const clusterAnnotation = "kcp.io/cluster"
var kindGVKs = map[string]schema.GroupVersionKind{
market.KindBid: v1alpha1.GroupVersion.WithKind(market.KindBid),
market.KindComputeVM: v1alpha1.GroupVersion.WithKind(market.KindComputeVM),
market.KindStateSync: v1alpha1.GroupVersion.WithKind(market.KindStateSync),
market.KindComputeContract: v1alpha1.GroupVersion.WithKind(market.KindComputeContract),
market.KindExposure: exposurev1alpha1.GroupVersion.WithKind(market.KindExposure),
market.KindMarketIdentity: v1alpha1.GroupVersion.WithKind(market.KindMarketIdentity),
}
func (s *subscriber) matches(workspace, kind string) bool {
return s.workspaces[workspace] && (len(s.kinds) == 0 || s.kinds[kind])
}
func (c *Controller) listed(ctx context.Context, s *subscriber) (map[key]string, error) {
out := map[key]string{}
for workspace := range s.workspaces {
if c.scope != nil && !c.scope(workspace) {
continue
}
cl, err := c.clients(workspace)
if err != nil {
return nil, fmt.Errorf("marketintents: the client of %s: %w", workspace, err)
}
for kind, gvk := range kindGVKs {
if !s.matches(workspace, kind) {
continue
}
list := &metav1.PartialObjectMetadataList{}
list.SetGroupVersionKind(gvk.GroupVersion().WithKind(gvk.Kind + "List"))
if err := cl.List(ctx, list); err != nil {
if apierrors.IsNotFound(err) || apimeta.IsNoMatchError(err) {
continue
}
return nil, fmt.Errorf("marketintents: listing the %s objects of %s: %w", kind, workspace, err)
}
for _, item := range list.Items {
out[key{workspace: workspace, kind: kind, name: item.GetName()}] = item.GetAnnotations()[clusterAnnotation]
}
}
}
return out, nil
}
func (c *Controller) clusterOfLocked(k key, annotated string) string {
if cluster, ok := c.clusters[k]; ok {
return cluster
}
if annotated != "" {
return annotated
}
for other, cluster := range c.clusters {
if other.workspace == k.workspace {
return cluster
}
}
return ""
}
func (c *Controller) observed(s *subscriber, k key) {
c.mu.Lock()
if s.pending == nil || s.fired {
c.mu.Unlock()
return
}
delete(s.pending, k)
c.mu.Unlock()
c.settle(s)
}
func (c *Controller) settle(s *subscriber) {
c.mu.Lock()
if s.pending == nil || s.fired || len(s.pending) > 0 {
c.mu.Unlock()
return
}
s.fired = true
replay := c.replayLocked(s)
c.mu.Unlock()
s.synced()
c.enqueue(replay)
}
func (c *Controller) replayLocked(s *subscriber) map[string][]ctlruntime.Request {
replay := map[string][]ctlruntime.Request{}
for k, cluster := range c.clusters {
if s.matches(k.workspace, k.kind) {
replay[k.kind] = append(replay[k.kind], ctlruntime.Request{Cluster: cluster, Name: k.name})
}
}
return replay
}
func (c *Controller) enqueue(replay map[string][]ctlruntime.Request) {
c.mu.Lock()
queues := map[string]*ctlruntime.Queue{}
for kind := range replay {
queues[kind] = c.events[kind]
}
c.mu.Unlock()
for kind, reqs := range replay {
if q := queues[kind]; q != nil {
q.Add(reqs...)
}
}
}
