Skip to content
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

Jump to bottom

Fix legacy fallback stale for aggregated discovery #115770

Merged
merged 1 commit into from Feb 22, 2023
Merged

Fix legacy fallback stale for aggregated discovery #115770

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.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 3 additions & 3 deletions staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery.go
View file
Open in desktop
Expand Up @@ -292,9 +292,9 @@ func (dm *discoveryManager) fetchFreshDiscoveryForService(gv metav1.GroupVersion
lastUpdated: now,
}

// Save the resolve, because it is still useful in case other services
// are already marked dirty. THey can use it without making http request
dm.setCacheEntryForService(info.service, cached)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

prior to this change, the discovery check interval was effectively once every minute for each apiservice. After this change it is an unbounded number of checks per unit time. Can we get back the "once every minute" aspect of the original?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The one check every minute is done here:

kubernetes/staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery.go

Lines 426 to 439 in ed98c1d

wait.PollUntil(1*time.Minute, func() (done bool, err error) {
dm.servicesLock.Lock()
defer dm.servicesLock.Unlock()
now := time.Now()
// Mark all non-local APIServices as dirty
for key, info := range dm.apiServices {
info.lastMarkedDirty = now
dm.apiServices[key] = info
dm.dirtyAPIServiceQueue.Add(key)
}
return false, nil
}, stopCh)
and is untouched in this PR.

The behavior before is that the check is once a minute, but if multiple apiservices share the same service, a fetch will not be performed (and cached result used instead) so the behavior was effectively at most one check per apiservice per minute. Now, the behavior is exactly one check per minute per apiservice for apiservices that need to use the legacy fallback, while for apiservices who supported aggregated discovery, the behavior is unchanged at most one check per apiservice per minute.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now, the behavior is exactly one check per minute per apiservice for apiservices that need to use the legacy fallback, while for apiservices who supported aggregated discovery

If AddAPIService is called more than once for whatever reason (maybe status got touched?), after this PR it is checked more than once per minute, right?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think also prior to this PR, calling AddAPIService more than once per minute would result in the service being added to the queue every time, if it had synced by then, and result in refetching the discovery document.

Before and after this PR, the controller conflates an update to the APIService as a possible change to the underlying service. I think it is worth filing a separate issue to fix.

I think at the very least maybe we should add a check to at least see if the Spec.Service has changed at all before marking the APIService/service as dirty.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think also prior to this PR, calling AddAPIService more than once per minute would result in the service being added to the queue every time

I thought the dirty time was set for one minute later, but I can't find that line now. I'll double check tomorrow, but if I can't find it, I'm ok with this change.

I think at the very least maybe we should add a check to at least see if the Spec.Service has changed at all before marking the APIService/service as dirty.

The check needs to be slightly smarter, since you do need to periodically check the content to see if the etag changed in the modern case and in the legacy case, you probably want to check every few minutes.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like now everywhere. I must have mis-remembered.

// Do not save the resolve as the legacy fallback only fetches
// one group version and an API Service may serve multiple
// group versions.
return &cached, nil

case http.StatusOK:
Expand Down
158 changes: 158 additions & 0 deletions staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_discovery_test.go
View file
Open in desktop
Expand Up @@ -221,6 +221,164 @@ func TestRemoveAPIService(t *testing.T) {
}
}

func TestLegacyFallbackNoCache(t *testing.T) {
aggregatedResourceManager := discoveryendpoint.NewResourceManager()
rootAPIsHandler := discovery.NewRootAPIsHandler(discovery.DefaultAddresses{DefaultAddress: "192.168.1.1"}, scheme.Codecs)

legacyGroupHandler := discovery.NewAPIGroupHandler(scheme.Codecs, metav1.APIGroup{
Name: "stable.example.com",
PreferredVersion: metav1.GroupVersionForDiscovery{
GroupVersion: "stable.example.com/v1",
Version: "v1",
},
Versions: []metav1.GroupVersionForDiscovery{
{
GroupVersion: "stable.example.com/v1",
Version: "v1",
},
{
GroupVersion: "stable.example.com/v1beta1",
Version: "v1beta1",
},
{
GroupVersion: "stable.example.com/v1alpha1",
Version: "v1alpha1",
},
},
})

generateVersionResource := func(version string) metav1.APIResource {
return metav1.APIResource{
Name: "foos",
SingularName: "foo",
Group: "stable.example.com",
Version: version,
Namespaced: false,
Kind: "Foo",
Verbs: []string{"get", "list", "watch", "create", "update", "patch", "delete", "deletecollection"},
Categories: []string{"all"},
}
}

resources := map[string]metav1.APIResource{
"v1": generateVersionResource("v1"),
"v1beta1": generateVersionResource("v1beta1"),
"v1alpha1": generateVersionResource("v1alpha1"),
}

legacyResourceHandlerV1 := discovery.NewAPIVersionHandler(scheme.Codecs, schema.GroupVersion{
Group: "stable.example.com",
Version: "v1",
}, discovery.APIResourceListerFunc(func() []metav1.APIResource {
return []metav1.APIResource{
resources["v1"],
}
}))

legacyResourceHandlerV1Beta1 := discovery.NewAPIVersionHandler(scheme.Codecs, schema.GroupVersion{
Group: "stable.example.com",
Version: "v1beta1",
}, discovery.APIResourceListerFunc(func() []metav1.APIResource {
return []metav1.APIResource{
resources["v1beta1"],
}
}))

legacyResourceHandlerV1Alpha1 := discovery.NewAPIVersionHandler(scheme.Codecs, schema.GroupVersion{
Group: "stable.example.com",
Version: "v1alpha1",
}, discovery.APIResourceListerFunc(func() []metav1.APIResource {
return []metav1.APIResource{
resources["v1alpha1"],
}
}))

handlerFunc := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/apis/stable.example.com" {
legacyGroupHandler.ServeHTTP(w, r)
} else if r.URL.Path == "/apis/stable.example.com/v1" {
// defer to legacy discovery
legacyResourceHandlerV1.ServeHTTP(w, r)
} else if r.URL.Path == "/apis/stable.example.com/v1beta1" {
// defer to legacy discovery
legacyResourceHandlerV1Beta1.ServeHTTP(w, r)
} else if r.URL.Path == "/apis/stable.example.com/v1alpha1" {
legacyResourceHandlerV1Alpha1.ServeHTTP(w, r)
} else if r.URL.Path == "/apis" {
rootAPIsHandler.ServeHTTP(w, r)
} else {
// Unknown url
t.Fatalf("unexpected request sent to %v", r.URL.Path)
}
})

aggregatedManager := newDiscoveryManager(aggregatedResourceManager)
aggregatedManager.AddAPIService(&apiregistrationv1.APIService{
ObjectMeta: metav1.ObjectMeta{
Name: "v1.stable.example.com",
},
Spec: apiregistrationv1.APIServiceSpec{
Group: "stable.example.com",
Version: "v1",
Service: &apiregistrationv1.ServiceReference{
Name: "test-service",
},
},
}, handlerFunc)
aggregatedManager.AddAPIService(&apiregistrationv1.APIService{
ObjectMeta: metav1.ObjectMeta{
Name: "v1beta1.stable.example.com",
},
Spec: apiregistrationv1.APIServiceSpec{
Group: "stable.example.com",
Version: "v1beta1",
Service: &apiregistrationv1.ServiceReference{
Name: "test-service",
},
},
}, handlerFunc)
aggregatedManager.AddAPIService(&apiregistrationv1.APIService{
ObjectMeta: metav1.ObjectMeta{
Name: "v1alpha1.stable.example.com",
},
Spec: apiregistrationv1.APIServiceSpec{
Group: "stable.example.com",
Version: "v1alpha1",
Service: &apiregistrationv1.ServiceReference{
Name: "test-service",
},
},
}, handlerFunc)

testCtx, cancel := context.WithCancel(context.Background())
defer cancel()

go aggregatedManager.Run(testCtx.Done())
require.True(t, waitForEmptyQueue(testCtx.Done(), aggregatedManager))

// At this point external services have synced. Check if discovery document
// includes the legacy resources
_, _, doc := fetchPath(aggregatedResourceManager, "")

aggregatedVersions := []apidiscoveryv2beta1.APIVersionDiscovery{}
for _, resource := range resources {
converted, err := endpoints.ConvertGroupVersionIntoToDiscovery([]metav1.APIResource{resource})
require.NoError(t, err)
aggregatedVersions = append(aggregatedVersions, apidiscoveryv2beta1.APIVersionDiscovery{
Version: resource.Version,
Resources: converted,
Freshness: apidiscoveryv2beta1.DiscoveryFreshnessCurrent,
})
}
aggregatedDiscovery := []apidiscoveryv2beta1.APIGroupDiscovery{{
ObjectMeta: metav1.ObjectMeta{
Name: resources["v1"].Group,
},
Versions: aggregatedVersions,
}}
require.Equal(t, aggregatedDiscovery, doc.Items)
}

func TestLegacyFallback(t *testing.T) {
aggregatedResourceManager := discoveryendpoint.NewResourceManager()
rootAPIsHandler := discovery.NewRootAPIsHandler(discovery.DefaultAddresses{DefaultAddress: "192.168.1.1"}, scheme.Codecs)
Expand Down