Merge pull request oracle#372 in OKE/oci-cloud-controller-manager from feature/serverless-1.24-service-controller to release-1.24

* commit 'd61ea944fc461864626a157bc721d3f5e4186023':
  Introducing custom service controller to support SKE clusters

Author: Akshay Kumar

cmd/cloud-provider-oci/app/cloud_provider_oci.go

@@ -52,6 +52,7 @@ import (
 	csicontroller "github.com/oracle/oci-cloud-controller-manager/cmd/oci-csi-controller-driver/csi-controller"
 	"github.com/oracle/oci-cloud-controller-manager/cmd/oci-csi-controller-driver/csioptions"
 	"github.com/oracle/oci-cloud-controller-manager/controllers"
+	"github.com/oracle/oci-cloud-controller-manager/pkg/cloudprovider/providers/oci"
 	providercfg "github.com/oracle/oci-cloud-controller-manager/pkg/cloudprovider/providers/oci/config"
 	"github.com/oracle/oci-cloud-controller-manager/pkg/logging"
 	"github.com/oracle/oci-cloud-controller-manager/pkg/metrics"
@@ -231,7 +232,8 @@ func run(logger *zap.SugaredLogger, config *cloudControllerManagerConfig.Complet
 		defer wg.Done()
 		// Run starts all the cloud controller manager control loops.
 		cloudProvider := cloudInitializer(logger, config)
-		controllerInitializers := cloudControllerManager.ConstructControllerInitializers(cloudControllerManager.DefaultInitFuncConstructors, config, cloudProvider)
+
+		controllerInitializers := cloudControllerManager.ConstructControllerInitializers(getInitFuncConstructors(), config, cloudProvider)
 		// TODO move to newer cloudControllerManager dependency that provides a way to pass channel/context
 		if err := cloudControllerManager.Run(config, cloudProvider, controllerInitializers, ctx.Done()); err != nil {
 			logger.With(zap.Error(err)).Error("Error running cloud controller manager")
@@ -368,3 +370,19 @@ func getOCIClient(logger *zap.SugaredLogger, config *providercfg.Config) client.
 	}
 	return c
 }
+
+func getInitFuncConstructors() map[string]cloudControllerManager.ControllerInitFuncConstructor{
+	// Disable default service controller
+	cloudControllerManager.ControllersDisabledByDefault.Insert("service")
+
+	// Add custom service controller init func
+	defaultConstructors := cloudControllerManager.DefaultInitFuncConstructors
+	defaultConstructors["oci-service"] = cloudControllerManager.ControllerInitFuncConstructor{
+		InitContext: cloudControllerManager.ControllerInitContext{
+			ClientName: "service-controller",
+		},
+		Constructor: oci.StartOciServiceControllerWrapper,
+	}
+
+	return defaultConstructors
+}

cmd/oci-cloud-controller-manager/main.go

@@ -21,7 +21,7 @@ import (
 	"os"
 	"time"
 
-	_ "github.com/oracle/oci-cloud-controller-manager/pkg/cloudprovider/providers/oci"
+	"github.com/oracle/oci-cloud-controller-manager/pkg/cloudprovider/providers/oci"
 	"github.com/oracle/oci-cloud-controller-manager/pkg/logging"
 	"github.com/spf13/pflag"
 	"go.uber.org/zap"
@@ -53,7 +53,7 @@ func main() {
 	}
 
 	fss := cliflag.NamedFlagSets{}
-	command := app.NewCloudControllerManagerCommand(s, cloudInitializer, app.DefaultInitFuncConstructors, fss, wait.NeverStop)
+	command := app.NewCloudControllerManagerCommand(s, cloudInitializer, getInitFuncConstructors(), fss, wait.NeverStop)
 
 	// TODO: once we switch everything over to Cobra commands, we can go back to calling
 	// utilflag.InitFlags() (by removing its pflag.Parse() call). For now, we have to set the
@@ -94,3 +94,19 @@ func cloudInitializer(config *config.CompletedConfig) cloudprovider.Interface {
 
 	return cloud
 }
+
+func getInitFuncConstructors() map[string]app.ControllerInitFuncConstructor{
+	// Disable default service controller
+	app.ControllersDisabledByDefault.Insert("service")
+
+	// Add custom service controller init func
+	defaultConstructors := app.DefaultInitFuncConstructors
+	defaultConstructors["oci-service"] = app.ControllerInitFuncConstructor{
+		InitContext: app.ControllerInitContext{
+			ClientName: "service-controller",
+		},
+		Constructor: oci.StartOciServiceControllerWrapper,
+	}
+
+	return defaultConstructors
+}

go.mod

@@ -61,6 +61,7 @@ require (
 	github.com/spf13/cobra v1.4.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.8.1
+	github.com/stretchr/testify v1.7.0
 	go.uber.org/zap v1.19.1
 	golang.org/x/net v0.0.0-20220403103023-749bd193bc2b
 	golang.org/x/sys v0.0.0-20220406163625-3f8b81556e12
@@ -74,6 +75,7 @@ require (
 	k8s.io/cloud-provider v0.24.1
 	k8s.io/component-base v0.24.1
 	k8s.io/component-helpers v0.24.1
+	k8s.io/controller-manager v0.24.1
 	k8s.io/csi-translation-lib v0.24.1
 	k8s.io/klog v1.0.0
 	k8s.io/klog/v2 v2.60.1
@@ -136,6 +138,7 @@ require (
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/selinux v1.10.0 // indirect
 	github.com/pelletier/go-toml v1.9.3 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.33.0 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
@@ -179,7 +182,6 @@ require (
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.24.1 // indirect
-	k8s.io/controller-manager v0.24.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20220401212409-b28bf2818661 // indirect
 	k8s.io/kube-scheduler v0.0.0 // indirect
 	k8s.io/kubectl v0.0.0 // indirect

manifests/cloud-controller-manager/oci-cloud-controller-manager-rbac.yaml

@@ -133,6 +133,23 @@ rules:
   - list
   - watch
   - patch
+
+- apiGroups:
+  - ""
+  resources:
+  - endpointslices
+  verbs:
+  - get
+  - list
+  - watch
+
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  verbs:
+  - get
+  - list
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1beta1

pkg/cloudprovider/providers/oci/ccm.go

@@ -26,11 +26,19 @@ import (
 	"go.uber.org/zap"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
 	listersv1 "k8s.io/client-go/listers/core/v1"
+	discoverylistersv1 "k8s.io/client-go/listers/discovery/v1"
 	"k8s.io/client-go/tools/cache"
 	cloudprovider "k8s.io/cloud-provider"
+	cloudControllerManager "k8s.io/cloud-provider/app"
+	"k8s.io/cloud-provider/app/config"
+	cloudcontrollerconfig "k8s.io/cloud-provider/app/config"
+	genericcontrollermanager "k8s.io/controller-manager/app"
+	"k8s.io/controller-manager/controller"
+	"k8s.io/klog/v2"
 
 	providercfg "github.com/oracle/oci-cloud-controller-manager/pkg/cloudprovider/providers/oci/config"
 	"github.com/oracle/oci-cloud-controller-manager/pkg/metrics"
@@ -58,7 +66,8 @@ type CloudProvider struct {
 	// NodeLister provides a cache to lookup nodes for deleting a load balancer.
 	// Due to limitations in the OCI API around going from an IP to a subnet
 	// we use the node lister to go from IP -> node / provider id -> ... -> subnet
-	NodeLister listersv1.NodeLister
+	NodeLister          listersv1.NodeLister
+	EndpointSliceLister discoverylistersv1.EndpointSliceLister
 
 	client     client.Interface
 	kubeclient clientset.Interface
@@ -175,15 +184,21 @@ func (cp *CloudProvider) Initialize(clientBuilder cloudprovider.ControllerClient
 
 	nodeInformer := factory.Core().V1().Nodes()
 	go nodeInformer.Informer().Run(wait.NeverStop)
+
 	serviceInformer := factory.Core().V1().Services()
 	go serviceInformer.Informer().Run(wait.NeverStop)
+
+	endpointSliceInformer := factory.Discovery().V1().EndpointSlices()
+	go endpointSliceInformer.Informer().Run(wait.NeverStop)
+
 	go nodeInfoController.Run(wait.NeverStop)
 
 	cp.logger.Info("Waiting for node informer cache to sync")
-	if !cache.WaitForCacheSync(wait.NeverStop, nodeInformer.Informer().HasSynced, serviceInformer.Informer().HasSynced) {
+	if !cache.WaitForCacheSync(wait.NeverStop, nodeInformer.Informer().HasSynced, serviceInformer.Informer().HasSynced, endpointSliceInformer.Informer().HasSynced) {
 		utilruntime.HandleError(fmt.Errorf("Timed out waiting for informers to sync"))
 	}
 	cp.NodeLister = nodeInformer.Lister()
+	cp.EndpointSliceLister = endpointSliceInformer.Lister()
 
 	cp.securityListManagerFactory = func(mode string) securityListManager {
 		if cp.config.LoadBalancer.Disabled {
@@ -248,3 +263,31 @@ func (cp *CloudProvider) HasClusterID() bool {
 func instanceCacheKeyFn(obj interface{}) (string, error) {
 	return *obj.(*core.Instance).Id, nil
 }
+
+func StartOciServiceControllerWrapper(initContext cloudControllerManager.ControllerInitContext, completedConfig *cloudcontrollerconfig.CompletedConfig, cloud cloudprovider.Interface) cloudControllerManager.InitFunc {
+	return func(ctx context.Context, controllerContext genericcontrollermanager.ControllerContext) (controller.Interface, bool, error) {
+		return startOciServiceController(ctx, initContext, completedConfig, cloud)
+	}
+}
+
+func startOciServiceController(ctx context.Context, initContext cloudControllerManager.ControllerInitContext, completedConfig *config.CompletedConfig, cloud cloudprovider.Interface) (controller.Interface, bool, error) {
+	// Start the service controller
+	serviceController, err := NewServiceController(
+		cloud,
+		completedConfig.ClientBuilder.ClientOrDie(initContext.ClientName),
+		completedConfig.SharedInformers.Core().V1().Services(),
+		completedConfig.SharedInformers.Core().V1().Nodes(),
+		completedConfig.SharedInformers.Discovery().V1().EndpointSlices(),
+		completedConfig.ComponentConfig.KubeCloudShared.ClusterName,
+		5*time.Second,
+		utilfeature.DefaultFeatureGate,
+	)
+	if err != nil {
+		klog.Errorf("Failed to start OCI service controller: %v", err)
+		return nil, false, err
+	}
+
+	go serviceController.Run(ctx, int(completedConfig.ComponentConfig.ServiceController.ConcurrentServiceSyncs))
+
+	return nil, true, nil
+}