Skip to content

Commit f87db1f

Browse files
embikvincepristtts
committed
WIP: Add fleet-namespace example for pseudo-multi-cluster support
On-behalf-of: SAP [email protected] Co-authored-by: Vince Prignano <[email protected]> Co-authored-by: Dr. Stefan Schimanski <[email protected]> Signed-off-by: Marvin Beckers <[email protected]>
1 parent 3d3c7c8 commit f87db1f

File tree

7 files changed

+1060
-0
lines changed

7 files changed

+1060
-0
lines changed

‎examples/fleet-namespace/cache.go

+204
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,204 @@
1+
/*
2+
Copyright 2024 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package main
18+
19+
import (
20+
"context"
21+
"errors"
22+
"fmt"
23+
"time"
24+
25+
apierrors "k8s.io/apimachinery/pkg/api/errors"
26+
"k8s.io/apimachinery/pkg/api/meta"
27+
"k8s.io/apimachinery/pkg/runtime"
28+
"k8s.io/apimachinery/pkg/runtime/schema"
29+
toolscache "k8s.io/client-go/tools/cache"
30+
"sigs.k8s.io/controller-runtime/pkg/cache"
31+
"sigs.k8s.io/controller-runtime/pkg/client"
32+
)
33+
34+
const (
35+
ClusterNameIndex = "cluster/name"
36+
ClusterIndex = "cluster"
37+
)
38+
39+
var _ cache.Cache = &NamespacedCache{}
40+
41+
type NamespacedCache struct {
42+
clusterName string
43+
cache.Cache
44+
}
45+
46+
func (c *NamespacedCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
47+
if key.Namespace != "default" {
48+
return apierrors.NewNotFound(schema.GroupResource{}, key.Name)
49+
}
50+
key.Namespace = c.clusterName
51+
if err := c.Cache.Get(ctx, key, obj, opts...); err != nil {
52+
return err
53+
}
54+
obj.SetNamespace("default")
55+
return nil
56+
}
57+
58+
func (c *NamespacedCache) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
59+
var listOpts client.ListOptions
60+
for _, o := range opts {
61+
o.ApplyToList(&listOpts)
62+
}
63+
64+
switch {
65+
case listOpts.FieldSelector != nil:
66+
reqs := listOpts.FieldSelector.Requirements()
67+
flds := make(map[string]string, len(reqs))
68+
for i := range reqs {
69+
flds[fmt.Sprintf("cluster/%s", reqs[i].Field)] = fmt.Sprintf("%s/%s", c.clusterName, reqs[i].Value)
70+
}
71+
opts = append(opts, client.MatchingFields(flds))
72+
case listOpts.Namespace != "":
73+
opts = append(opts, client.MatchingFields(map[string]string{ClusterIndex: c.clusterName}))
74+
if c.clusterName == "*" {
75+
listOpts.Namespace = ""
76+
} else if listOpts.Namespace == "default" {
77+
listOpts.Namespace = c.clusterName
78+
}
79+
default:
80+
opts = append(opts, client.MatchingFields(map[string]string{ClusterIndex: c.clusterName}))
81+
}
82+
83+
if err := c.Cache.List(ctx, list, opts...); err != nil {
84+
return err
85+
}
86+
87+
return meta.EachListItem(list, func(obj runtime.Object) error {
88+
obj.(client.Object).SetNamespace("default")
89+
return nil
90+
})
91+
}
92+
93+
func (c *NamespacedCache) GetInformer(ctx context.Context, obj client.Object, opts ...cache.InformerGetOption) (cache.Informer, error) {
94+
inf, err := c.Cache.GetInformer(ctx, obj, opts...)
95+
if err != nil {
96+
return nil, err
97+
}
98+
return &ScopedInformer{clusterName: c.clusterName, Informer: inf}, nil
99+
}
100+
101+
func (c *NamespacedCache) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind, opts ...cache.InformerGetOption) (cache.Informer, error) {
102+
inf, err := c.Cache.GetInformerForKind(ctx, gvk, opts...)
103+
if err != nil {
104+
return nil, err
105+
}
106+
return &ScopedInformer{clusterName: c.clusterName, Informer: inf}, nil
107+
}
108+
109+
func (c *NamespacedCache) RemoveInformer(ctx context.Context, obj client.Object) error {
110+
return errors.New("informer cannot be removed from scoped cache")
111+
}
112+
113+
type ScopedInformer struct {
114+
clusterName string
115+
cache.Informer
116+
}
117+
118+
func (i *ScopedInformer) AddEventHandler(handler toolscache.ResourceEventHandler) (toolscache.ResourceEventHandlerRegistration, error) {
119+
return i.Informer.AddEventHandler(toolscache.ResourceEventHandlerDetailedFuncs{
120+
AddFunc: func(obj interface{}, isInInitialList bool) {
121+
cobj := obj.(client.Object)
122+
if cobj.GetNamespace() == i.clusterName {
123+
cobj := cobj.DeepCopyObject().(client.Object)
124+
cobj.SetNamespace("default")
125+
handler.OnAdd(cobj, isInInitialList)
126+
}
127+
},
128+
UpdateFunc: func(oldObj, newObj interface{}) {
129+
cobj := newObj.(client.Object)
130+
if cobj.GetNamespace() == i.clusterName {
131+
cobj := cobj.DeepCopyObject().(client.Object)
132+
cobj.SetNamespace("default")
133+
handler.OnUpdate(oldObj, cobj)
134+
}
135+
},
136+
DeleteFunc: func(obj interface{}) {
137+
if tombStone, ok := obj.(toolscache.DeletedFinalStateUnknown); ok {
138+
obj = tombStone.Obj
139+
}
140+
cobj := obj.(client.Object)
141+
if cobj.GetNamespace() == i.clusterName {
142+
cobj := cobj.DeepCopyObject().(client.Object)
143+
cobj.SetNamespace("default")
144+
handler.OnDelete(cobj)
145+
}
146+
},
147+
})
148+
}
149+
150+
func (i *ScopedInformer) AddEventHandlerWithResyncPeriod(handler toolscache.ResourceEventHandler, resyncPeriod time.Duration) (toolscache.ResourceEventHandlerRegistration, error) {
151+
return i.Informer.AddEventHandlerWithResyncPeriod(toolscache.ResourceEventHandlerDetailedFuncs{
152+
AddFunc: func(obj interface{}, isInInitialList bool) {
153+
cobj := obj.(client.Object)
154+
if cobj.GetNamespace() == i.clusterName {
155+
cobj := cobj.DeepCopyObject().(client.Object)
156+
cobj.SetNamespace("default")
157+
handler.OnAdd(cobj, isInInitialList)
158+
}
159+
},
160+
UpdateFunc: func(oldObj, newObj interface{}) {
161+
cobj := newObj.(client.Object)
162+
if cobj.GetNamespace() == i.clusterName {
163+
cobj := cobj.DeepCopyObject().(client.Object)
164+
cobj.SetNamespace("default")
165+
handler.OnUpdate(oldObj, cobj)
166+
}
167+
},
168+
DeleteFunc: func(obj interface{}) {
169+
if tombStone, ok := obj.(toolscache.DeletedFinalStateUnknown); ok {
170+
obj = tombStone.Obj
171+
}
172+
cobj := obj.(client.Object)
173+
if cobj.GetNamespace() == i.clusterName {
174+
cobj := cobj.DeepCopyObject().(client.Object)
175+
cobj.SetNamespace("default")
176+
handler.OnDelete(cobj)
177+
}
178+
},
179+
}, resyncPeriod)
180+
}
181+
182+
func (i *ScopedInformer) AddIndexers(indexers toolscache.Indexers) error {
183+
return errors.New("indexes cannot be added to scoped informers")
184+
}
185+
186+
type NamespaceScopeableCache struct {
187+
cache.Cache
188+
}
189+
190+
func (f *NamespaceScopeableCache) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
191+
return f.IndexField(ctx, obj, "cluster/"+field, func(obj client.Object) []string {
192+
keys := extractValue(obj)
193+
withCluster := make([]string, len(keys)*2)
194+
for i, key := range keys {
195+
withCluster[i] = fmt.Sprintf("%s/%s", obj.GetNamespace(), key)
196+
withCluster[i+len(keys)] = fmt.Sprintf("*/%s", key)
197+
}
198+
return withCluster
199+
})
200+
}
201+
202+
func (f *NamespaceScopeableCache) Start(ctx context.Context) error {
203+
return nil // no-op as this is shared
204+
}

‎examples/fleet-namespace/client.go

+91
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
/*
2+
Copyright 2024 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package main
18+
19+
import (
20+
"context"
21+
22+
apierrors "k8s.io/apimachinery/pkg/api/errors"
23+
"k8s.io/apimachinery/pkg/api/meta"
24+
"k8s.io/apimachinery/pkg/runtime"
25+
"k8s.io/apimachinery/pkg/runtime/schema"
26+
"sigs.k8s.io/controller-runtime/pkg/client"
27+
)
28+
29+
var _ client.Client = &NamespacedClient{}
30+
31+
type NamespacedClient struct {
32+
clusterName string
33+
client.Client
34+
}
35+
36+
func (n *NamespacedClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
37+
if key.Namespace != "default" {
38+
return apierrors.NewNotFound(schema.GroupResource{}, key.Name)
39+
}
40+
key.Namespace = n.clusterName
41+
if err := n.Client.Get(ctx, key, obj, opts...); err != nil {
42+
return err
43+
}
44+
obj.SetNamespace("default")
45+
return nil
46+
}
47+
48+
func (n *NamespacedClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
49+
var copts client.ListOptions
50+
for _, o := range opts {
51+
o.ApplyToList(&copts)
52+
}
53+
if copts.Namespace != "default" {
54+
return apierrors.NewNotFound(schema.GroupResource{}, copts.Namespace)
55+
}
56+
if err := n.Client.List(ctx, list, append(opts, client.InNamespace(n.clusterName))...); err != nil {
57+
return err
58+
}
59+
return meta.EachListItem(list, func(obj runtime.Object) error {
60+
obj.(client.Object).SetNamespace("default")
61+
return nil
62+
})
63+
}
64+
65+
func (n *NamespacedClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error {
66+
panic("implement me")
67+
}
68+
69+
func (n *NamespacedClient) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) error {
70+
panic("implement me")
71+
}
72+
73+
func (n *NamespacedClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) error {
74+
panic("implement me")
75+
}
76+
77+
func (n *NamespacedClient) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) error {
78+
panic("implement me")
79+
}
80+
81+
func (n *NamespacedClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error {
82+
panic("implement me")
83+
}
84+
85+
func (n *NamespacedClient) Status() client.SubResourceWriter {
86+
panic("implement me")
87+
}
88+
89+
func (n *NamespacedClient) SubResource(subResource string) client.SubResourceClient {
90+
panic("implement me")
91+
}

‎examples/fleet-namespace/cluster.go

+70
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
/*
2+
Copyright 2024 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package main
18+
19+
import (
20+
"context"
21+
22+
"k8s.io/apimachinery/pkg/types"
23+
"k8s.io/client-go/tools/record"
24+
"sigs.k8s.io/controller-runtime/pkg/cache"
25+
"sigs.k8s.io/controller-runtime/pkg/client"
26+
"sigs.k8s.io/controller-runtime/pkg/cluster"
27+
"sigs.k8s.io/controller-runtime/pkg/reconcile"
28+
)
29+
30+
type clusterRequest struct {
31+
reconcile.Request
32+
ClusterName string
33+
}
34+
35+
// String returns the general purpose string representation.
36+
func (cr *clusterRequest) String() string {
37+
if cr.ClusterName == "" {
38+
return cr.NamespacedName.String()
39+
}
40+
return "cluster://" + cr.ClusterName + string(types.Separator) + cr.NamespacedName.String()
41+
}
42+
43+
type NamespacedCluster struct {
44+
clusterName string
45+
cluster.Cluster
46+
}
47+
48+
func (c *NamespacedCluster) Name() string {
49+
return c.clusterName
50+
}
51+
52+
func (c *NamespacedCluster) GetCache() cache.Cache {
53+
return &NamespacedCache{clusterName: c.clusterName, Cache: c.Cluster.GetCache()}
54+
}
55+
56+
func (c *NamespacedCluster) GetClient() client.Client {
57+
return &NamespacedClient{clusterName: c.clusterName, Client: c.Cluster.GetClient()}
58+
}
59+
60+
func (c *NamespacedCluster) GetEventRecorderFor(name string) record.EventRecorder {
61+
panic("implement me")
62+
}
63+
64+
func (c *NamespacedCluster) GetAPIReader() client.Reader {
65+
return c.GetClient()
66+
}
67+
68+
func (c *NamespacedCluster) Start(ctx context.Context) error {
69+
return nil // no-op as this is shared
70+
}

0 commit comments

Comments
 (0)