Merge pull request kubernetes#122148 from pohly/controllers-context-support

controllers + apiserver: enhance context support

Author: k8s-ci-robot

cmd/kube-apiserver/app/server.go

@@ -20,6 +20,7 @@ limitations under the License.
 package app
 
 import (
+	"context"
 	"crypto/tls"
 	"fmt"
 	"net/http"
@@ -114,7 +115,7 @@ cluster's shared state through which all other components interact.`,
 			}
 			// add feature enablement metrics
 			utilfeature.DefaultMutableFeatureGate.AddMetrics()
-			return Run(completedOptions, genericapiserver.SetupSignalHandler())
+			return Run(cmd.Context(), completedOptions)
 		},
 		Args: func(cmd *cobra.Command, args []string) error {
 			for _, arg := range args {
@@ -125,6 +126,7 @@ cluster's shared state through which all other components interact.`,
 			return nil
 		},
 	}
+	cmd.SetContext(genericapiserver.SetupSignalContext())
 
 	fs := cmd.Flags()
 	namedFlagSets := s.Flags()
@@ -142,7 +144,7 @@ cluster's shared state through which all other components interact.`,
 }
 
 // Run runs the specified APIServer.  This should never exit.
-func Run(opts options.CompletedOptions, stopCh <-chan struct{}) error {
+func Run(ctx context.Context, opts options.CompletedOptions) error {
 	// To help debugging, immediately log version
 	klog.Infof("Version: %+v", version.Get())
 
@@ -166,7 +168,7 @@ func Run(opts options.CompletedOptions, stopCh <-chan struct{}) error {
 		return err
 	}
 
-	return prepared.Run(stopCh)
+	return prepared.Run(ctx)
 }
 
 // CreateServerChain creates the apiservers connected via delegation.

cmd/kube-apiserver/app/testing/testserver.go

@@ -32,6 +32,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"testing"
 	"time"
 
 	"github.com/spf13/pflag"
@@ -56,10 +57,11 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/kube-aggregator/pkg/apiserver"
 	"k8s.io/kubernetes/pkg/features"
+	testutil "k8s.io/kubernetes/test/utils"
+	"k8s.io/kubernetes/test/utils/ktesting"
 
 	"k8s.io/kubernetes/cmd/kube-apiserver/app"
 	"k8s.io/kubernetes/cmd/kube-apiserver/app/options"
-	testutil "k8s.io/kubernetes/test/utils"
 )
 
 func init() {
@@ -139,7 +141,9 @@ func NewDefaultTestServerOptions() *TestServerInstanceOptions {
 // Note: we return a tear-down func instead of a stop channel because the later will leak temporary
 // files that because Golang testing's call to os.Exit will not give a stop channel go routine
 // enough time to remove temporary files.
-func StartTestServer(t Logger, instanceOptions *TestServerInstanceOptions, customFlags []string, storageConfig *storagebackend.Config) (result TestServer, err error) {
+func StartTestServer(t ktesting.TB, instanceOptions *TestServerInstanceOptions, customFlags []string, storageConfig *storagebackend.Config) (result TestServer, err error) {
+	tCtx := ktesting.Init(t)
+
 	if instanceOptions == nil {
 		instanceOptions = NewDefaultTestServerOptions()
 	}
@@ -149,12 +153,11 @@ func StartTestServer(t Logger, instanceOptions *TestServerInstanceOptions, custo
 		return result, fmt.Errorf("failed to create temp dir: %v", err)
 	}
 
-	stopCh := make(chan struct{})
 	var errCh chan error
 	tearDown := func() {
-		// Closing stopCh is stopping apiserver and cleaning up
+		// Cancel is stopping apiserver and cleaning up
 		// after itself, including shutting down its storage layer.
-		close(stopCh)
+		tCtx.Cancel("tearing down")
 
 		// If the apiserver was started, let's wait for it to
 		// shutdown clearly.
@@ -359,15 +362,15 @@ func StartTestServer(t Logger, instanceOptions *TestServerInstanceOptions, custo
 	}
 
 	errCh = make(chan error)
-	go func(stopCh <-chan struct{}) {
+	go func() {
 		defer close(errCh)
 		prepared, err := server.PrepareRun()
 		if err != nil {
 			errCh <- err
-		} else if err := prepared.Run(stopCh); err != nil {
+		} else if err := prepared.Run(tCtx); err != nil {
 			errCh <- err
 		}
-	}(stopCh)
+	}()
 
 	client, err := kubernetes.NewForConfig(server.GenericAPIServer.LoopbackClientConfig)
 	if err != nil {
@@ -465,7 +468,7 @@ func StartTestServer(t Logger, instanceOptions *TestServerInstanceOptions, custo
 }
 
 // StartTestServerOrDie calls StartTestServer t.Fatal if it does not succeed.
-func StartTestServerOrDie(t Logger, instanceOptions *TestServerInstanceOptions, flags []string, storageConfig *storagebackend.Config) *TestServer {
+func StartTestServerOrDie(t testing.TB, instanceOptions *TestServerInstanceOptions, flags []string, storageConfig *storagebackend.Config) *TestServer {
 	result, err := StartTestServer(t, instanceOptions, flags, storageConfig)
 	if err == nil {
 		return &result

pkg/controlplane/controller/defaultservicecidr/default_servicecidr_controller.go

@@ -55,9 +55,6 @@ func NewController(
 	secondaryRange net.IPNet,
 	client clientset.Interface,
 ) *Controller {
-	broadcaster := record.NewBroadcaster()
-	recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: controllerName})
-
 	c := &Controller{
 		client:   client,
 		interval: 10 * time.Second, // same as DefaultEndpointReconcilerInterval
@@ -79,9 +76,6 @@ func NewController(
 	c.serviceCIDRLister = networkingv1alpha1listers.NewServiceCIDRLister(c.serviceCIDRInformer.GetIndexer())
 	c.serviceCIDRsSynced = c.serviceCIDRInformer.HasSynced
 
-	c.eventBroadcaster = broadcaster
-	c.eventRecorder = recorder
-
 	return c
 }
 
@@ -101,9 +95,12 @@ type Controller struct {
 }
 
 // Start will not return until the default ServiceCIDR exists or stopCh is closed.
-func (c *Controller) Start(stopCh <-chan struct{}) {
+func (c *Controller) Start(ctx context.Context) {
 	defer utilruntime.HandleCrash()
+	stopCh := ctx.Done()
 
+	c.eventBroadcaster = record.NewBroadcaster(record.WithContext(ctx))
+	c.eventRecorder = c.eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: controllerName})
 	c.eventBroadcaster.StartStructuredLogging(0)
 	c.eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: c.client.CoreV1().Events("")})
 	defer c.eventBroadcaster.Shutdown()
@@ -116,8 +113,6 @@ func (c *Controller) Start(stopCh <-chan struct{}) {
 		return
 	}
 
-	// derive a context from the stopCh so we can cancel the poll loop
-	ctx := wait.ContextForChannel(stopCh)
 	// wait until first successfully sync
 	// this blocks apiserver startup so poll with a short interval
 	err := wait.PollUntilContextCancel(ctx, 100*time.Millisecond, true, func(ctx context.Context) (bool, error) {

pkg/controlplane/instance.go

@@ -420,7 +420,7 @@ func (c CompletedConfig) New(delegationTarget genericapiserver.DelegationTarget)
 			)
 			// The default serviceCIDR must exist before the apiserver is healthy
 			// otherwise the allocators for Services will not work.
-			controller.Start(hookContext.StopCh)
+			controller.Start(hookContext)
 			return nil
 		})
 	}

staging/src/k8s.io/apiextensions-apiserver/main.go

@@ -25,8 +25,8 @@ import (
 )
 
 func main() {
-	stopCh := genericapiserver.SetupSignalHandler()
-	cmd := server.NewServerCommand(os.Stdout, os.Stderr, stopCh)
+	ctx := genericapiserver.SetupSignalContext()
+	cmd := server.NewServerCommand(ctx, os.Stdout, os.Stderr)
 	code := cli.Run(cmd)
 	os.Exit(code)
 }

staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/server.go

@@ -17,6 +17,7 @@ limitations under the License.
 package server
 
 import (
+	"context"
 	"io"
 
 	"github.com/spf13/cobra"
@@ -25,7 +26,7 @@ import (
 	genericapiserver "k8s.io/apiserver/pkg/server"
 )
 
-func NewServerCommand(out, errOut io.Writer, stopCh <-chan struct{}) *cobra.Command {
+func NewServerCommand(ctx context.Context, out, errOut io.Writer) *cobra.Command {
 	o := options.NewCustomResourceDefinitionsServerOptions(out, errOut)
 
 	cmd := &cobra.Command{
@@ -38,19 +39,20 @@ func NewServerCommand(out, errOut io.Writer, stopCh <-chan struct{}) *cobra.Comm
 			if err := o.Validate(); err != nil {
 				return err
 			}
-			if err := Run(o, stopCh); err != nil {
+			if err := Run(c.Context(), o); err != nil {
 				return err
 			}
 			return nil
 		},
 	}
+	cmd.SetContext(ctx)
 
 	fs := cmd.Flags()
 	o.AddFlags(fs)
 	return cmd
 }
 
-func Run(o *options.CustomResourceDefinitionsServerOptions, stopCh <-chan struct{}) error {
+func Run(ctx context.Context, o *options.CustomResourceDefinitionsServerOptions) error {
 	config, err := o.Config()
 	if err != nil {
 		return err
@@ -60,5 +62,5 @@ func Run(o *options.CustomResourceDefinitionsServerOptions, stopCh <-chan struct
 	if err != nil {
 		return err
 	}
-	return server.GenericAPIServer.PrepareRun().Run(stopCh)
+	return server.GenericAPIServer.PrepareRun().RunWithContext(ctx)
 }

staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/testing/testserver.go

@@ -18,6 +18,7 @@ package testing
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"os"
@@ -83,13 +84,15 @@ func NewDefaultTestServerOptions() *TestServerInstanceOptions {
 // files that because Golang testing's call to os.Exit will not give a stop channel go routine
 // enough time to remove temporary files.
 func StartTestServer(t Logger, _ *TestServerInstanceOptions, customFlags []string, storageConfig *storagebackend.Config) (result TestServer, err error) {
-	stopCh := make(chan struct{})
+	// TODO: this is a candidate for using what is now test/utils/ktesting,
+	// should that become a staging repo.
+	ctx, cancel := context.WithCancelCause(context.Background())
 	var errCh chan error
 	tearDown := func() {
-		// Closing stopCh is stopping apiextensions apiserver and its
+		// Cancel is stopping apiextensions apiserver and its
 		// delegates, which itself is cleaning up after itself,
 		// including shutting down its storage layer.
-		close(stopCh)
+		cancel(errors.New("tearing down"))
 
 		// If the apiextensions apiserver was started, let's wait for
 		// it to shutdown clearly.
@@ -166,13 +169,13 @@ func StartTestServer(t Logger, _ *TestServerInstanceOptions, customFlags []strin
 	}
 
 	errCh = make(chan error)
-	go func(stopCh <-chan struct{}) {
+	go func() {
 		defer close(errCh)
 
-		if err := server.GenericAPIServer.PrepareRun().Run(stopCh); err != nil {
+		if err := server.GenericAPIServer.PrepareRun().RunWithContext(ctx); err != nil {
 			errCh <- err
 		}
-	}(stopCh)
+	}()
 
 	t.Logf("Waiting for /healthz to be ok...")
 

staging/src/k8s.io/apiserver/pkg/server/config_test.go

@@ -17,6 +17,8 @@ limitations under the License.
 package server
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -42,6 +44,7 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/rest"
 	"k8s.io/component-base/tracing"
+	"k8s.io/klog/v2/ktesting"
 	netutils "k8s.io/utils/net"
 )
 
@@ -79,6 +82,9 @@ func TestAuthorizeClientBearerTokenNoops(t *testing.T) {
 }
 
 func TestNewWithDelegate(t *testing.T) {
+	_, ctx := ktesting.NewTestContext(t)
+	ctx, cancel := context.WithCancelCause(ctx)
+	defer cancel(errors.New("test is done"))
 	delegateConfig := NewConfig(codecs)
 	delegateConfig.ExternalAddress = "192.168.10.4:443"
 	delegateConfig.PublicAddress = netutils.ParseIPSloppy("192.168.10.4")
@@ -136,10 +142,8 @@ func TestNewWithDelegate(t *testing.T) {
 		return nil
 	})
 
-	stopCh := make(chan struct{})
-	defer close(stopCh)
 	wrappingServer.PrepareRun()
-	wrappingServer.RunPostStartHooks(stopCh)
+	wrappingServer.RunPostStartHooks(ctx)
 
 	server := httptest.NewServer(wrappingServer.Handler)
 	defer server.Close()