Merge pull request #44 from SteveKMin/stateful

Adding statefulset functionality

config/rbac/manager_role.yaml

@@ -42,6 +42,44 @@ rules:
   - create
   - update
   - patch
+- apiGroups:
+  - apps
+  resources:
+  - statefulsets
+  verbs:
+  - get
+  - list
+  - watch
+  - update
+  - patch
+- apiGroups:
+  - ""
+  resources:
+  - configmaps
+  verbs:
+  - get
+  - list
+  - watch
+  - update
+  - patch
+- apiGroups:
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - get
+  - list
+  - watch
+  - update
+  - patch
+- apiGroups:
+  - ""
+  resources:
+  - events
+  verbs:
+  - create
+  - update
+  - patch
 - apiGroups:
   - admissionregistration.k8s.io
   resources:

pkg/controller/add_statefulset.go

@@ -0,0 +1,26 @@
+/*
+Copyright 2018 Pusher Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"github.com/pusher/wave/pkg/controller/statefulset"
+)
+
+func init() {
+	// AddToManagerFuncs is a list of functions to create controllers and add them to a manager.
+	AddToManagerFuncs = append(AddToManagerFuncs, statefulset.Add)
+}

pkg/controller/deployment/deployment_controller_test.go

@@ -108,7 +108,7 @@ var _ = Describe("Deployment controller Suite", func() {
 		m.Create(deployment).Should(Succeed())
 		waitForDeploymentReconciled(deployment)
 
-		ownerRef = utils.GetOwnerRef(deployment)
+		ownerRef = utils.GetOwnerRefDeployment(deployment)
 	})
 
 	AfterEach(func() {

pkg/controller/statefulset/statefulset_controller.go

@@ -0,0 +1,111 @@
+/*
+Copyright 2018 Pusher Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package statefulset
+
+import (
+	"context"
+
+	"github.com/pusher/wave/pkg/core"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+// Add creates a new StatefulSet Controller and adds it to the Manager with default RBAC. The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func Add(mgr manager.Manager) error {
+	return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+	return &ReconcileStatefulSet{
+		scheme:  mgr.GetScheme(),
+		handler: core.NewHandler(mgr.GetClient(), mgr.GetEventRecorderFor("wave")),
+	}
+}
+
+// add adds a new Controller to mgr with r as the reconcile.Reconciler
+func add(mgr manager.Manager, r reconcile.Reconciler) error {
+	// Create a new controller
+	c, err := controller.New("statefulset-controller", mgr, controller.Options{Reconciler: r})
+	if err != nil {
+		return err
+	}
+
+	// Watch for changes to StatefulSet
+	err = c.Watch(&source.Kind{Type: &appsv1.StatefulSet{}}, &handler.EnqueueRequestForObject{})
+	if err != nil {
+		return err
+	}
+
+	// Watch ConfigMaps owned by a StatefulSet
+	err = c.Watch(&source.Kind{Type: &corev1.ConfigMap{}}, &handler.EnqueueRequestForOwner{
+		IsController: false,
+		OwnerType:    &appsv1.StatefulSet{},
+	})
+	if err != nil {
+		return err
+	}
+
+	// Watch Secrets owned by a StatefulSet
+	err = c.Watch(&source.Kind{Type: &corev1.Secret{}}, &handler.EnqueueRequestForOwner{
+		IsController: false,
+		OwnerType:    &appsv1.StatefulSet{},
+	})
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+var _ reconcile.Reconciler = &ReconcileStatefulSet{}
+
+// ReconcileStatefulSet reconciles a StatefulSet object
+type ReconcileStatefulSet struct {
+	scheme  *runtime.Scheme
+	handler *core.Handler
+}
+
+// Reconcile reads that state of the cluster for a StatefulSet object and
+// updates its PodSpec based on mounted configuration
+// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;update;patch
+// +kubebuilder:rbac:groups=,resources=configmaps,verbs=get;list;watch;update;patch
+// +kubebuilder:rbac:groups=,resources=secrets,verbs=get;list;watch;update;patch
+// +kubebuilder:rbac:groups=,resources=events,verbs=create;update;patch
+func (r *ReconcileStatefulSet) Reconcile(request reconcile.Request) (reconcile.Result, error) {
+	// Fetch the StatefulSet instance
+	instance := &appsv1.StatefulSet{}
+	err := r.handler.Get(context.TODO(), request.NamespacedName, instance)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			// Object not found, return.  Created objects are automatically garbage collected.
+			return reconcile.Result{}, nil
+		}
+		// Error reading the object - requeue the request.
+		return reconcile.Result{}, err
+	}
+
+	return r.handler.HandleStatefulSet(instance)
+}

pkg/controller/statefulset/statefulset_controller_suite_test.go

@@ -0,0 +1,89 @@
+/*
+Copyright 2018 Pusher Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package statefulset
+
+import (
+	"log"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"github.com/pusher/wave/test/reporters"
+
+	"github.com/go-logr/glogr"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/pusher/wave/pkg/apis"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+)
+
+var cfg *rest.Config
+
+func TestMain(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecsWithDefaultAndCustomReporters(t, "Wave Controller Suite", reporters.Reporters())
+}
+
+var t *envtest.Environment
+
+var _ = BeforeSuite(func() {
+	t = &envtest.Environment{
+		CRDDirectoryPaths: []string{filepath.Join("..", "..", "..", "config", "crds")},
+	}
+	apis.AddToScheme(scheme.Scheme)
+
+	logf.SetLogger(glogr.New())
+
+	var err error
+	if cfg, err = t.Start(); err != nil {
+		log.Fatal(err)
+	}
+})
+
+var _ = AfterSuite(func() {
+	t.Stop()
+})
+
+// SetupTestReconcile returns a reconcile.Reconcile implementation that delegates to inner and
+// writes the request to requests after Reconcile is finished.
+func SetupTestReconcile(inner reconcile.Reconciler) (reconcile.Reconciler, chan reconcile.Request) {
+	requests := make(chan reconcile.Request)
+	fn := reconcile.Func(func(req reconcile.Request) (reconcile.Result, error) {
+		result, err := inner.Reconcile(req)
+		requests <- req
+		return result, err
+	})
+	return fn, requests
+}
+
+// StartTestManager adds recFn
+func StartTestManager(mgr manager.Manager) (chan struct{}, *sync.WaitGroup) {
+	stop := make(chan struct{})
+	wg := &sync.WaitGroup{}
+	go func() {
+		defer GinkgoRecover()
+		wg.Add(1)
+		Expect(mgr.Start(stop)).NotTo(HaveOccurred())
+		wg.Done()
+	}()
+	return stop, wg
+}