Skip to content

Commit 4a5927c

Browse files
cemakdcemakd
committed
Refactor to move tenant related logic into tennacy package
1 parent f5ce0fb commit 4a5927c

File tree

3 files changed

+122
-54
lines changed

3 files changed

+122
-54
lines changed

‎pkg/gce-cloud-provider/compute/gce.go

Lines changed: 48 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@ import (
2929
"golang.org/x/time/rate"
3030
"google.golang.org/api/option"
3131
"gopkg.in/gcfg.v1"
32-
"k8s.io/client-go/tools/cache"
3332
"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
3433
"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute/tenancy"
3534

@@ -38,6 +37,7 @@ import (
3837
"golang.org/x/oauth2"
3938
computebeta "google.golang.org/api/compute/v0.beta"
4039
"google.golang.org/api/compute/v1"
40+
computev1 "google.golang.org/api/compute/v1"
4141
"google.golang.org/api/googleapi"
4242
"k8s.io/apimachinery/pkg/util/wait"
4343
"k8s.io/klog/v2"
@@ -156,6 +156,13 @@ func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath s
156156
return nil, err
157157
}
158158

159+
// token, err := tokenSource.Token()
160+
// if err != nil {
161+
// klog.Errorf("error getting initial token.: %v", err)
162+
// } else {
163+
// klog.Infof("test token success: %+v", token)
164+
// }
165+
159166
svc, err := createCloudService(ctx, vendorVersion, tokenSource, computeEndpoint, computeEnvironment)
160167
if err != nil {
161168
return nil, err
@@ -184,65 +191,55 @@ func CreateCloudProvider(ctx context.Context, vendorVersion string, configPath s
184191
listInstancesConfig: listInstancesConfig,
185192
// GCP has a rate limit of 600 requests per minute, restricting
186193
// here to 8 requests per second.
187-
tagsRateLimiter: common.NewLimiter(gcpTagsRequestRateLimit, gcpTagsRequestTokenBucketSize, true),
194+
tagsRateLimiter: common.NewLimiter(gcpTagsRequestRateLimit, gcpTagsRequestTokenBucketSize, true),
195+
tenantServiceMap: make(map[string]*compute.Service),
188196
}
189197

190198
if multiTenancyEnabled {
191199
klog.Info("Setting up multitenancy")
192-
// Setup informant for tenant CR to automatically create tenant specific clients with tenant identities
193200
ti, err := tenancy.NewTenantsInformer(multiTenancyEnabled)
194201
if err != nil {
195202
return nil, fmt.Errorf("failed initializing tenant informer: %w", err)
196203
}
197204
cp.TenantInformer = ti
198-
cp.tenantServiceMap = map[string]*compute.Service{}
199-
cp.TenantInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
200-
AddFunc: func(obj any) {
201-
// Handle tenant creation
202-
klog.Infof("Tenant %s created", obj)
203-
204-
tenantMeta, err := tenancy.GetMetadataFromTenantCR(obj)
205-
if err != nil {
206-
klog.Errorf("error while extracting tenant metadata: %v", err)
207-
}
208-
209-
tenantServiceMutex.Lock()
210-
defer tenantServiceMutex.Unlock()
211-
212-
if _, ok := cp.tenantServiceMap[tenantMeta.ProjectNumber]; ok {
213-
klog.Infof("Tenant GCE client already exists, skipping GCE client instantiation for tenant(%s) with project number(%s)", tenantMeta.TenantName, tenantMeta.ProjectNumber)
214-
return
215-
}
216-
217-
region, err := common.GetRegionFromZones([]string{zone})
218-
if err != nil {
219-
klog.Errorf("error getting region from zone(%s): %v", zone, err)
220-
return
221-
}
222-
tokenSource, err := NewTenantTokenSource(tenantMeta, region, configFile.Global.TokenURL, configFile.Global.TokenBody)
223-
if err != nil {
224-
klog.Errorf("error during tenant token generation: %v", err.Error())
225-
}
226-
227-
svc, err := createCloudService(ctx, vendorVersion, tokenSource, computeEndpoint, computeEnvironment)
228-
if err != nil {
229-
klog.Errorf("error while creating compute service with tenant identity: %v", err)
230-
return
231-
}
232-
cp.tenantServiceMap[tenantMeta.ProjectNumber] = svc
233-
},
234-
UpdateFunc: func(oldObj, newObj any) {},
235-
DeleteFunc: func(obj any) {
236-
klog.Infof("Tenant %s deleted", obj)
237-
tenantMeta, err := tenancy.GetMetadataFromTenantCR(obj)
238-
if err != nil {
239-
klog.Errorf("error while extracting teantn metadata: %v", err)
240-
}
241-
tenantServiceMutex.Lock()
242-
defer tenantServiceMutex.Unlock()
243-
delete(cp.tenantServiceMap, tenantMeta.ProjectNumber)
244-
},
245-
})
205+
addTenantCallback := func(tenantMeta *tenancy.Metadata, projectZone string) (*computev1.Service, error) {
206+
klog.Infof("Executing AddFunc callback for tenant: %s (Project: %s)", tenantMeta.TenantName, tenantMeta.ProjectNumber)
207+
208+
region, err := common.GetRegionFromZones([]string{zone})
209+
if err != nil {
210+
klog.Errorf("Error getting region from zone(%s) for tenant %s: %v", zone, tenantMeta.TenantName, err)
211+
return nil, fmt.Errorf("error getting region from zone(%s): %w", zone, err)
212+
}
213+
214+
tenantTokenSource, err := NewTenantTokenSource(tenantMeta, region, configFile.Global.TokenURL, configFile.Global.TokenBody)
215+
if err != nil {
216+
klog.Errorf("Error during tenant token source generation for %s: %v", tenantMeta.TenantName, err.Error())
217+
return nil, fmt.Errorf("error during tenant token source generation: %w", err)
218+
}
219+
220+
tenantComputeService, err := createCloudService(ctx, vendorVersion, tenantTokenSource, computeEndpoint, computeEnvironment)
221+
if err != nil {
222+
klog.Errorf("Error while creating compute service with tenant identity for %s: %v", tenantMeta.TenantName, err)
223+
return nil, fmt.Errorf("error while creating compute service with tenant identity: %w", err)
224+
}
225+
klog.Infof("Successfully created compute service for tenant %s (Project: %s)", tenantMeta.TenantName, tenantMeta.ProjectNumber)
226+
return tenantComputeService, nil
227+
}
228+
229+
lifecycleHandler := tenancy.TenantLifecycleHandler{
230+
AddFunc: addTenantCallback,
231+
}
232+
233+
err = tenancy.RegisterTenantEventHandlers(
234+
cp.TenantInformer,
235+
lifecycleHandler,
236+
zone,
237+
cp.tenantServiceMap,
238+
&tenantServiceMutex,
239+
)
240+
if err != nil {
241+
return nil, fmt.Errorf("failed to register tenant event handlers: %w", err)
242+
}
246243
}
247244

248245
return cp, nil

‎pkg/gce-cloud-provider/compute/tenancy/informer.go

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,16 @@ package tenancy
22

33
import (
44
"fmt"
5+
"sync"
56
"time"
67

8+
computev1 "google.golang.org/api/compute/v1"
79
"k8s.io/apimachinery/pkg/runtime/schema"
810
"k8s.io/client-go/dynamic"
911
"k8s.io/client-go/dynamic/dynamicinformer"
1012
"k8s.io/client-go/rest"
1113
"k8s.io/client-go/tools/cache"
14+
"k8s.io/klog/v2"
1215
)
1316

1417
// TenantsInformer is an interface that wraps a cache.SharedIndexInformer
@@ -19,6 +22,13 @@ type TenantsInformer interface {
1922
HasSynced() bool
2023
}
2124

25+
// TenantLifecycleHandler defines callbacks for tenant lifecycle events.
26+
// TenantMetadata should be the struct returned by GetMetadataFromTenantCR.
27+
type TenantLifecycleHandler struct {
28+
AddFunc func(tenantMeta *Metadata, zone string) (*computev1.Service, error)
29+
DeleteFunc func(tenantMeta *Metadata)
30+
}
31+
2232
// newDynamicClientForConfig creates a new dynamic client for the given
2333
// configuration.
2434
//
@@ -51,6 +61,67 @@ func NewTenantsInformer(isMultiTenantCluster bool) (TenantsInformer, error) {
5161
return dynamicFactory.ForResource(gvr).Informer(), nil
5262
}
5363

64+
func RegisterTenantEventHandlers(ti TenantsInformer, handler TenantLifecycleHandler, zone string, tenantServiceMap map[string]*computev1.Service, mutex *sync.Mutex) error {
65+
if ti == nil {
66+
return fmt.Errorf("TenantsInformer cannot be nil")
67+
}
68+
69+
_, err := ti.AddEventHandler(cache.ResourceEventHandlerFuncs{
70+
AddFunc: func(obj any) {
71+
klog.Infof("Tenant CR created: %v", obj)
72+
tenantMeta, err := GetMetadataFromTenantCR(obj)
73+
if err != nil {
74+
klog.Errorf("Error extracting tenant metadata from CR: %v", err)
75+
return
76+
}
77+
78+
mutex.Lock()
79+
defer mutex.Unlock()
80+
if _, ok := tenantServiceMap[tenantMeta.ProjectNumber]; ok {
81+
klog.Infof("Tenant GCE client already exists for tenant project number %s, skipping GCE client instantiation.", tenantMeta.ProjectNumber)
82+
mutex.Unlock()
83+
return
84+
}
85+
86+
svc, err := handler.AddFunc(&tenantMeta, zone)
87+
if err != nil {
88+
klog.Errorf("Error in AddFunc callback for tenant %s (project %s): %v", tenantMeta.TenantName, tenantMeta.ProjectNumber, err)
89+
return
90+
}
91+
92+
if svc != nil {
93+
tenantServiceMap[tenantMeta.ProjectNumber] = svc
94+
klog.Infof("Successfully processed AddFunc for tenant %s (project %s) and updated service map.", tenantMeta.TenantName, tenantMeta.ProjectNumber)
95+
}
96+
},
97+
DeleteFunc: func(obj any) {
98+
klog.Infof("Tenant CR deleted: %v", obj)
99+
tenantMeta, err := GetMetadataFromTenantCR(obj)
100+
if err != nil {
101+
klog.Errorf("Error while extracting tenant metadata on delete: %v", err)
102+
return
103+
}
104+
105+
mutex.Lock()
106+
defer mutex.Unlock()
107+
if _, ok := tenantServiceMap[tenantMeta.ProjectNumber]; ok {
108+
delete(tenantServiceMap, tenantMeta.ProjectNumber)
109+
klog.Infof("Deleted GCE client for tenant project number %s from map.", tenantMeta.ProjectNumber)
110+
} else {
111+
klog.Warningf("Attempted to delete GCE client for tenant project %s, but it was not found in the map.", tenantMeta.ProjectNumber)
112+
}
113+
114+
if handler.DeleteFunc != nil {
115+
handler.DeleteFunc(&tenantMeta)
116+
}
117+
},
118+
})
119+
if err != nil {
120+
return fmt.Errorf("failed to add event handler to tenant informer: %w", err)
121+
}
122+
return nil
123+
}
124+
54125
// noopTenantsInformer is a TenantsInformer that does nothing and
55126
// always returns true for HasSynced.
56127
// This should only be used when GKE multi-tenancy is disabled or in tests.

‎pkg/gce-cloud-provider/compute/token_source.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ func NewAltTokenSource(tokenURL, tokenBody string) oauth2.TokenSource {
9393
return oauth2.ReuseTokenSource(nil, a)
9494
}
9595

96-
func NewTenantTokenSource(tenantMeta tenancy.Metadata, region, existingTokenURL, existingTokenBody string) (oauth2.TokenSource, error) {
96+
func NewTenantTokenSource(tenantMeta *tenancy.Metadata, region, existingTokenURL, existingTokenBody string) (oauth2.TokenSource, error) {
9797
tenantTokenUrl, err := getTenantTokenURL(tenantMeta, existingTokenURL)
9898
if err != nil {
9999
return nil, err
@@ -105,7 +105,7 @@ func NewTenantTokenSource(tenantMeta tenancy.Metadata, region, existingTokenURL,
105105
return NewAltTokenSource(tenantTokenUrl, tenantTokenBody), nil
106106
}
107107

108-
func getTenantTokenURL(tenantMeta tenancy.Metadata, existingTokenURL string) (string, error) {
108+
func getTenantTokenURL(tenantMeta *tenancy.Metadata, existingTokenURL string) (string, error) {
109109
location := extractLocationFromTokenURL(existingTokenURL)
110110
if location == "" {
111111
return "", fmt.Errorf("could not extract location from existing token URL: %s", existingTokenURL)
@@ -136,7 +136,7 @@ func extractLocationFromTokenURL(tokenURL string) string {
136136
return ""
137137
}
138138

139-
func getTenantTokenBody(tenantMeta tenancy.Metadata, existingTokenBody string) (string, error) {
139+
func getTenantTokenBody(tenantMeta *tenancy.Metadata, existingTokenBody string) (string, error) {
140140
// Check if the token body is a quoted JSON string
141141
// Quoted example: "{\"projectNumber\":12345,\"clusterId\":\"example-cluster\"}"
142142
// Non-quoted example: {"projectNumber":12345,"clusterId":"example-cluster"}

0 commit comments

Comments
 (0)