Feed controller refactor and tests (#213)

* Set up controller-runtime manager and example

This is an example of a controller-runtime controller-manager command
and a sample controller with table-driven controller tests.

The controller-runtime project contains packages meant to make
controllers easier to write with less boilerplate.

Table testing is inspired by similar work in knative/serving, modified
to work with controller-runtime reconcilers and clients.

* Refactor feed reconcile to be level-based

The previous feed controller waited for the start/stop job to finish,
blocking reconcile. That behavior was untestable without actually running
containers, and would have scaled poorly.

This refactor changes that to create Jobs without waiting for them to
complete, updating Feed status when Jobs are updated. This follows the
standard controller architecture used by Kubernetes and
knative/serving controllers.

Table tests have been added with 49% coverage, which isn't great but
better than zero.

This refactored controller uses the controller-runtime libraries to
reduce the boilerplate required.

* use name feedlet

* removing other status types expect ready.

* Renamed BIND to FEED like master had.

* Watch Job events

When a Job event occurs and that Job is owned by a Feed, enqueue the
owning Feed.

* Missed a feetlet.'

* Update e2e test for the new FeedReady condition

* Use the feeds kind when getting a feed

$KIND isn't defined so it breaks the command. Also, get the full yaml of
the object instead of just the name.

* Clean up debug logging

* Use correct service account for controller-manager

* Remove sample controller and clean up comments

Sample controller isn't that useful now that the feed controller
exists.

* Revert "Update e2e test for the new FeedReady condition"

This reverts commit f200c13.

* Also dump the feed start job in e2e test

* remove stray argument to kubectl get

This seems like a copy/paste error.

* Correct name of test helper

getNewBindJob was incorrectly renamed to getNewFeedJob as if Bind was a
noun, but it was actually a verb (so corresponds to Start instead).

* Pass an empty encoded context to start jobs

The event source wrapper expects context to always be valid json, event
for start jobs which ignore the decoded context. The pod template now
creates an empty context for start jobs.

* Add job failed reason.

* use the correct condition reasons

The e2e test expects FeedSuccess and FeedFailed.

Gopkg.toml

@@ -54,21 +54,21 @@ required = [
   name = "gopkg.in/yaml.v2"
   version = "v2.2.1"
 
-[[constraint]]
+[[override]]
   name = "k8s.io/api"
-  version = "kubernetes-1.10.0"
+  version = "kubernetes-1.10.1"
 
-[[constraint]]
+[[override]]
   name = "k8s.io/apimachinery"
-  version = "kubernetes-1.10.0"
+  version = "kubernetes-1.10.1"
 
-[[constraint]]
+[[override]]
   name = "k8s.io/code-generator"
-  version = "kubernetes-1.10.0"
+  version = "kubernetes-1.10.1"
 
-[[constraint]]
+[[override]]
   name = "k8s.io/client-go"
-  version = "kubernetes-1.10.0"
+  version = "kubernetes-1.10.1"
 
 [[override]]
   name = "github.com/golang/protobuf"

cmd/controller-manager/main.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2018 The Knative Authors
+
+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 main
+
+import (
+	"flag"
+	"log"
+
+	buildv1alpha1 "github.com/knative/build/pkg/apis/build/v1alpha1"
+	channelsv1alpha1 "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
+	feedsv1alpha1 "github.com/knative/eventing/pkg/apis/feeds/v1alpha1"
+	"github.com/knative/eventing/pkg/controller/feed"
+	istiov1alpha3 "github.com/knative/serving/pkg/apis/istio/v1alpha3"
+	servingv1alpha1 "github.com/knative/serving/pkg/apis/serving/v1alpha1"
+	"k8s.io/apimachinery/pkg/runtime"
+	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+	"sigs.k8s.io/controller-runtime/pkg/runtime/signals"
+)
+
+// SchemeFunc adds types to a Scheme.
+type SchemeFunc func(*runtime.Scheme) error
+
+// ProvideFunc adds a controller to a Manager.
+type ProvideFunc func(manager.Manager) (controller.Controller, error)
+
+func main() {
+	flag.Parse()
+	logf.SetLogger(logf.ZapLogger(false))
+
+	// Setup a Manager
+	mrg, err := manager.New(config.GetConfigOrDie(), manager.Options{})
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// Add custom types to this array to get them into the manager's scheme.
+	schemeFuncs := []SchemeFunc{
+		buildv1alpha1.AddToScheme,
+		servingv1alpha1.AddToScheme,
+		feedsv1alpha1.AddToScheme,
+		channelsv1alpha1.AddToScheme,
+		istiov1alpha3.AddToScheme,
+	}
+	for _, schemeFunc := range schemeFuncs {
+		schemeFunc(mrg.GetScheme())
+	}
+
+	// Add each controller's ProvideController func to this list to have the
+	// manager run it.
+	providers := []ProvideFunc{
+		feed.ProvideController,
+	}
+
+	for _, provider := range providers {
+		if _, err := provider(mrg); err != nil {
+			log.Fatal(err)
+		}
+	}
+
+	log.Fatal(mrg.Start(signals.SetupSignalHandler()))
+}

cmd/controller/main.go

@@ -39,7 +39,6 @@ import (
 	"github.com/knative/eventing/pkg/controller/bus"
 	"github.com/knative/eventing/pkg/controller/channel"
 	"github.com/knative/eventing/pkg/controller/clusterbus"
-	"github.com/knative/eventing/pkg/controller/feed"
 	"github.com/knative/eventing/pkg/controller/flow"
 	"github.com/knative/eventing/pkg/signals"
 
@@ -96,7 +95,6 @@ func main() {
 
 	// Add new controllers here.
 	ctors := []controller.Constructor{
-		feed.NewController,
 		flow.NewController,
 		bus.NewController,
 		clusterbus.NewController,

config/controller-manager.yaml

@@ -0,0 +1,33 @@
+# Copyright 2018 The Knative Authors
+#
+# 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.
+apiVersion: apps/v1beta1
+kind: Deployment
+metadata:
+  name: controller-manager
+  namespace: knative-eventing
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: controller-manager
+    spec:
+      serviceAccountName: eventing-controller
+      containers:
+      - name: controller-manager
+        image: github.com/knative/eventing/cmd/controller-manager
+        args: [
+          "-logtostderr",
+          "-stderrthreshold", "INFO",
+        ]

pkg/apis/feeds/v1alpha1/feed_types.go

@@ -19,6 +19,7 @@ package v1alpha1
 import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"k8s.io/apimachinery/pkg/runtime"
 )
@@ -236,3 +237,63 @@ func (fs *FeedStatus) RemoveCondition(t FeedConditionType) {
 	}
 	fs.Conditions = conditions
 }
+
+func (fs *FeedStatus) InitializeConditions() {
+	for _, cond := range []FeedConditionType{
+		FeedConditionReady,
+	} {
+		if fc := fs.GetCondition(cond); fc == nil {
+			fs.SetCondition(&FeedCondition{
+				Type:   cond,
+				Status: corev1.ConditionUnknown,
+			})
+		}
+	}
+}
+
+// AddFinalizer adds the given value to the list of finalizers if it doesn't
+// already exist.
+func (f *Feed) AddFinalizer(value string) {
+	finalizers := sets.NewString(f.GetFinalizers()...)
+	finalizers.Insert(value)
+	f.SetFinalizers(finalizers.List())
+}
+
+// RemoveFinalizer removes the given value from the list of finalizers if it
+// exists.
+func (f *Feed) RemoveFinalizer(value string) {
+	finalizers := sets.NewString(f.GetFinalizers()...)
+	finalizers.Delete(value)
+	if finalizers.Len() == 0 {
+		// if no finalizers, set to nil list, not an empty slice.
+		f.SetFinalizers([]string(nil))
+	} else {
+		f.SetFinalizers(finalizers.List())
+	}
+}
+
+// HasFinalizer returns true if a finalizer exists, or false otherwise.
+func (f *Feed) HasFinalizer(value string) bool {
+	for _, f := range f.GetFinalizers() {
+		if f == value {
+			return true
+		}
+	}
+	return false
+}
+
+// SetOwnerReference adds the given owner reference to the list of owner
+// references, replacing the corresponding owner reference if it exists.
+func (f *Feed) SetOwnerReference(or *metav1.OwnerReference) {
+	var refs []metav1.OwnerReference
+
+	for _, ref := range f.GetOwnerReferences() {
+		if !(ref.APIVersion == or.APIVersion &&
+			ref.Kind == or.Kind &&
+			ref.Name == or.Name) {
+			refs = append(refs, ref)
+		}
+	}
+	refs = append(refs, *or)
+	f.SetOwnerReferences(refs)
+}