github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-19 09:52:49 +00:00
Code Issues Projects Releases Wiki Activity

namespace controller: use contextual logging

Browse Source
This commit is contained in:
JunYang 2022-10-29 22:34:47 +08:00 committed by 杨军10092085
parent a35650b833
commit f5bd8c86d4
7 changed files with 113 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/kube-controller-manager/app/core.go
View File

@ -456,7 +456,9 @@ func startModifiedNamespaceController(ctx context.Context, controllerContext Con
discoverResourcesFn := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources discoverResourcesFn := namespaceKubeClient.Discovery().ServerPreferredNamespacedResources
ctx = klog.NewContext(ctx, klog.LoggerWithName(klog.FromContext(ctx), "namespace"))
namespaceController := namespacecontroller.NewNamespaceController( namespaceController := namespacecontroller.NewNamespaceController(
ctx,
namespaceKubeClient, namespaceKubeClient,
metadataClient, metadataClient,
discoverResourcesFn, discoverResourcesFn,
@ -464,7 +466,7 @@ func startModifiedNamespaceController(ctx context.Context, controllerContext Con
controllerContext.ComponentConfig.NamespaceController.NamespaceSyncPeriod.Duration, controllerContext.ComponentConfig.NamespaceController.NamespaceSyncPeriod.Duration,
v1.FinalizerKubernetes, v1.FinalizerKubernetes,
) )
go namespaceController.Run(int(controllerContext.ComponentConfig.NamespaceController.ConcurrentNamespaceSyncs), ctx.Done()) go namespaceController.Run(ctx, int(controllerContext.ComponentConfig.NamespaceController.ConcurrentNamespaceSyncs))
return nil, true, nil return nil, true, nil
} }

98
pkg/controller/namespace/deletion/namespaced_resources_deleter.go
View File

@ -39,11 +39,11 @@ import (
// NamespacedResourcesDeleterInterface is the interface to delete a namespace with all resources in it. // NamespacedResourcesDeleterInterface is the interface to delete a namespace with all resources in it.
type NamespacedResourcesDeleterInterface interface { type NamespacedResourcesDeleterInterface interface {
Delete(nsName string) error Delete(ctx context.Context, nsName string) error
} }
// NewNamespacedResourcesDeleter returns a new NamespacedResourcesDeleter. // NewNamespacedResourcesDeleter returns a new NamespacedResourcesDeleter.
func NewNamespacedResourcesDeleter(nsClient v1clientset.NamespaceInterface, func NewNamespacedResourcesDeleter(ctx context.Context, nsClient v1clientset.NamespaceInterface,
metadataClient metadata.Interface, podsGetter v1clientset.PodsGetter, metadataClient metadata.Interface, podsGetter v1clientset.PodsGetter,
discoverResourcesFn func() ([]*metav1.APIResourceList, error), discoverResourcesFn func() ([]*metav1.APIResourceList, error),
finalizerToken v1.FinalizerName) NamespacedResourcesDeleterInterface { finalizerToken v1.FinalizerName) NamespacedResourcesDeleterInterface {
@ -57,7 +57,7 @@ func NewNamespacedResourcesDeleter(nsClient v1clientset.NamespaceInterface,
discoverResourcesFn: discoverResourcesFn, discoverResourcesFn: discoverResourcesFn,
finalizerToken: finalizerToken, finalizerToken: finalizerToken,
} }
d.initOpCache() d.initOpCache(ctx)
return d return d
} }
@ -93,7 +93,7 @@ type namespacedResourcesDeleter struct {
// Returns ResourcesRemainingError if it deleted some resources but needs // Returns ResourcesRemainingError if it deleted some resources but needs
// to wait for them to go away. // to wait for them to go away.
// Caller is expected to keep calling this until it succeeds. // Caller is expected to keep calling this until it succeeds.
func (d *namespacedResourcesDeleter) Delete(nsName string) error { func (d *namespacedResourcesDeleter) Delete(ctx context.Context, nsName string) error {
// Multiple controllers may edit a namespace during termination // Multiple controllers may edit a namespace during termination
// first get the latest state of the namespace before proceeding // first get the latest state of the namespace before proceeding
// if the namespace was deleted already, don't do anything // if the namespace was deleted already, don't do anything
@ -108,7 +108,7 @@ func (d *namespacedResourcesDeleter) Delete(nsName string) error {
return nil return nil
} }
klog.V(5).Infof("namespace controller - syncNamespace - namespace: %s, finalizerToken: %s", namespace.Name, d.finalizerToken) klog.FromContext(ctx).V(5).Info("Namespace controller - syncNamespace", "namespace", namespace.Name, "finalizerToken", d.finalizerToken)
// ensure that the status is up to date on the namespace // ensure that the status is up to date on the namespace
// if we get a not found error, we assume the namespace is truly gone // if we get a not found error, we assume the namespace is truly gone
@ -131,7 +131,7 @@ func (d *namespacedResourcesDeleter) Delete(nsName string) error {
} }
// there may still be content for us to remove // there may still be content for us to remove
estimate, err := d.deleteAllContent(namespace) estimate, err := d.deleteAllContent(ctx, namespace)
if err != nil { if err != nil {
return err return err
} }
@ -153,7 +153,7 @@ func (d *namespacedResourcesDeleter) Delete(nsName string) error {
return nil return nil
} }
func (d *namespacedResourcesDeleter) initOpCache() { func (d *namespacedResourcesDeleter) initOpCache(ctx context.Context) {
// pre-fill opCache with the discovery info // pre-fill opCache with the discovery info
// //
// TODO(sttts): get rid of opCache and http 405 logic around it and trust discovery info // TODO(sttts): get rid of opCache and http 405 logic around it and trust discovery info
@ -161,14 +161,16 @@ func (d *namespacedResourcesDeleter) initOpCache() {
if err != nil { if err != nil {
utilruntime.HandleError(fmt.Errorf("unable to get all supported resources from server: %v", err)) utilruntime.HandleError(fmt.Errorf("unable to get all supported resources from server: %v", err))
} }
logger := klog.FromContext(ctx)
if len(resources) == 0 { if len(resources) == 0 {
klog.Fatalf("Unable to get any supported resources from server: %v", err) logger.Error(err, "Unable to get any supported resources from server")
klog.FlushAndExit(klog.ExitFlushTimeout, 1)
} }
for _, rl := range resources { for _, rl := range resources {
gv, err := schema.ParseGroupVersion(rl.GroupVersion) gv, err := schema.ParseGroupVersion(rl.GroupVersion)
if err != nil { if err != nil {
klog.Errorf("Failed to parse GroupVersion %q, skipping: %v", rl.GroupVersion, err) logger.Error(err, "Failed to parse GroupVersion, skipping", "groupVersion", rl.GroupVersion)
continue continue
} }
@ -177,7 +179,7 @@ func (d *namespacedResourcesDeleter) initOpCache() {
verbs := sets.NewString([]string(r.Verbs)...) verbs := sets.NewString([]string(r.Verbs)...)
if !verbs.Has("delete") { if !verbs.Has("delete") {
klog.V(6).Infof("Skipping resource %v because it cannot be deleted.", gvr) logger.V(6).Info("Skipping resource because it cannot be deleted", "resource", gvr)
} }
for _, op := range []operation{operationList, operationDeleteCollection} { for _, op := range []operation{operationList, operationDeleteCollection} {
@ -205,7 +207,7 @@ const (
operationDeleteCollection operation = "deletecollection" operationDeleteCollection operation = "deletecollection"
operationList operation = "list" operationList operation = "list"
// assume a default estimate for finalizers to complete when found on items pending deletion. // assume a default estimate for finalizers to complete when found on items pending deletion.
finalizerEstimateSeconds int64 = int64(15) finalizerEstimateSeconds = int64(15)
) )
// operationKey is an entry in a cache. // operationKey is an entry in a cache.
@ -304,12 +306,13 @@ func (d *namespacedResourcesDeleter) finalizeNamespace(namespace *v1.Namespace)
// deleteCollection is a helper function that will delete the collection of resources // deleteCollection is a helper function that will delete the collection of resources
// it returns true if the operation was supported on the server. // it returns true if the operation was supported on the server.
// it returns an error if the operation was supported on the server but was unable to complete. // it returns an error if the operation was supported on the server but was unable to complete.
func (d *namespacedResourcesDeleter) deleteCollection(gvr schema.GroupVersionResource, namespace string) (bool, error) { func (d *namespacedResourcesDeleter) deleteCollection(ctx context.Context, gvr schema.GroupVersionResource, namespace string) (bool, error) {
klog.V(5).Infof("namespace controller - deleteCollection - namespace: %s, gvr: %v", namespace, gvr) logger := klog.FromContext(ctx)
logger.V(5).Info("Namespace controller - deleteCollection", "namespace", namespace, "resource", gvr)
key := operationKey{operation: operationDeleteCollection, gvr: gvr} key := operationKey{operation: operationDeleteCollection, gvr: gvr}
if !d.opCache.isSupported(key) { if !d.opCache.isSupported(key) {
klog.V(5).Infof("namespace controller - deleteCollection ignored since not supported - namespace: %s, gvr: %v", namespace, gvr) logger.V(5).Info("Namespace controller - deleteCollection ignored since not supported", "namespace", namespace, "resource", gvr)
return false, nil return false, nil
} }
@ -330,11 +333,11 @@ func (d *namespacedResourcesDeleter) deleteCollection(gvr schema.GroupVersionRes
// /apis/extensions/v1beta1/namespaces/default/replicationcontrollers // /apis/extensions/v1beta1/namespaces/default/replicationcontrollers
// when working with this resource type, we will get a literal not found error rather than expected method not supported // when working with this resource type, we will get a literal not found error rather than expected method not supported
if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) { if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) {
klog.V(5).Infof("namespace controller - deleteCollection not supported - namespace: %s, gvr: %v", namespace, gvr) logger.V(5).Info("Namespace controller - deleteCollection not supported", "namespace", namespace, "resource", gvr)
return false, nil return false, nil
} }
klog.V(5).Infof("namespace controller - deleteCollection unexpected error - namespace: %s, gvr: %v, error: %v", namespace, gvr, err) logger.V(5).Info("Namespace controller - deleteCollection unexpected error", "namespace", namespace, "resource", gvr, "err", err)
return true, err return true, err
} }
@ -344,12 +347,13 @@ func (d *namespacedResourcesDeleter) deleteCollection(gvr schema.GroupVersionRes
// the list of items in the collection (if found) // the list of items in the collection (if found)
// a boolean if the operation is supported // a boolean if the operation is supported
// an error if the operation is supported but could not be completed. // an error if the operation is supported but could not be completed.
func (d *namespacedResourcesDeleter) listCollection(gvr schema.GroupVersionResource, namespace string) (*metav1.PartialObjectMetadataList, bool, error) { func (d *namespacedResourcesDeleter) listCollection(ctx context.Context, gvr schema.GroupVersionResource, namespace string) (*metav1.PartialObjectMetadataList, bool, error) {
klog.V(5).Infof("namespace controller - listCollection - namespace: %s, gvr: %v", namespace, gvr) logger := klog.FromContext(ctx)
logger.V(5).Info("Namespace controller - listCollection", "namespace", namespace, "resource", gvr)
key := operationKey{operation: operationList, gvr: gvr} key := operationKey{operation: operationList, gvr: gvr}
if !d.opCache.isSupported(key) { if !d.opCache.isSupported(key) {
klog.V(5).Infof("namespace controller - listCollection ignored since not supported - namespace: %s, gvr: %v", namespace, gvr) logger.V(5).Info("Namespace controller - listCollection ignored since not supported", "namespace", namespace, "resource", gvr)
return nil, false, nil return nil, false, nil
} }
@ -364,7 +368,7 @@ func (d *namespacedResourcesDeleter) listCollection(gvr schema.GroupVersionResou
// /apis/extensions/v1beta1/namespaces/default/replicationcontrollers // /apis/extensions/v1beta1/namespaces/default/replicationcontrollers
// when working with this resource type, we will get a literal not found error rather than expected method not supported // when working with this resource type, we will get a literal not found error rather than expected method not supported
if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) { if errors.IsMethodNotSupported(err) || errors.IsNotFound(err) {
klog.V(5).Infof("namespace controller - listCollection not supported - namespace: %s, gvr: %v", namespace, gvr) logger.V(5).Info("Namespace controller - listCollection not supported", "namespace", namespace, "resource", gvr)
return nil, false, nil return nil, false, nil
} }
@ -372,10 +376,10 @@ func (d *namespacedResourcesDeleter) listCollection(gvr schema.GroupVersionResou
} }
// deleteEachItem is a helper function that will list the collection of resources and delete each item 1 by 1. // deleteEachItem is a helper function that will list the collection of resources and delete each item 1 by 1.
func (d *namespacedResourcesDeleter) deleteEachItem(gvr schema.GroupVersionResource, namespace string) error { func (d *namespacedResourcesDeleter) deleteEachItem(ctx context.Context, gvr schema.GroupVersionResource, namespace string) error {
klog.V(5).Infof("namespace controller - deleteEachItem - namespace: %s, gvr: %v", namespace, gvr) klog.FromContext(ctx).V(5).Info("Namespace controller - deleteEachItem", "namespace", namespace, "resource", gvr)
unstructuredList, listSupported, err := d.listCollection(gvr, namespace) unstructuredList, listSupported, err := d.listCollection(ctx, gvr, namespace)
if err != nil { if err != nil {
return err return err
} }
@ -406,27 +410,29 @@ type gvrDeletionMetadata struct {
// It returns an estimate of the time remaining before the remaining resources are deleted. // It returns an estimate of the time remaining before the remaining resources are deleted.
// If estimate > 0, not all resources are guaranteed to be gone. // If estimate > 0, not all resources are guaranteed to be gone.
func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource( func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource(
ctx context.Context,
gvr schema.GroupVersionResource, namespace string, gvr schema.GroupVersionResource, namespace string,
namespaceDeletedAt metav1.Time) (gvrDeletionMetadata, error) { namespaceDeletedAt metav1.Time) (gvrDeletionMetadata, error) {
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - namespace: %s, gvr: %v", namespace, gvr) logger := klog.FromContext(ctx)
logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource", "namespace", namespace, "resource", gvr)
// estimate how long it will take for the resource to be deleted (needed for objects that support graceful delete) // estimate how long it will take for the resource to be deleted (needed for objects that support graceful delete)
estimate, err := d.estimateGracefulTermination(gvr, namespace, namespaceDeletedAt) estimate, err := d.estimateGracefulTermination(ctx, gvr, namespace, namespaceDeletedAt)
if err != nil { if err != nil {
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - unable to estimate - namespace: %s, gvr: %v, err: %v", namespace, gvr, err) logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource - unable to estimate", "namespace", namespace, "resource", gvr, "err", err)
return gvrDeletionMetadata{}, err return gvrDeletionMetadata{}, err
} }
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - estimate - namespace: %s, gvr: %v, estimate: %v", namespace, gvr, estimate) logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource - estimate", "namespace", namespace, "resource", gvr, "estimate", estimate)
// first try to delete the entire collection // first try to delete the entire collection
deleteCollectionSupported, err := d.deleteCollection(gvr, namespace) deleteCollectionSupported, err := d.deleteCollection(ctx, gvr, namespace)
if err != nil { if err != nil {
return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, err return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, err
} }
// delete collection was not supported, so we list and delete each item... // delete collection was not supported, so we list and delete each item...
if !deleteCollectionSupported { if !deleteCollectionSupported {
err = d.deleteEachItem(gvr, namespace) err = d.deleteEachItem(ctx, gvr, namespace)
if err != nil { if err != nil {
return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, err return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, err
} }
@ -434,16 +440,16 @@ func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource(
// verify there are no more remaining items // verify there are no more remaining items
// it is not an error condition for there to be remaining items if local estimate is non-zero // it is not an error condition for there to be remaining items if local estimate is non-zero
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - checking for no more items in namespace: %s, gvr: %v", namespace, gvr) logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource - checking for no more items in namespace", "namespace", namespace, "resource", gvr)
unstructuredList, listSupported, err := d.listCollection(gvr, namespace) unstructuredList, listSupported, err := d.listCollection(ctx, gvr, namespace)
if err != nil { if err != nil {
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - error verifying no items in namespace: %s, gvr: %v, err: %v", namespace, gvr, err) logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource - error verifying no items in namespace", "namespace", namespace, "resource", gvr, "err", err)
return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, err return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, err
} }
if !listSupported { if !listSupported {
return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, nil return gvrDeletionMetadata{finalizerEstimateSeconds: estimate}, nil
} }
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - items remaining - namespace: %s, gvr: %v, items: %v", namespace, gvr, len(unstructuredList.Items)) logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource - items remaining", "namespace", namespace, "resource", gvr, "items", len(unstructuredList.Items))
if len(unstructuredList.Items) == 0 { if len(unstructuredList.Items) == 0 {
// we're done // we're done
return gvrDeletionMetadata{finalizerEstimateSeconds: 0, numRemaining: 0}, nil return gvrDeletionMetadata{finalizerEstimateSeconds: 0, numRemaining: 0}, nil
@ -458,7 +464,7 @@ func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource(
} }
if estimate != int64(0) { if estimate != int64(0) {
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - estimate is present - namespace: %s, gvr: %v, finalizers: %v", namespace, gvr, finalizersToNumRemaining) logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource - estimate is present", "namespace", namespace, "resource", gvr, "finalizers", finalizersToNumRemaining)
return gvrDeletionMetadata{ return gvrDeletionMetadata{
finalizerEstimateSeconds: estimate, finalizerEstimateSeconds: estimate,
numRemaining: len(unstructuredList.Items), numRemaining: len(unstructuredList.Items),
@ -468,7 +474,7 @@ func (d *namespacedResourcesDeleter) deleteAllContentForGroupVersionResource(
// if any item has a finalizer, we treat that as a normal condition, and use a default estimation to allow for GC to complete. // if any item has a finalizer, we treat that as a normal condition, and use a default estimation to allow for GC to complete.
if len(finalizersToNumRemaining) > 0 { if len(finalizersToNumRemaining) > 0 {
klog.V(5).Infof("namespace controller - deleteAllContentForGroupVersionResource - items remaining with finalizers - namespace: %s, gvr: %v, finalizers: %v", namespace, gvr, finalizersToNumRemaining) logger.V(5).Info("Namespace controller - deleteAllContentForGroupVersionResource - items remaining with finalizers", "namespace", namespace, "resource", gvr, "finalizers", finalizersToNumRemaining)
return gvrDeletionMetadata{ return gvrDeletionMetadata{
finalizerEstimateSeconds: finalizerEstimateSeconds, finalizerEstimateSeconds: finalizerEstimateSeconds,
numRemaining: len(unstructuredList.Items), numRemaining: len(unstructuredList.Items),
@ -493,13 +499,14 @@ type allGVRDeletionMetadata struct {
// deleteAllContent will use the dynamic client to delete each resource identified in groupVersionResources. // deleteAllContent will use the dynamic client to delete each resource identified in groupVersionResources.
// It returns an estimate of the time remaining before the remaining resources are deleted. // It returns an estimate of the time remaining before the remaining resources are deleted.
// If estimate > 0, not all resources are guaranteed to be gone. // If estimate > 0, not all resources are guaranteed to be gone.
func (d *namespacedResourcesDeleter) deleteAllContent(ns *v1.Namespace) (int64, error) { func (d *namespacedResourcesDeleter) deleteAllContent(ctx context.Context, ns *v1.Namespace) (int64, error) {
namespace := ns.Name namespace := ns.Name
namespaceDeletedAt := *ns.DeletionTimestamp namespaceDeletedAt := *ns.DeletionTimestamp
var errs []error var errs []error
conditionUpdater := namespaceConditionUpdater{} conditionUpdater := namespaceConditionUpdater{}
estimate := int64(0) estimate := int64(0)
klog.V(4).Infof("namespace controller - deleteAllContent - namespace: %s", namespace) logger := klog.FromContext(ctx)
logger.V(4).Info("namespace controller - deleteAllContent", "namespace", namespace)
resources, err := d.discoverResourcesFn() resources, err := d.discoverResourcesFn()
if err != nil { if err != nil {
@ -521,7 +528,7 @@ func (d *namespacedResourcesDeleter) deleteAllContent(ns *v1.Namespace) (int64,
finalizersToNumRemaining: map[string]int{}, finalizersToNumRemaining: map[string]int{},
} }
for gvr := range groupVersionResources { for gvr := range groupVersionResources {
gvrDeletionMetadata, err := d.deleteAllContentForGroupVersionResource(gvr, namespace, namespaceDeletedAt) gvrDeletionMetadata, err := d.deleteAllContentForGroupVersionResource(ctx, gvr, namespace, namespaceDeletedAt)
if err != nil { if err != nil {
// If there is an error, hold on to it but proceed with all the remaining // If there is an error, hold on to it but proceed with all the remaining
// groupVersionResources. // groupVersionResources.
@ -553,19 +560,20 @@ func (d *namespacedResourcesDeleter) deleteAllContent(ns *v1.Namespace) (int64,
} }
// if len(errs)==0, NewAggregate returns nil. // if len(errs)==0, NewAggregate returns nil.
klog.V(4).Infof("namespace controller - deleteAllContent - namespace: %s, estimate: %v, errors: %v", namespace, estimate, utilerrors.NewAggregate(errs)) err = utilerrors.NewAggregate(errs)
return estimate, utilerrors.NewAggregate(errs) logger.V(4).Info("namespace controller - deleteAllContent", "namespace", namespace, "estimate", estimate, "err", err)
return estimate, err
} }
// estimateGracefulTermination will estimate the graceful termination required for the specific entity in the namespace // estimateGracefulTermination will estimate the graceful termination required for the specific entity in the namespace
func (d *namespacedResourcesDeleter) estimateGracefulTermination(gvr schema.GroupVersionResource, ns string, namespaceDeletedAt metav1.Time) (int64, error) { func (d *namespacedResourcesDeleter) estimateGracefulTermination(ctx context.Context, gvr schema.GroupVersionResource, ns string, namespaceDeletedAt metav1.Time) (int64, error) {
groupResource := gvr.GroupResource() groupResource := gvr.GroupResource()
klog.V(5).Infof("namespace controller - estimateGracefulTermination - group %s, resource: %s", groupResource.Group, groupResource.Resource) klog.FromContext(ctx).V(5).Info("Namespace controller - estimateGracefulTermination", "group", groupResource.Group, "resource", groupResource.Resource)
estimate := int64(0) estimate := int64(0)
var err error var err error
switch groupResource { switch groupResource {
case schema.GroupResource{Group: "", Resource: "pods"}: case schema.GroupResource{Group: "", Resource: "pods"}:
estimate, err = d.estimateGracefulTerminationForPods(ns) estimate, err = d.estimateGracefulTerminationForPods(ctx, ns)
} }
if err != nil { if err != nil {
return 0, err return 0, err
@ -580,8 +588,8 @@ func (d *namespacedResourcesDeleter) estimateGracefulTermination(gvr schema.Grou
} }
// estimateGracefulTerminationForPods determines the graceful termination period for pods in the namespace // estimateGracefulTerminationForPods determines the graceful termination period for pods in the namespace
func (d *namespacedResourcesDeleter) estimateGracefulTerminationForPods(ns string) (int64, error) { func (d *namespacedResourcesDeleter) estimateGracefulTerminationForPods(ctx context.Context, ns string) (int64, error) {
klog.V(5).Infof("namespace controller - estimateGracefulTerminationForPods - namespace %s", ns) klog.FromContext(ctx).V(5).Info("Namespace controller - estimateGracefulTerminationForPods", "namespace", ns)
estimate := int64(0) estimate := int64(0)
podsGetter := d.podsGetter podsGetter := d.podsGetter
if podsGetter == nil || reflect.ValueOf(podsGetter).IsNil() { if podsGetter == nil || reflect.ValueOf(podsGetter).IsNil() {

19
pkg/controller/namespace/deletion/namespaced_resources_deleter_test.go
View File

@ -38,6 +38,7 @@ import (
metadatafake "k8s.io/client-go/metadata/fake" metadatafake "k8s.io/client-go/metadata/fake"
restclient "k8s.io/client-go/rest" restclient "k8s.io/client-go/rest"
core "k8s.io/client-go/testing" core "k8s.io/client-go/testing"
"k8s.io/klog/v2/ktesting"
api "k8s.io/kubernetes/pkg/apis/core" api "k8s.io/kubernetes/pkg/apis/core"
) )
@ -200,8 +201,9 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, versions *metav1.APIVersio
fn := func() ([]*metav1.APIResourceList, error) { fn := func() ([]*metav1.APIResourceList, error) {
return resources, testInput.gvrError return resources, testInput.gvrError
} }
d := NewNamespacedResourcesDeleter(mockClient.CoreV1().Namespaces(), metadataClient, mockClient.CoreV1(), fn, v1.FinalizerKubernetes) _, ctx := ktesting.NewTestContext(t)
if err := d.Delete(testInput.testNamespace.Name); !matchErrors(err, testInput.expectErrorOnDelete) { d := NewNamespacedResourcesDeleter(ctx, mockClient.CoreV1().Namespaces(), metadataClient, mockClient.CoreV1(), fn, v1.FinalizerKubernetes)
if err := d.Delete(ctx, testInput.testNamespace.Name); !matchErrors(err, testInput.expectErrorOnDelete) {
t.Errorf("expected error %q when syncing namespace, got %q, %v", testInput.expectErrorOnDelete, err, testInput.expectErrorOnDelete == err) t.Errorf("expected error %q when syncing namespace, got %q, %v", testInput.expectErrorOnDelete, err, testInput.expectErrorOnDelete == err)
} }
@ -297,9 +299,10 @@ func TestSyncNamespaceThatIsActive(t *testing.T) {
fn := func() ([]*metav1.APIResourceList, error) { fn := func() ([]*metav1.APIResourceList, error) {
return testResources(), nil return testResources(), nil
} }
d := NewNamespacedResourcesDeleter(mockClient.CoreV1().Namespaces(), nil, mockClient.CoreV1(), _, ctx := ktesting.NewTestContext(t)
d := NewNamespacedResourcesDeleter(ctx, mockClient.CoreV1().Namespaces(), nil, mockClient.CoreV1(),
fn, v1.FinalizerKubernetes) fn, v1.FinalizerKubernetes)
err := d.Delete(testNamespace.Name) err := d.Delete(ctx, testNamespace.Name)
if err != nil { if err != nil {
t.Errorf("Unexpected error when synching namespace %v", err) t.Errorf("Unexpected error when synching namespace %v", err)
} }
@ -429,12 +432,12 @@ func TestDeleteEncounters404(t *testing.T) {
APIResources: []metav1.APIResource{{Name: "flakes", Namespaced: true, Kind: "Flake", Verbs: []string{"get", "list", "delete", "deletecollection", "create", "update"}}}, APIResources: []metav1.APIResource{{Name: "flakes", Namespaced: true, Kind: "Flake", Verbs: []string{"get", "list", "delete", "deletecollection", "create", "update"}}},
}}, nil }}, nil
} }
_, ctx := ktesting.NewTestContext(t)
d := NewNamespacedResourcesDeleter(mockClient.CoreV1().Namespaces(), mockMetadataClient, mockClient.CoreV1(), resourcesFn, v1.FinalizerKubernetes) d := NewNamespacedResourcesDeleter(ctx, mockClient.CoreV1().Namespaces(), mockMetadataClient, mockClient.CoreV1(), resourcesFn, v1.FinalizerKubernetes)
// Delete ns1 and get NotFound errors for the flakes resource // Delete ns1 and get NotFound errors for the flakes resource
mockMetadataClient.ClearActions() mockMetadataClient.ClearActions()
if err := d.Delete(ns1.Name); err != nil { if err := d.Delete(ctx, ns1.Name); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(mockMetadataClient.Actions()) != 3 || if len(mockMetadataClient.Actions()) != 3 ||
@ -449,7 +452,7 @@ func TestDeleteEncounters404(t *testing.T) {
// Delete ns2 // Delete ns2
mockMetadataClient.ClearActions() mockMetadataClient.ClearActions()
if err := d.Delete(ns2.Name); err != nil { if err := d.Delete(ctx, ns2.Name); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if len(mockMetadataClient.Actions()) != 2 || if len(mockMetadataClient.Actions()) != 2 ||

40
pkg/controller/namespace/namespace_controller.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package namespace package namespace
import ( import (
"context"
"fmt" "fmt"
"time" "time"
@ -63,6 +64,7 @@ type NamespaceController struct {
// NewNamespaceController creates a new NamespaceController // NewNamespaceController creates a new NamespaceController
func NewNamespaceController( func NewNamespaceController(
ctx context.Context,
kubeClient clientset.Interface, kubeClient clientset.Interface,
metadataClient metadata.Interface, metadataClient metadata.Interface,
discoverResourcesFn func() ([]*metav1.APIResourceList, error), discoverResourcesFn func() ([]*metav1.APIResourceList, error),
@ -73,7 +75,7 @@ func NewNamespaceController(
// create the controller so we can inject the enqueue function // create the controller so we can inject the enqueue function
namespaceController := &NamespaceController{ namespaceController := &NamespaceController{
queue: workqueue.NewNamedRateLimitingQueue(nsControllerRateLimiter(), "namespace"), queue: workqueue.NewNamedRateLimitingQueue(nsControllerRateLimiter(), "namespace"),
namespacedResourcesDeleter: deletion.NewNamespacedResourcesDeleter(kubeClient.CoreV1().Namespaces(), metadataClient, kubeClient.CoreV1(), discoverResourcesFn, finalizerToken), namespacedResourcesDeleter: deletion.NewNamespacedResourcesDeleter(ctx, kubeClient.CoreV1().Namespaces(), metadataClient, kubeClient.CoreV1(), discoverResourcesFn, finalizerToken),
} }
// configure the namespace informer event handlers // configure the namespace informer event handlers
@ -132,15 +134,15 @@ func (nm *NamespaceController) enqueueNamespace(obj interface{}) {
// Each namespace can be in the queue at most once. // Each namespace can be in the queue at most once.
// The system ensures that no two workers can process // The system ensures that no two workers can process
// the same namespace at the same time. // the same namespace at the same time.
func (nm *NamespaceController) worker() { func (nm *NamespaceController) worker(ctx context.Context) {
workFunc := func() bool { workFunc := func(ctx context.Context) bool {
key, quit := nm.queue.Get() key, quit := nm.queue.Get()
if quit { if quit {
return true return true
} }
defer nm.queue.Done(key) defer nm.queue.Done(key)
err := nm.syncNamespaceFromKey(key.(string)) err := nm.syncNamespaceFromKey(ctx, key.(string))
if err == nil { if err == nil {
// no error, forget this entry and return // no error, forget this entry and return
nm.queue.Forget(key) nm.queue.Forget(key)
@ -149,7 +151,7 @@ func (nm *NamespaceController) worker() {
if estimate, ok := err.(*deletion.ResourcesRemainingError); ok { if estimate, ok := err.(*deletion.ResourcesRemainingError); ok {
t := estimate.Estimate/2 + 1 t := estimate.Estimate/2 + 1
klog.V(4).Infof("Content remaining in namespace %s, waiting %d seconds", key, t) klog.FromContext(ctx).V(4).Info("Content remaining in namespace", "namespace", key, "waitSeconds", t)
nm.queue.AddAfter(key, time.Duration(t)*time.Second) nm.queue.AddAfter(key, time.Duration(t)*time.Second)
} else { } else {
// rather than wait for a full resync, re-add the namespace to the queue to be processed // rather than wait for a full resync, re-add the namespace to the queue to be processed
@ -158,9 +160,8 @@ func (nm *NamespaceController) worker() {
} }
return false return false
} }
for { for {
quit := workFunc() quit := workFunc(ctx)
if quit { if quit {
return return
@ -169,39 +170,40 @@ func (nm *NamespaceController) worker() {
} }
// syncNamespaceFromKey looks for a namespace with the specified key in its store and synchronizes it // syncNamespaceFromKey looks for a namespace with the specified key in its store and synchronizes it
func (nm *NamespaceController) syncNamespaceFromKey(key string) (err error) { func (nm *NamespaceController) syncNamespaceFromKey(ctx context.Context, key string) (err error) {
startTime := time.Now() startTime := time.Now()
logger := klog.FromContext(ctx)
defer func() { defer func() {
klog.V(4).Infof("Finished syncing namespace %q (%v)", key, time.Since(startTime)) logger.V(4).Info("Finished syncing namespace", "namespace", key, "duration", time.Since(startTime))
}() }()
namespace, err := nm.lister.Get(key) namespace, err := nm.lister.Get(key)
if errors.IsNotFound(err) { if errors.IsNotFound(err) {
klog.Infof("Namespace has been deleted %v", key) logger.Info("Namespace has been deleted", "namespace", key)
return nil return nil
} }
if err != nil { if err != nil {
utilruntime.HandleError(fmt.Errorf("Unable to retrieve namespace %v from store: %v", key, err)) utilruntime.HandleError(fmt.Errorf("Unable to retrieve namespace %v from store: %v", key, err))
return err return err
} }
return nm.namespacedResourcesDeleter.Delete(namespace.Name) return nm.namespacedResourcesDeleter.Delete(ctx, namespace.Name)
} }
// Run starts observing the system with the specified number of workers. // Run starts observing the system with the specified number of workers.
func (nm *NamespaceController) Run(workers int, stopCh <-chan struct{}) { func (nm *NamespaceController) Run(ctx context.Context, workers int) {
defer utilruntime.HandleCrash() defer utilruntime.HandleCrash()
defer nm.queue.ShutDown() defer nm.queue.ShutDown()
logger := klog.FromContext(ctx)
logger.Info("Starting namespace controller")
defer logger.Info("Shutting down namespace controller")
klog.Infof("Starting namespace controller") if !cache.WaitForNamedCacheSync("namespace", ctx.Done(), nm.listerSynced) {
defer klog.Infof("Shutting down namespace controller")
if !cache.WaitForNamedCacheSync("namespace", stopCh, nm.listerSynced) {
return return
} }
klog.V(5).Info("Starting workers of namespace controller") logger.V(5).Info("Starting workers of namespace controller")
for i := 0; i < workers; i++ { for i := 0; i < workers; i++ {
go wait.Until(nm.worker, time.Second, stopCh) go wait.UntilWithContext(ctx, nm.worker, time.Second)
} }
<-stopCh <-ctx.Done()
} }

11
test/e2e_node/services/internal_services.go
View File

@ -17,11 +17,13 @@ limitations under the License.
package services package services
import ( import (
"context"
"os" "os"
"testing" "testing"
etcd3testing "k8s.io/apiserver/pkg/storage/etcd3/testing" etcd3testing "k8s.io/apiserver/pkg/storage/etcd3/testing"
"k8s.io/apiserver/pkg/storage/storagebackend" "k8s.io/apiserver/pkg/storage/storagebackend"
"k8s.io/klog/v2/ktesting"
"k8s.io/kubernetes/test/e2e/framework" "k8s.io/kubernetes/test/e2e/framework"
"k8s.io/klog/v2" "k8s.io/klog/v2"
@ -55,6 +57,7 @@ func (es *e2eServices) run(t *testing.T) error {
// start starts the tests embedded services or returns an error. // start starts the tests embedded services or returns an error.
func (es *e2eServices) start(t *testing.T) error { func (es *e2eServices) start(t *testing.T) error {
_, ctx := ktesting.NewTestContext(t)
klog.Info("Starting e2e services...") klog.Info("Starting e2e services...")
err := es.startEtcd(t) err := es.startEtcd(t)
if err != nil { if err != nil {
@ -64,7 +67,7 @@ func (es *e2eServices) start(t *testing.T) error {
if err != nil { if err != nil {
return err return err
} }
err = es.startNamespaceController() err = es.startNamespaceController(ctx)
if err != nil { if err != nil {
return nil return nil
} }
@ -124,10 +127,10 @@ func (es *e2eServices) startAPIServer(etcdStorage *storagebackend.Config) error
} }
// startNamespaceController starts the embedded namespace controller or returns an error. // startNamespaceController starts the embedded namespace controller or returns an error.
func (es *e2eServices) startNamespaceController() error { func (es *e2eServices) startNamespaceController(ctx context.Context) error {
klog.Info("Starting namespace controller") klog.FromContext(ctx).Info("Starting namespace controller")
es.nsController = NewNamespaceController(framework.TestContext.Host) es.nsController = NewNamespaceController(framework.TestContext.Host)
return es.nsController.Start() return es.nsController.Start(ctx)
} }
// getServicesHealthCheckURLs returns the health check urls for the internal services. // getServicesHealthCheckURLs returns the health check urls for the internal services.

7
test/e2e_node/services/namespace_controller.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package services package services
import ( import (
"context"
"time" "time"
v1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1"
@ -49,7 +50,7 @@ func NewNamespaceController(host string) *NamespaceController {
} }
// Start starts the namespace controller. // Start starts the namespace controller.
func (n *NamespaceController) Start() error { func (n *NamespaceController) Start(ctx context.Context) error {
config := restclient.AddUserAgent(&restclient.Config{ config := restclient.AddUserAgent(&restclient.Config{
Host: n.host, Host: n.host,
BearerToken: framework.TestContext.BearerToken, BearerToken: framework.TestContext.BearerToken,
@ -72,7 +73,9 @@ func (n *NamespaceController) Start() error {
} }
discoverResourcesFn := client.Discovery().ServerPreferredNamespacedResources discoverResourcesFn := client.Discovery().ServerPreferredNamespacedResources
informerFactory := informers.NewSharedInformerFactory(client, ncResyncPeriod) informerFactory := informers.NewSharedInformerFactory(client, ncResyncPeriod)
nc := namespacecontroller.NewNamespaceController( nc := namespacecontroller.NewNamespaceController(
ctx,
client, client,
metadataClient, metadataClient,
discoverResourcesFn, discoverResourcesFn,
@ -80,7 +83,7 @@ func (n *NamespaceController) Start() error {
ncResyncPeriod, v1.FinalizerKubernetes, ncResyncPeriod, v1.FinalizerKubernetes,
) )
informerFactory.Start(n.stopCh) informerFactory.Start(n.stopCh)
go nc.Run(ncConcurrency, n.stopCh) go nc.Run(ctx, ncConcurrency)
return nil return nil
} }

21
test/integration/namespace/ns_conditions_test.go
View File

@ -35,6 +35,7 @@ import (
clientset "k8s.io/client-go/kubernetes" clientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/metadata" "k8s.io/client-go/metadata"
restclient "k8s.io/client-go/rest" restclient "k8s.io/client-go/rest"
"k8s.io/klog/v2/ktesting"
kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing" kubeapiservertesting "k8s.io/kubernetes/cmd/kube-apiserver/app/testing"
"k8s.io/kubernetes/pkg/controller/namespace" "k8s.io/kubernetes/pkg/controller/namespace"
"k8s.io/kubernetes/test/integration/etcd" "k8s.io/kubernetes/test/integration/etcd"
@ -55,10 +56,12 @@ func TestNamespaceCondition(t *testing.T) {
} }
// Start informer and controllers // Start informer and controllers
stopCh := make(chan struct{}) _, ctx := ktesting.NewTestContext(t)
defer close(stopCh) ctx, cancel := context.WithCancel(ctx)
informers.Start(stopCh) defer cancel()
go nsController.Run(5, stopCh)
informers.Start(ctx.Done())
go nsController.Run(ctx, 5)
data := etcd.GetEtcdStorageDataForNamespace(nsName) data := etcd.GetEtcdStorageDataForNamespace(nsName)
podJSON, err := jsonToUnstructured(data[corev1.SchemeGroupVersion.WithResource("pods")].Stub, "v1", "Pod") podJSON, err := jsonToUnstructured(data[corev1.SchemeGroupVersion.WithResource("pods")].Stub, "v1", "Pod")
@ -124,9 +127,10 @@ func TestNamespaceLabels(t *testing.T) {
// Even though nscontroller isn't used in this test, its creation is already // Even though nscontroller isn't used in this test, its creation is already
// spawning some goroutines. So we need to run it to ensure they won't leak. // spawning some goroutines. So we need to run it to ensure they won't leak.
stopCh := make(chan struct{}) _, ctx := ktesting.NewTestContext(t)
close(stopCh) ctx, cancel := context.WithCancel(ctx)
go nsController.Run(5, stopCh) defer cancel()
go nsController.Run(ctx, 5)
nsName := "test-namespace-labels-generated" nsName := "test-namespace-labels-generated"
// Create a new namespace w/ no name // Create a new namespace w/ no name
@ -192,8 +196,9 @@ func namespaceLifecycleSetup(t *testing.T) (kubeapiservertesting.TearDownFunc, *
} }
discoverResourcesFn := clientSet.Discovery().ServerPreferredNamespacedResources discoverResourcesFn := clientSet.Discovery().ServerPreferredNamespacedResources
_, ctx := ktesting.NewTestContext(t)
controller := namespace.NewNamespaceController( controller := namespace.NewNamespaceController(
ctx,
clientSet, clientSet,
metadataClient, metadataClient,
discoverResourcesFn, discoverResourcesFn,