Merge pull request #424 from mengqiy/nonleaderelection

support HA (non leader election) components

Author: k8s-ci-robot

examples/crd/pkg/doc.go renamed to examples/crd/pkg/groupversion_info.go

@@ -14,10 +14,25 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+// +kubebuilder:object:generate=true
+// +groupName=chaosapps.metamagical.io
 package pkg
 
 import (
+"k8s.io/apimachinery/pkg/runtime/schema"
 logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+"sigs.k8s.io/controller-runtime/pkg/scheme"
 )
 
-var log = logf.Log.WithName("chaospod-resource")
+var (
+log = logf.Log.WithName("chaospod-resource")
+
+// SchemeGroupVersion is group version used to register these objects
+SchemeGroupVersion = schema.GroupVersion{Group: "chaosapps.metamagical.io", Version: "v1"}
+
+// SchemeBuilder is used to add go types to the GroupVersionKind scheme
+SchemeBuilder = &scheme.Builder{GroupVersion: SchemeGroupVersion}
+
+// AddToScheme is required by pkg/client/...
+AddToScheme = SchemeBuilder.AddToScheme
+)

examples/crd/pkg/resource.go

@@ -23,8 +23,6 @@ import (
 corev1 "k8s.io/api/core/v1"
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 "k8s.io/apimachinery/pkg/runtime"
-"k8s.io/apimachinery/pkg/runtime/schema"
-"sigs.k8s.io/controller-runtime/pkg/scheme"
 "sigs.k8s.io/controller-runtime/pkg/webhook"
 )
 
@@ -41,12 +39,11 @@ type ChaosPodStatus struct {
 LastRun metav1.Time `json:"lastRun,omitempty"`
 }
 
-// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +kubebuilder:object:root=true
 
 // ChaosPod is the Schema for the randomjobs API
 // +kubebuilder:printcolumn:name="next stop",type="string",JSONPath=".spec.nextStop",format="date"
 // +kubebuilder:printcolumn:name="last run",type="string",JSONPath=".status.lastRun",format="date"
-// +k8s:openapi-gen=true
 type ChaosPod struct {
 metav1.TypeMeta `json:",inline"`
 metav1.ObjectMeta `json:"metadata,omitempty"`
@@ -55,7 +52,7 @@ type ChaosPod struct {
 Status ChaosPodStatus `json:"status,omitempty"`
 }
 
-// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +kubebuilder:object:root=true
 
 // ChaosPodList contains a list of ChaosPod
 type ChaosPodList struct {
@@ -106,14 +103,3 @@ func (c *ChaosPod) Default() {
 func init() {
 SchemeBuilder.Register(&ChaosPod{}, &ChaosPodList{})
 }
-
-var (
-// SchemeGroupVersion is group version used to register these objects
-SchemeGroupVersion = schema.GroupVersion{Group: "chaosapps.metamagical.io", Version: "v1"}
-
-// SchemeBuilder is used to add go types to the GroupVersionKind scheme
-SchemeBuilder = &scheme.Builder{GroupVersion: SchemeGroupVersion}
-
-// AddToScheme is required by pkg/client/...
-AddToScheme = SchemeBuilder.AddToScheme
-)

pkg/manager/internal.go

@@ -58,8 +58,12 @@ type controllerManager struct {
 // to scheme.scheme.
 scheme *runtime.Scheme
 
-// runnables is the set of Controllers that the controllerManager injects deps into and Starts.
-runnables []Runnable
+// leaderElectionRunnables is the set of Controllers that the controllerManager injects deps into and Starts.
+// These Runnables are managed by lead election.
+leaderElectionRunnables []Runnable
+// nonLeaderElectionRunnables is the set of webhook servers that the controllerManager injects deps into and Starts.
+// These Runnables will not be blocked by lead election.
+nonLeaderElectionRunnables []Runnable
 
 cache cache.Cache
 
@@ -121,7 +125,7 @@ type controllerManager struct {
 retryPeriod time.Duration
 }
 
-// Add sets dependencies on i, and adds it to the list of runnables to start.
+// Add sets dependencies on i, and adds it to the list of Runnables to start.
 func (cm *controllerManager) Add(r Runnable) error {
 cm.mu.Lock()
 defer cm.mu.Unlock()
@@ -131,8 +135,13 @@ func (cm *controllerManager) Add(r Runnable) error {
 return err
 }
 
-// Add the runnable to the list
-cm.runnables = append(cm.runnables, r)
+// Add the runnable to the leader election or the non-leaderelection list
+if leRunnable, ok := r.(LeaderElectionRunnable); ok && !leRunnable.NeedLeaderElection() {
+cm.nonLeaderElectionRunnables = append(cm.nonLeaderElectionRunnables, r)
+} else {
+cm.leaderElectionRunnables = append(cm.leaderElectionRunnables, r)
+}
+
 if cm.started {
 // If already started, start the controller
 go func() {
@@ -254,13 +263,15 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 go cm.serveMetrics(cm.internalStop)
 }
 
+go cm.startNonLeaderElectionRunnables()
+
 if cm.resourceLock != nil {
 err := cm.startLeaderElection()
 if err != nil {
 return err
 }
 } else {
-go cm.start()
+go cm.startLeaderElectionRunnables()
 }
 
 select {
@@ -273,7 +284,7 @@ func (cm *controllerManager) Start(stop <-chan struct{}) error {
 }
 }
 
-func (cm *controllerManager) start() {
+func (cm *controllerManager) startNonLeaderElectionRunnables() {
 cm.mu.Lock()
 defer cm.mu.Unlock()
 
@@ -291,8 +302,26 @@ func (cm *controllerManager) start() {
 // TODO(community): Check the return value and write a test
 cm.cache.WaitForCacheSync(cm.internalStop)
 
-// Start the runnables after the cache has synced
-for _, c := range cm.runnables {
+// Start the non-leaderelection Runnables after the cache has synced
+for _, c := range cm.nonLeaderElectionRunnables {
+// Controllers block, but we want to return an error if any have an error starting.
+// Write any Start errors to a channel so we can return them
+ctrl := c
+go func() {
+cm.errChan <- ctrl.Start(cm.internalStop)
+}()
+}
+
+cm.started = true
+}
+
+func (cm *controllerManager) startLeaderElectionRunnables() {
+// Wait for the caches to sync.
+// TODO(community): Check the return value and write a test
+cm.cache.WaitForCacheSync(cm.internalStop)
+
+// Start the leader election Runnables after the cache has synced
+for _, c := range cm.leaderElectionRunnables {
 // Controllers block, but we want to return an error if any have an error starting.
 // Write any Start errors to a channel so we can return them
 ctrl := c
@@ -312,7 +341,7 @@ func (cm *controllerManager) startLeaderElection() (err error) {
 RetryPeriod: cm.retryPeriod,
 Callbacks: leaderelection.LeaderCallbacks{
 OnStartedLeading: func(_ context.Context) {
-cm.start()
+cm.startLeaderElectionRunnables()
 },
 OnStoppedLeading: func() {
 // Most implementations of leader election log.Fatal() here.

pkg/manager/manager.go

@@ -42,9 +42,11 @@ import (
 // Manager initializes shared dependencies such as Caches and Clients, and provides them to Runnables.
 // A Manager is required to create Controllers.
 type Manager interface {
-// Add will set reqeusted dependencies on the component, and cause the component to be
+// Add will set requested dependencies on the component, and cause the component to be
 // started when Start is called. Add will inject any dependencies for which the argument
-// implements the inject interface - e.g. inject.Client
+// implements the inject interface - e.g. inject.Client.
+// Depending on if a Runnable implements LeaderElectionRunnable interface, a Runnable can be run in either
+// non-leaderelection mode (always running) or leader election mode (managed by leader election if enabled).
 Add(Runnable) error
 
 // SetFields will set any dependencies on an object for which the object has implemented the inject
@@ -183,6 +185,13 @@ func (r RunnableFunc) Start(s <-chan struct{}) error {
 return r(s)
 }
 
+// LeaderElectionRunnable knows if a Runnable needs to be run in the leader election mode.
+type LeaderElectionRunnable interface {
+// NeedLeaderElection returns true if the Runnable needs to be run in the leader election mode.
+// e.g. controllers need to be run in leader election mode, while webhook server doesn't.
+NeedLeaderElection() bool
+}
+
 // New returns a new Manager for creating Controllers.
 func New(config *rest.Config, options Options) (Manager, error) {
 // Initialize a rest.config if none was specified

pkg/manager/manager_test.go

@@ -284,6 +284,10 @@ var _ = Describe("manger.Manager", func() {
 <-c2
 <-c3
 })
+
+It("should return an error if any non-leaderelection Components fail to Start", func() {
+// TODO(mengqiy): implement this after resolving https://github.com/kubernetes-sigs/controller-runtime/issues/429
+})
 }
 
 Context("with defaults", func() {