Skip to content

Handle Multiple Projects In MIG Cache #8512

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
+73 −20
Open

Handle Multiple Projects In MIG Cache #8512

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion cluster-autoscaler/cloudprovider/gce/gce_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ func CreateGceManager(configReader io.Reader, discoveryOpts cloudprovider.NodeGr
cache: cache,
GceService: gceService,
migLister: migLister,
migInfoProvider: NewCachingMigInfoProvider(cache, migLister, gceService, projectId, concurrentGceRefreshes, migInstancesMinRefreshWaitTime, bulkGceMigInstancesListingEnabled),
migInfoProvider: NewCachingMigInfoProvider(cache, migLister, gceService, projectId, concurrentGceRefreshes, migInstancesMinRefreshWaitTime, bulkGceMigInstancesListingEnabled, false),
location: location,
regional: regional,
projectId: projectId,
Expand Down
6 changes: 3 additions & 3 deletions cluster-autoscaler/cloudprovider/gce/gce_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,7 +253,7 @@ const listInstanceGroupManagerResponsePartTemplate = `
}
}
],
"instanceGroup": "https://www.googleapis.com/compute/v1/projects/lukaszos-gke-dev2/zones/%v/instanceGroups/%v",
"instanceGroup": "https://www.googleapis.com/compute/v1/projects/project1/zones/%v/instanceGroups/%v",
"baseInstanceName": "%s",
"fingerprint": "ASJwTpesjDI=",
"currentActions": {
Expand All @@ -271,7 +271,7 @@ const listInstanceGroupManagerResponsePartTemplate = `
"isStable": true
},
"targetSize": %v,
"selfLink": "https://www.googleapis.com/compute/v1/projects/lukaszos-gke-dev2/zones/us-west1-b/instanceGroupManagers/gke-blah-default-pool-67b773a0-grp",
"selfLink": "https://www.googleapis.com/compute/v1/projects/project1/zones/us-west1-b/instanceGroupManagers/gke-blah-default-pool-67b773a0-grp",
"updatePolicy": {
"type": "OPPORTUNISTIC",
"minimalAction": "REPLACE",
Expand Down Expand Up @@ -354,7 +354,7 @@ func newTestGceManager(t *testing.T, testServerURL string, regional bool) *gceMa
manager := &gceManagerImpl{
cache: cache,
migLister: migLister,
migInfoProvider: NewCachingMigInfoProvider(cache, migLister, gceService, projectId, 1, 0*time.Second, false),
migInfoProvider: NewCachingMigInfoProvider(cache, migLister, gceService, projectId, 1, 0*time.Second, false, false),
GceService: gceService,
projectId: projectId,
regional: regional,
Expand Down
36 changes: 34 additions & 2 deletions cluster-autoscaler/cloudprovider/gce/mig_info_provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
"fmt"
"net/url"
"path"
"regexp"
"strings"
"sync"
"time"
Expand Down Expand Up @@ -62,6 +63,11 @@ type timeProvider interface {
Now() time.Time
}

var (
// Compile a regular expression to find the text between "projects/" and the next "/".
migProjectSelfLinkRe = regexp.MustCompile(`projects/([^/]+)`)
)

type cachingMigInfoProvider struct {
migInfoMutex sync.Mutex
cache *GceCache
Expand All @@ -73,6 +79,7 @@ type cachingMigInfoProvider struct {
migInstancesMinRefreshWaitTime time.Duration
timeProvider timeProvider
bulkGceMigInstancesListingEnabled bool
multiProjectCachingEnabled bool
}

type realTime struct{}
Expand All @@ -82,7 +89,7 @@ func (r *realTime) Now() time.Time {
}

// NewCachingMigInfoProvider creates an instance of caching MigInfoProvider
func NewCachingMigInfoProvider(cache *GceCache, migLister MigLister, gceClient AutoscalingGceClient, projectId string, concurrentGceRefreshes int, migInstancesMinRefreshWaitTime time.Duration, bulkGceMigInstancesListingEnabled bool) MigInfoProvider {
func NewCachingMigInfoProvider(cache *GceCache, migLister MigLister, gceClient AutoscalingGceClient, projectId string, concurrentGceRefreshes int, migInstancesMinRefreshWaitTime time.Duration, bulkGceMigInstancesListingEnabled bool, multiProjectCachingEnabled bool) MigInfoProvider {
return &cachingMigInfoProvider{
cache: cache,
migLister: migLister,
Expand All @@ -92,6 +99,7 @@ func NewCachingMigInfoProvider(cache *GceCache, migLister MigLister, gceClient A
migInstancesMinRefreshWaitTime: migInstancesMinRefreshWaitTime,
timeProvider: &realTime{},
bulkGceMigInstancesListingEnabled: bulkGceMigInstancesListingEnabled,
multiProjectCachingEnabled: multiProjectCachingEnabled,
}
}

Expand Down Expand Up @@ -479,8 +487,19 @@ func (c *cachingMigInfoProvider) fillMigInfoCache() error {

for idx, zone := range zones {
for _, zoneMig := range migs[idx] {
projectId := c.projectId
if c.multiProjectCachingEnabled {
var err error
projectId, err = extractProjectWithRegex(zoneMig.SelfLink)
if err != nil {
// At this point we assume its the default project but this could eventually lead to a cache miss
// if the project information is incorrect.
projectId = c.projectId
klog.Errorf("Unable to extract projectID from MIG self link: %s, err: %v", zoneMig.SelfLink, err)
}
}
zoneMigRef := GceRef{
c.projectId,
projectId,
zone,
zoneMig.Name,
}
Expand Down Expand Up @@ -508,6 +527,19 @@ func (c *cachingMigInfoProvider) fillMigInfoCache() error {
return nil
}

// extractProjectWithRegex uses a regular expression to find and return the project name
// from the selfLink of a MIG.
func extractProjectWithRegex(selflink string) (string, error) {
// FindStringSubmatch returns an array with the full match and all captured groups.
// matches[0] will be the full matched string (e.g., "/projects/some-project").
// matches[1] will be the content of the first capturing group (e.g., "some-project").
matches := migProjectSelfLinkRe.FindStringSubmatch(selflink)
if len(matches) < 2 {
return "", fmt.Errorf("could not find project name in self link: %s", selflink)
}
return matches[1], nil
}

func (c *cachingMigInfoProvider) getRegisteredMigRefs() map[GceRef]bool {
migRefs := make(map[GceRef]bool)
for _, mig := range c.migLister.GetMigs() {
Expand Down
49 changes: 35 additions & 14 deletions cluster-autoscaler/cloudprovider/gce/mig_info_provider_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,7 +264,7 @@ func TestFillMigInstances(t *testing.T) {
fetchMigInstances: fetchMigInstancesWithCounter(newInstances, callCounter),
}

provider, ok := NewCachingMigInfoProvider(tc.cache, NewMigLister(tc.cache), client, mig.GceRef().Project, 1, time.Hour, false).(*cachingMigInfoProvider)
provider, ok := NewCachingMigInfoProvider(tc.cache, NewMigLister(tc.cache), client, mig.GceRef().Project, 1, time.Hour, false, false).(*cachingMigInfoProvider)
assert.True(t, ok)
provider.timeProvider = &fakeTime{now: timeNow}

Expand Down Expand Up @@ -409,7 +409,7 @@ func TestMigInfoProviderGetMigForInstance(t *testing.T) {
fetchMigs: fetchMigsConst(nil),
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false)

mig, err := provider.GetMigForInstance(instanceRef)

Expand Down Expand Up @@ -492,7 +492,7 @@ func TestGetMigInstances(t *testing.T) {
fetchMigInstances: tc.fetchMigInstances,
}
migLister := NewMigLister(tc.cache)
provider, ok := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false).(*cachingMigInfoProvider)
provider, ok := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false).(*cachingMigInfoProvider)
assert.True(t, ok)
provider.timeProvider = &fakeTime{now: newRefreshTime}

Expand Down Expand Up @@ -759,7 +759,7 @@ func TestRegenerateMigInstancesCache(t *testing.T) {
fetchAllInstances: tc.fetchAllInstances,
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, tc.projectId, 1, 0*time.Second, tc.bulkGceMigInstancesListingEnabled)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, tc.projectId, 1, 0*time.Second, tc.bulkGceMigInstancesListingEnabled, false)
err := provider.RegenerateMigInstancesCache()

assert.Equal(t, tc.expectedErr, err)
Expand Down Expand Up @@ -787,11 +787,19 @@ func TestGetMigTargetSize(t *testing.T) {
Zone: mig.GceRef().Zone,
Name: mig.GceRef().Name,
TargetSize: targetSize,
SelfLink: fmt.Sprintf("projects/%s/zones/%s/instanceGroups/%s", mig.GceRef().Project, mig.GceRef().Zone, mig.GceRef().Name),
}
instanceGroupManager1 := &gce.InstanceGroupManager{
Zone: mig1.GceRef().Zone,
Name: mig1.GceRef().Name,
TargetSize: targetSize,
SelfLink: fmt.Sprintf("projects/%s/zones/%s/instanceGroups/%s", mig1.GceRef().Project, mig1.GceRef().Zone, mig1.GceRef().Name),
}

testCases := []struct {
name string
cache *GceCache
migQuery *gceMig
fetchMigs func(string) ([]*gce.InstanceGroupManager, error)
fetchMigTargetSize func(GceRef) (int64, error)
expectedTargetSize int64
Expand All @@ -804,40 +812,53 @@ func TestGetMigTargetSize(t *testing.T) {
migTargetSizeCache: map[GceRef]int64{mig.GceRef(): targetSize},
},
expectedTargetSize: targetSize,
migQuery: mig,
},
{
name: "target size from cache fill",
cache: emptyCache(),
fetchMigs: fetchMigsConst([]*gce.InstanceGroupManager{instanceGroupManager}),
expectedTargetSize: targetSize,
migQuery: mig,
},
{
name: "target size from cache fill, multiple MIG projects",
cache: emptyCache(),
fetchMigs: fetchMigsConst([]*gce.InstanceGroupManager{instanceGroupManager, instanceGroupManager1}),
expectedTargetSize: targetSize,
migQuery: mig1,
},
{
name: "cache fill without mig, fallback success",
cache: emptyCache(),
fetchMigs: fetchMigsConst([]*gce.InstanceGroupManager{}),
fetchMigTargetSize: fetchMigTargetSizeConst(targetSize),
expectedTargetSize: targetSize,
migQuery: mig,
},
{
name: "cache fill failure, fallback success",
cache: emptyCache(),
fetchMigs: fetchMigsFail,
fetchMigTargetSize: fetchMigTargetSizeConst(targetSize),
expectedTargetSize: targetSize,
migQuery: mig,
},
{
name: "cache fill without mig, fallback failure",
cache: emptyCache(),
fetchMigs: fetchMigsConst([]*gce.InstanceGroupManager{}),
fetchMigTargetSize: fetchMigTargetSizeFail,
expectedErr: errFetchMigTargetSize,
migQuery: mig,
},
{
name: "cache fill failure, fallback failure",
cache: emptyCache(),
fetchMigs: fetchMigsFail,
fetchMigTargetSize: fetchMigTargetSizeFail,
expectedErr: errFetchMigTargetSize,
migQuery: mig,
},
}

Expand All @@ -848,10 +869,10 @@ func TestGetMigTargetSize(t *testing.T) {
fetchMigTargetSize: tc.fetchMigTargetSize,
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, true)

targetSize, err := provider.GetMigTargetSize(mig.GceRef())
cachedTargetSize, found := tc.cache.GetMigTargetSize(mig.GceRef())
targetSize, err := provider.GetMigTargetSize(tc.migQuery.GceRef())
cachedTargetSize, found := tc.cache.GetMigTargetSize(tc.migQuery.GceRef())

assert.Equal(t, tc.expectedErr, err)
assert.Equal(t, tc.expectedErr == nil, found)
Expand Down Expand Up @@ -930,7 +951,7 @@ func TestGetMigBasename(t *testing.T) {
fetchMigBasename: tc.fetchMigBasename,
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false)

basename, err := provider.GetMigBasename(mig.GceRef())
cachedBasename, found := tc.cache.GetMigBasename(mig.GceRef())
Expand Down Expand Up @@ -1011,7 +1032,7 @@ func TestGetListManagedInstancesResults(t *testing.T) {
fetchListManagedInstancesResults: tc.fetchResults,
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false)

results, err := provider.GetListManagedInstancesResults(mig.GceRef())
cachedResults, found := tc.cache.GetListManagedInstancesResults(mig.GceRef())
Expand Down Expand Up @@ -1106,7 +1127,7 @@ func TestGetMigInstanceTemplateName(t *testing.T) {
fetchMigTemplateName: tc.fetchMigTemplateName,
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false)

instanceTemplateName, err := provider.GetMigInstanceTemplateName(mig.GceRef())
cachedInstanceTemplateName, found := tc.cache.GetMigInstanceTemplateName(mig.GceRef())
Expand Down Expand Up @@ -1212,7 +1233,7 @@ func TestGetMigInstanceTemplate(t *testing.T) {
fetchMigTemplate: tc.fetchMigTemplate,
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false)

template, err := provider.GetMigInstanceTemplate(mig.GceRef())
cachedTemplate, found := tc.cache.GetMigInstanceTemplate(mig.GceRef())
Expand Down Expand Up @@ -1418,7 +1439,7 @@ func TestGetMigInstanceKubeEnv(t *testing.T) {
fetchMigTemplate: tc.fetchMigTemplate,
}
migLister := NewMigLister(tc.cache)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(tc.cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false)

kubeEnv, err := provider.GetMigKubeEnv(mig.GceRef())
cachedKubeEnv, found := tc.cache.GetMigKubeEnv(mig.GceRef())
Expand Down Expand Up @@ -1513,7 +1534,7 @@ func TestGetMigMachineType(t *testing.T) {
fetchMachineType: tc.fetchMachineType,
}
migLister := NewMigLister(cache)
provider := NewCachingMigInfoProvider(cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false)
provider := NewCachingMigInfoProvider(cache, migLister, client, mig.GceRef().Project, 1, 0*time.Second, false, false)
machine, err := provider.GetMigMachineType(mig.GceRef())
if tc.expectError {
assert.Error(t, err)
Expand Down Expand Up @@ -1916,7 +1937,7 @@ func (f *fakeTime) Now() time.Time {

func emptyCache() *GceCache {
return &GceCache{
migs: map[GceRef]Mig{mig.GceRef(): mig},
migs: map[GceRef]Mig{mig.GceRef(): mig, mig1.GceRef(): mig1},
instances: make(map[GceRef][]GceInstance),
instancesUpdateTime: make(map[GceRef]time.Time),
migTargetSizeCache: make(map[GceRef]int64),
Expand Down
Loading