Move servicelb into cloudprovider LoadBalancer interface

manifests/ccm.yaml

@@ -24,21 +24,22 @@ rules:
   resources:
   - nodes
   verbs:
-  - '*'
+  - "*"
 - apiGroups:
   - ""
   resources:
   - nodes/status
+  - services/status
   verbs:
   - patch
 - apiGroups:
   - ""
   resources:
   - services
+  - pods
   verbs:
+  - get
   - list
-  - patch
-  - update
   - watch
 - apiGroups:
   - ""
@@ -49,22 +50,16 @@ rules:
 - apiGroups:
   - ""
   resources:
-  - persistentvolumes
+  - namespaces
   verbs:
+  - create
   - get
-  - list
-  - update
-  - watch
 - apiGroups:
-  - ""
+  - apps
   resources:
-  - endpoints
+  - daemonsets
   verbs:
-  - create
-  - get
-  - list
-  - watch
-  - update
+  - "*"
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding

pkg/cloudprovider/cloudprovider.go

 package cloudprovider
 
 import (
+	"encoding/json"
 	"io"
+	"io/ioutil"
 
+	"github.com/k3s-io/k3s/pkg/util"
 	"github.com/k3s-io/k3s/pkg/version"
+	"github.com/rancher/wrangler/pkg/apply"
+	"github.com/rancher/wrangler/pkg/generated/controllers/apps"
+	appsclient "github.com/rancher/wrangler/pkg/generated/controllers/apps/v1"
+	"github.com/rancher/wrangler/pkg/generated/controllers/core"
+	coreclient "github.com/rancher/wrangler/pkg/generated/controllers/core/v1"
+	"github.com/rancher/wrangler/pkg/generic"
+	"github.com/rancher/wrangler/pkg/start"
+	"github.com/sirupsen/logrus"
+	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
 	cloudprovider "k8s.io/cloud-provider"
 )
 
+// Config describes externally-configurable cloud provider configuration.
+// This is normally unmarshalled from a JSON config file.
+type Config struct {
+	LBEnabled   bool   `json:"lbEnabled"`
+	LBImage     string `json:"lbImage"`
+	LBNamespace string `json:"lbNamespace"`
+	Rootless    bool   `json:"rootless"`
+}
+
 type k3s struct {
+	Config
+
+	client   kubernetes.Interface
+	recorder record.EventRecorder
+
+	processor      apply.Apply
+	daemonsetCache appsclient.DaemonSetCache
+	nodeCache      coreclient.NodeCache
+	podCache       coreclient.PodCache
+	workqueue      workqueue.RateLimitingInterface
 }
 
 var _ cloudprovider.Interface = &k3s{}
 
 func init() {
 	cloudprovider.RegisterCloudProvider(version.Program, func(config io.Reader) (cloudprovider.Interface, error) {
-		return &k3s{}, nil
+		var err error
+		k := k3s{
+			Config: Config{
+				LBEnabled:   true,
+				LBImage:     DefaultLBImage,
+				LBNamespace: DefaultLBNS,
+			},
+		}
+
+		if config != nil {
+			var bytes []byte
+			bytes, err = ioutil.ReadAll(config)
+			if err == nil {
+				err = json.Unmarshal(bytes, &k.Config)
+			}
+		}
+
+		return &k, err
 	})
 }
 
 func (k *k3s) Initialize(clientBuilder cloudprovider.ControllerClientBuilder, stop <-chan struct{}) {
+	ctx, _ := util.ContextForChannel(stop)
+	config := clientBuilder.ConfigOrDie(controllerName)
+	k.client = kubernetes.NewForConfigOrDie(config)
+
+	if k.LBEnabled {
+		// Wrangler controller and caches are only needed if the load balancer controller is enabled.
+		k.recorder = util.BuildControllerEventRecorder(k.client, controllerName, meta.NamespaceAll)
+		coreFactory := core.NewFactoryFromConfigOrDie(config)
+		k.nodeCache = coreFactory.Core().V1().Node().Cache()
+
+		lbCoreFactory := core.NewFactoryFromConfigWithOptionsOrDie(config, &generic.FactoryOptions{Namespace: k.LBNamespace})
+		lbAppsFactory := apps.NewFactoryFromConfigWithOptionsOrDie(config, &generic.FactoryOptions{Namespace: k.LBNamespace})
+
+		processor, err := apply.NewForConfig(config)
+		if err != nil {
+			logrus.Fatalf("Failed to create apply processor for %s: %v", controllerName, err)
+		}
+		k.processor = processor.WithDynamicLookup().WithCacheTypes(lbAppsFactory.Apps().V1().DaemonSet())
+		k.daemonsetCache = lbAppsFactory.Apps().V1().DaemonSet().Cache()
+		k.podCache = lbCoreFactory.Core().V1().Pod().Cache()
+		k.workqueue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+
+		if err := k.Register(ctx, coreFactory.Core().V1().Node(), lbCoreFactory.Core().V1().Pod()); err != nil {
+			logrus.Fatalf("Failed to register %s handlers: %v", controllerName, err)
+		}
+
+		if err := start.All(ctx, 1, coreFactory, lbCoreFactory, lbAppsFactory); err != nil {
+			logrus.Fatalf("Failed to start %s controllers: %v", controllerName, err)
+		}
+	} else {
+		// If load-balancer functionality has not been enabled, delete managed daemonsets.
+		// This uses the raw kubernetes client, as the controllers are not started when the load balancer controller is disabled.
+		if err := k.deleteAllDaemonsets(ctx); err != nil {
+			logrus.Fatalf("Failed to clean up %s daemonsets: %v", controllerName, err)
+		}
+	}
 }
 
 func (k *k3s) Instances() (cloudprovider.Instances, bool) {
@@ -30,7 +117,7 @@ func (k *k3s) InstancesV2() (cloudprovider.InstancesV2, bool) {
 }
 
 func (k *k3s) LoadBalancer() (cloudprovider.LoadBalancer, bool) {
-	return nil, false
+	return k, k.LBEnabled
 }
 
 func (k *k3s) Zones() (cloudprovider.Zones, bool) {

pkg/cloudprovider/loadbalancer.go

+package cloudprovider
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	cloudprovider "k8s.io/cloud-provider"
+)
+
+var _ cloudprovider.LoadBalancer = &k3s{}
+
+// GetLoadBalancer returns whether the specified load balancer exists, and if so, what its status is.
+func (k *k3s) GetLoadBalancer(ctx context.Context, clusterName string, service *corev1.Service) (*corev1.LoadBalancerStatus, bool, error) {
+	if _, err := k.getDaemonSet(service); err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil, false, nil
+		}
+		return nil, false, err
+	}
+
+	status, err := k.getStatus(service)
+	return status, true, err
+}
+
+// GetLoadBalancerName returns the name of the load balancer.
+func (k *k3s) GetLoadBalancerName(ctx context.Context, clusterName string, service *corev1.Service) string {
+	return generateName(service)
+}
+
+// EnsureLoadBalancer creates a new load balancer 'name', or updates the existing one. Returns the status of the balancer.
+// The node list is unused; see the comment on UpdateLoadBalancer for information on why.
+// This is called when the Service is created or changes.
+func (k *k3s) EnsureLoadBalancer(ctx context.Context, clusterName string, service *corev1.Service, nodes []*corev1.Node) (*corev1.LoadBalancerStatus, error) {
+	if err := k.deployDaemonSet(ctx, service); err != nil {
+		return nil, err
+	}
+	return nil, cloudprovider.ImplementedElsewhere
+}
+
+// UpdateLoadBalancer updates hosts under the specified load balancer.
+// This is not used, as it filters node updates based on criteria not compatible with how our DaemonSet selects
+// nodes for inclusion.  It also does not provide any opportunity to update the load balancer status.
+// https://github.com/kubernetes/kubernetes/blob/v1.25.0/staging/src/k8s.io/cloud-provider/controllers/service/controller.go#L985-L993
+func (k *k3s) UpdateLoadBalancer(ctx context.Context, clusterName string, service *corev1.Service, nodes []*corev1.Node) error {
+	return cloudprovider.ImplementedElsewhere
+}
+
+// EnsureLoadBalancerDeleted deletes the specified load balancer if it exists,
+// returning nil if the load balancer specified either didn't exist or was successfully deleted.
+func (k *k3s) EnsureLoadBalancerDeleted(ctx context.Context, clusterName string, service *corev1.Service) error {
+	return k.deleteDaemonSet(ctx, service)
+}

pkg/daemons/agent/agent_linux.go

@@ -173,8 +173,8 @@ func kubeletArgs(cfg *config.Agent) map[string]string {
 		argsMap["protect-kernel-defaults"] = "true"
 	}
 
-	if !cfg.DisableServiceLB && cfg.EnableIPv6 {
-		argsMap["allowed-unsafe-sysctls"] = "net.ipv6.conf.all.forwarding"
+	if !cfg.DisableServiceLB {
+		argsMap["allowed-unsafe-sysctls"] = "net.ipv4.ip_forward,net.ipv6.conf.all.forwarding"
 	}
 
 	return argsMap

pkg/daemons/config/types.go

@@ -308,7 +308,8 @@ type ControlRuntime struct {
 	Tunnel             http.Handler
 	Authenticator      authenticator.Request
 
-	EgressSelectorConfig string
+	EgressSelectorConfig  string
+	CloudControllerConfig string
 
 	ClientAuthProxyCert string
 	ClientAuthProxyKey  string

pkg/daemons/control/deps/deps.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/k3s-io/k3s/pkg/clientaccess"
+	"github.com/k3s-io/k3s/pkg/cloudprovider"
 	"github.com/k3s-io/k3s/pkg/daemons/config"
 	"github.com/k3s-io/k3s/pkg/passwd"
 	"github.com/k3s-io/k3s/pkg/token"
@@ -138,6 +139,7 @@ func CreateRuntimeCertFiles(config *config.Control) {
 	runtime.ServingKubeletKey = filepath.Join(config.DataDir, "tls", "serving-kubelet.key")
 
 	runtime.EgressSelectorConfig = filepath.Join(config.DataDir, "etc", "egress-selector-config.yaml")
+	runtime.CloudControllerConfig = filepath.Join(config.DataDir, "etc", "cloud-config.yaml")
 
 	runtime.ClientAuthProxyCert = filepath.Join(config.DataDir, "tls", "client-auth-proxy.crt")
 	runtime.ClientAuthProxyKey = filepath.Join(config.DataDir, "tls", "client-auth-proxy.key")
@@ -187,6 +189,10 @@ func GenServerDeps(config *config.Control) error {
 		return err
 	}
 
+	if err := genCloudConfig(config); err != nil {
+		return err
+	}
+
 	return readTokens(runtime)
 }
 
@@ -764,3 +770,21 @@ func genEgressSelectorConfig(controlConfig *config.Control) error {
 	}
 	return ioutil.WriteFile(controlConfig.Runtime.EgressSelectorConfig, b, 0600)
 }
+
+func genCloudConfig(controlConfig *config.Control) error {
+	cloudConfig := cloudprovider.Config{
+		LBEnabled:   !controlConfig.DisableServiceLB,
+		LBNamespace: controlConfig.ServiceLBNamespace,
+		LBImage:     cloudprovider.DefaultLBImage,
+		Rootless:    controlConfig.Rootless,
+	}
+	if controlConfig.SystemDefaultRegistry != "" {
+		cloudConfig.LBImage = controlConfig.SystemDefaultRegistry + "/" + cloudConfig.LBImage
+	}
+	b, err := json.Marshal(cloudConfig)
+	if err != nil {
+		return err
+	}
+	return ioutil.WriteFile(controlConfig.Runtime.CloudControllerConfig, b, 0600)
+
+}

pkg/daemons/control/server.go

@@ -302,6 +302,7 @@ func cloudControllerManager(ctx context.Context, cfg *config.Control) error {
 		"profiling":                    "false",
 		"allocate-node-cidrs":          "true",
 		"cloud-provider":               version.Program,
+		"cloud-config":                 runtime.CloudControllerConfig,
 		"cluster-cidr":                 util.JoinIPNets(cfg.ClusterIPRanges),
 		"configure-cloud-routes":       "false",
 		"kubeconfig":                   runtime.KubeConfigCloudController,
@@ -371,9 +372,9 @@ func checkForCloudControllerPrivileges(ctx context.Context, runtime *config.Cont
 			User: version.Program + "-cloud-controller-manager",
 			ResourceAttributes: &authorizationv1.ResourceAttributes{
 				Namespace: metav1.NamespaceSystem,
-				Verb:      "get",
-				Resource:  "configmaps",
-				Name:      "extension-apiserver-authentication",
+				Verb:      "*",
+				Resource:  "daemonsets",
+				Group:     "apps",
 			},
 		},
 	}

pkg/daemons/executor/embed.go

@@ -18,7 +18,6 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/apiserver/pkg/authentication/authenticator"
 	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -164,27 +163,17 @@ func (*Embedded) CloudControllerManager(ctx context.Context, ccmRBACReady <-chan
 	}
 
 	cloudInitializer := func(config *cloudcontrollerconfig.CompletedConfig) cloudprovider.Interface {
-		cloud, err := ccm.InitCloudProvider(version.Program, "")
+		cloud, err := ccm.InitCloudProvider(version.Program, config.ComponentConfig.KubeCloudShared.CloudProvider.CloudConfigFile)
 		if err != nil {
 			logrus.Fatalf("Cloud provider could not be initialized: %v", err)
 		}
 		if cloud == nil {
 			logrus.Fatalf("Cloud provider is nil")
 		}
-
-		cloud.Initialize(config.ClientBuilder, make(chan struct{}))
-		if informerUserCloud, ok := cloud.(ccm.InformerUser); ok {
-			informerUserCloud.SetInformers(config.SharedInformers)
-		}
-
 		return cloud
 	}
 
-	controllerInitializers := ccmapp.DefaultInitFuncConstructors
-	delete(controllerInitializers, "service")
-	delete(controllerInitializers, "route")
-
-	command := ccmapp.NewCloudControllerManagerCommand(ccmOptions, cloudInitializer, controllerInitializers, cliflag.NamedFlagSets{}, wait.NeverStop)
+	command := ccmapp.NewCloudControllerManagerCommand(ccmOptions, cloudInitializer, ccmapp.DefaultInitFuncConstructors, cliflag.NamedFlagSets{}, ctx.Done())
 	command.SetArgs(args)
 
 	go func() {

pkg/deploy/zz_generated_bindata.go

@@ -91,7 +91,7 @@ func (fi bindataFileInfo) Sys() interface{} {
 	return nil
 }
 
-var _ccmYaml = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xcc\x94\x41\x8f\x13\x31\x0c\x85\xef\xf9\x15\xd1\x5e\x56\x42\x4a\x57\x88\x0b\x9a\x23\x1c\xb8\xaf\x04\x77\x37\x79\x74\x43\x33\x71\x14\x3b\xb3\xc0\xaf\x47\xe9\x2c\x68\x99\xa1\x55\x5b\x40\x70\x8b\x2a\xfb\x7b\xcf\xcf\xf5\x50\x89\x1f\x50\x25\x72\x1e\x6c\xdd\x92\xdf\x50\xd3\x07\xae\xf1\x2b\x69\xe4\xbc\xd9\xbf\x96\x4d\xe4\xbb\xe9\xa5\xd9\xc7\x1c\x06\xfb\x36\x35\x51\xd4\x7b\x4e\x30\x23\x94\x02\x29\x0d\xc6\xda\x4c\x23\x06\xbb\x7f\x25\xce\x27\x6e\xc1\x79\xce\x5a\x39\x25\x54\x37\x52\xa6\x1d\xaa\xa9\x2d\x41\x06\xe3\x2c\x95\xf8\xae\x72\x2b\xd2\x1b\x9d\xf5\xcc\x35\xc4\xfc\x5c\xcf\x58\x5b\x21\xdc\xaa\xc7\x53\x51\x02\x09\xc4\x58\x3b\xa1\x6e\x9f\x7e\xdb\x41\x67\x40\x05\x29\x0e\xcf\x56\x42\x7f\xae\x34\x6e\x6e\xd6\x48\x4c\xc8\xba\x40\x3e\x43\x15\x52\xff\x70\x31\x34\x73\x58\xda\xbc\x7d\x71\x7b\x41\xef\x9d\x28\x69\x5b\x20\x66\x2f\x67\x41\x04\x75\x8a\x7e\xe9\x21\x45\xd1\x5f\x4f\xd5\x9f\x8f\x17\xe3\xc9\x7b\x6e\xc7\xd2\x3b\x0b\x54\xfa\x9f\x4e\x14\x59\x27\x4e\x6d\x3c\xb6\xdb\x1f\xc6\xaf\xb3\x8b\x1c\x0a\xc7\x53\x6b\x5e\x09\x3d\xae\xf6\xee\x9c\xb9\xfe\x4a\xde\xc4\x1c\x62\xde\x5d\x7c\x2c\x9c\x70\x8f\x8f\xbd\xfa\xfb\x98\x27\x94\x8d\xb5\xeb\xf3\x3c\x4b\x47\xda\xf6\x13\xbc\x1e\xee\x72\x46\xbc\x17\xd4\xf3\x7a\xe7\x22\x29\xe4\x7b\x65\xdb\xc2\xc9\x17\x51\x8c\xff\x24\x31\xd7\xf9\x2e\x20\x61\x47\xca\x7f\x34\xc0\x79\xaa\x61\x21\xf0\xbf\x24\xf7\x9b\x91\x21\x6b\xf4\x07\xb2\xab\xa0\x70\xca\xdc\x95\x91\xfe\x94\x25\x3e\x2b\x72\x9f\xcd\x51\x89\xfd\x63\x72\xd4\xc6\x5f\xc9\xf7\x5b\x00\x00\x00\xff\xff\xc2\xa7\x17\xb8\xee\x06\x00\x00")
+var _ccmYaml = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xcc\x94\x4f\x8f\x13\x31\x0c\xc5\xef\xf9\x14\x51\x8f\x48\xe9\x0a\x71\x41\x73\x84\x03\xf7\x95\xe0\xee\x26\x8f\x6e\x68\x26\x8e\x62\xa7\xfc\xf9\xf4\x68\x66\x76\xc5\x30\xa3\xae\x3a\x05\xc4\xde\x2c\x2b\xfe\xf9\xf9\x39\x09\x95\xf8\x09\x55\x22\xe7\xce\xd6\x03\xf9\x3d\x35\x7d\xe0\x1a\x7f\x90\x46\xce\xfb\xd3\x5b\xd9\x47\xbe\x3b\xbf\x36\xa7\x98\x43\x67\xdf\xa7\x26\x8a\x7a\xcf\x09\xa6\x87\x52\x20\xa5\xce\x58\x9b\xa9\x47\x67\x4f\x6f\xc4\xf9\xc4\x2d\x38\xcf\x59\x2b\xa7\x84\xea\x7a\xca\x74\x44\x35\xb5\x25\x48\x67\x9c\xa5\x12\x3f\x54\x6e\x45\x86\x42\x67\x3d\x73\x0d\x31\xcf\xfb\x19\x6b\x2b\x84\x5b\xf5\x78\x3c\x94\x40\x02\x31\xd6\x9e\x51\x0f\x8f\xb9\x23\x74\x02\x54\x90\x62\x0c\x5b\x09\x43\xb8\xea\xb1\xdb\xad\x91\x38\x23\xeb\x02\x39\x43\x15\x52\xff\xb0\x19\x9a\x39\x2c\x65\xee\x5e\xed\x36\xd4\xde\x89\x92\x36\x19\x13\x82\x7a\x8e\x7e\x9e\x9b\x61\x27\x7d\x57\x81\x9f\x38\x53\x1d\x87\x0b\x3e\xa6\x28\x53\xf0\x75\x33\x9a\xbc\xe7\x76\xc9\xcd\xeb\x86\xa7\x1e\x52\xc8\x2f\xdd\x9b\x6d\x64\xd0\xb9\x62\x51\x29\xb2\xa6\x05\x42\xcf\x59\xb0\x54\x34\xee\xc2\x39\x73\xfb\xad\x7f\x17\x73\x88\xf9\xb8\xf9\xf2\x73\xc2\x3d\x3e\x0f\xa7\x9f\x06\x78\xa6\xb3\xb1\x76\xfd\xdc\xae\xea\x23\xed\xf0\x05\x5e\xc7\x77\x36\x21\x3e\x0a\xea\x75\xb5\xf6\xd7\x12\x3a\x7b\x6a\x07\x38\xf9\x2e\x8a\xfe\xbf\x38\xe6\x06\xbe\x0b\x48\x38\x92\xf2\x5f\x35\x70\x9a\xaa\x5b\x34\x78\x29\xce\xfd\xa1\x65\xc8\x1a\xfd\x48\x76\x15\x14\x9e\x13\x77\xa3\xa5\xbf\x79\x89\x6f\x8a\x3c\xcc\xe6\xa8\xc4\xe1\x33\xb8\x28\xe3\x9f\xf8\xfb\x33\x00\x00\xff\xff\x4c\x59\x10\xd5\xbe\x06\x00\x00")
 
 func ccmYamlBytes() ([]byte, error) {
 	return bindataRead(

pkg/server/server.go

@@ -23,7 +23,6 @@ import (
 	"github.com/k3s-io/k3s/pkg/nodepassword"
 	"github.com/k3s-io/k3s/pkg/rootlessports"
 	"github.com/k3s-io/k3s/pkg/secretsencrypt"
-	"github.com/k3s-io/k3s/pkg/servicelb"
 	"github.com/k3s-io/k3s/pkg/static"
 	"github.com/k3s-io/k3s/pkg/util"
 	"github.com/k3s-io/k3s/pkg/version"
@@ -187,10 +186,9 @@ func coreControllers(ctx context.Context, sc *Context, config *Config) error {
 		return err
 	}
 
-	// apply SystemDefaultRegistry setting to Helm and ServiceLB before starting controllers
+	// apply SystemDefaultRegistry setting to Helm before starting controllers
 	if config.ControlConfig.SystemDefaultRegistry != "" {
 		helm.DefaultJobImage = config.ControlConfig.SystemDefaultRegistry + "/" + helm.DefaultJobImage
-		servicelb.DefaultLBImage = config.ControlConfig.SystemDefaultRegistry + "/" + servicelb.DefaultLBImage
 	}
 
 	if !config.ControlConfig.DisableHelmController {
@@ -205,21 +203,6 @@ func coreControllers(ctx context.Context, sc *Context, config *Config) error {
 			sc.Core.Core().V1().ConfigMap())
 	}
 
-	if err := servicelb.Register(ctx,
-		sc.K8s,
-		sc.Apply,
-		sc.Apps.Apps().V1().DaemonSet(),
-		sc.Apps.Apps().V1().Deployment(),
-		sc.Core.Core().V1().Node(),
-		sc.Core.Core().V1().Pod(),
-		sc.Core.Core().V1().Service(),
-		sc.Core.Core().V1().Endpoints(),
-		config.ControlConfig.ServiceLBNamespace,
-		!config.ControlConfig.DisableServiceLB,
-		config.ControlConfig.Rootless); err != nil {
-		return err
-	}
-
 	if config.ControlConfig.EncryptSecrets {
 		if err := secretsencrypt.Register(ctx,
 			sc.K8s,

pkg/util/wait.go

+package util
+
+import "context"
+
+// ContextForChannel derives a child context from a parent channel.
+//
+// The derived context's Done channel is closed when the returned cancel function
+// is called or when the parent channel is closed, whichever happens first.
+//
+// Note the caller must *always* call the CancelFunc, otherwise resources may be leaked.
+//
+// Backported from https://github.com/kubernetes/apimachinery/blob/v0.25.2/pkg/util/wait/wait.go#L299
+func ContextForChannel(parentCh <-chan struct{}) (context.Context, context.CancelFunc) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	go func() {
+		select {
+		case <-parentCh:
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+	return ctx, cancel
+}