[WIP] Convert Flow controller to provider/reconcile model

cmd/controller-manager/main.go

@@ -22,7 +22,9 @@ import (
 	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"
+	flowsv1alpha1 "github.com/knative/eventing/pkg/apis/flows/v1alpha1"
 	"github.com/knative/eventing/pkg/controller/feed"
+	"github.com/knative/eventing/pkg/controller/flow"
 	istiov1alpha3 "github.com/knative/serving/pkg/apis/istio/v1alpha3"
 	servingv1alpha1 "github.com/knative/serving/pkg/apis/serving/v1alpha1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -53,10 +55,11 @@ func main() {
 	// Add custom types to this array to get them into the manager's scheme.
 	schemeFuncs := []SchemeFunc{
 		buildv1alpha1.AddToScheme,
-		servingv1alpha1.AddToScheme,
-		feedsv1alpha1.AddToScheme,
 		channelsv1alpha1.AddToScheme,
+		feedsv1alpha1.AddToScheme,
+		flowsv1alpha1.AddToScheme,
 		istiov1alpha3.AddToScheme,
+		servingv1alpha1.AddToScheme,
 	}
 	for _, schemeFunc := range schemeFuncs {
 		schemeFunc(mrg.GetScheme())
@@ -66,6 +69,7 @@ func main() {
 	// manager run it.
 	providers := []ProvideFunc{
 		feed.ProvideController,
+		flow.ProvideController,
 	}
 
 	for _, provider := range providers {

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/flow"
 	"github.com/knative/eventing/pkg/signals"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -82,6 +81,8 @@ func main() {
 		glog.Fatalf("Error building serving clientset: %s", err.Error())
 	}
 
+	// TODO: Rip this out from all the controllers since we can get it
+	// from provider.
 	// Build a rest.Config from configuration injected into the Pod by
 	// Kubernetes. Clients will use the Pod's ServiceAccount principal.
 	restConfig, err := rest.InClusterConfig()
@@ -95,7 +96,6 @@ func main() {
 
 	// Add new controllers here.
 	ctors := []controller.Constructor{
-		flow.NewController,
 		bus.NewController,
 		clusterbus.NewController,
 		channel.NewController,

pkg/apis/flows/v1alpha1/flow_types.go

@@ -278,6 +278,19 @@ func (fs *FlowStatus) PropagateActionTargetResolved(status corev1.ConditionStatu
 	fs.checkAndMarkReady()
 }
 
+func (fs *FlowStatus) InitializeConditions() {
+	for _, cond := range []FlowConditionType{
+		FlowConditionReady,
+	} {
+		if fc := fs.GetCondition(cond); fc == nil {
+			fs.setCondition(&FlowCondition{
+				Type:   cond,
+				Status: corev1.ConditionUnknown,
+			})
+		}
+	}
+}
+
 func (fs *FlowStatus) PropagateChannelStatus(cs channelsv1alpha1.ChannelStatus) {
 	cc := cs.GetCondition(channelsv1alpha1.ChannelReady)
 

pkg/apis/flows/v1alpha1/flow_types_test.go

@@ -339,3 +339,18 @@ func TestFlowCondition_PropagateStatus(t *testing.T) {
 		})
 	}
 }
+
+func TestFlowCondition_InitializeConditions(t *testing.T) {
+	flow := Flow{}
+	flow.Status.InitializeConditions()
+	cond := flow.Status.GetCondition(FlowConditionReady)
+	if cond == nil {
+		t.Fatalf("InitializeConditions didn't set FlowConditionReady")
+	}
+	if want, got := cond.Status, corev1.ConditionUnknown; want != got {
+		t.Fatalf("Status was not set correctly : \nwant:\t%#v\ngot:\t%#v", want, got)
+	}
+	if flow.Status.IsReady() {
+		t.Fatalf("InitializeConditions marked the flow Ready")
+	}
+}