Flow controller

cmd/controller/main.go

@@ -25,6 +25,7 @@ import (
 	"github.com/golang/glog"
 	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
@@ -39,6 +40,7 @@ import (
 	"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"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -81,13 +83,21 @@ func main() {
 		glog.Fatalf("Error building serving clientset: %s", err.Error())
 	}
 
+	// Build a rest.Config from configuration injected into the Pod by
+	// Kubernetes. Clients will use the Pod's ServiceAccount principal.
+	restConfig, err := rest.InClusterConfig()
+	if err != nil {
+		glog.Fatalf("Error building rest config: %v", err.Error())
+	}
+
 	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)
 	informerFactory := informers.NewSharedInformerFactory(client, time.Second*30)
 	servingInformerFactory := servinginformers.NewSharedInformerFactory(servingClient, time.Second*30)
 
 	// Add new controllers here.
 	ctors := []controller.Constructor{
 		feed.NewController,
+		flow.NewController,
 		bus.NewController,
 		clusterbus.NewController,
 		channel.NewController,
@@ -97,7 +107,7 @@ func main() {
 	controllers := make([]controller.Interface, 0, len(ctors))
 	for _, ctor := range ctors {
 		controllers = append(controllers,
-			ctor(kubeClient, client, servingClient, kubeInformerFactory, informerFactory, servingInformerFactory))
+			ctor(kubeClient, client, servingClient, restConfig, kubeInformerFactory, informerFactory, servingInformerFactory))
 	}
 
 	go kubeInformerFactory.Start(stopCh)

config/flow.yaml

@@ -0,0 +1,24 @@
+# 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: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: flows.flows.knative.dev
+spec:
+  group: flows.knative.dev
+  version: v1alpha1
+  names:
+    kind: Flow
+    plural: flows
+  scope: Namespaced

hack/update-codegen.sh

@@ -27,7 +27,7 @@ CODEGEN_PKG=${CODEGEN_PKG:-$(cd ${SCRIPT_ROOT}; ls -d -1 ./vendor/k8s.io/code-ge
 #                  instead of the $GOPATH directly. For normal projects this can be dropped.
 ${CODEGEN_PKG}/generate-groups.sh "deepcopy,client,informer,lister" \
   github.com/knative/eventing/pkg/client github.com/knative/eventing/pkg/apis \
-  "channels:v1alpha1 feeds:v1alpha1" \
+  "channels:v1alpha1 feeds:v1alpha1 flows:v1alpha1" \
   --go-header-file ${SCRIPT_ROOT}/hack/boilerplate/boilerplate.go.txt
 
 # Make sure our dependencies are up-to-date

pkg/apis/flows/v1alpha1/flow_types.go

@@ -20,6 +20,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	feedsv1alpha1 "github.com/knative/eventing/pkg/apis/feeds/v1alpha1"
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -162,24 +163,7 @@ type EventTrigger struct {
 	// by the actual trigger actuator.
 	// NOTE: experimental field. All secrets in ParametersFrom will be
 	// resolved and given to event sources in the Parameters field.
-	ParametersFrom []ParametersFromSource `json:"parametersFrom,omitempty"`
-}
-
-// ParametersFromSource represents the source of a set of Parameters
-// TODO: consider making this into a new secret type.
-type ParametersFromSource struct {
-	// The Secret key to select from.
-	// The value must be a JSON object.
-	//+optional
-	SecretKeyRef *SecretKeyReference `json:"secretKeyRef,omitempty"`
-}
-
-// SecretKeyReference references a key of a Secret.
-type SecretKeyReference struct {
-	// The name of the secret in the resource's namespace to select from.
-	Name string `json:"name"`
-	// The key of the secret to select from.  Must be a valid secret key.
-	Key string `json:"key"`
+	ParametersFrom []feedsv1alpha1.ParametersFromSource `json:"parametersFrom,omitempty"`
 }
 
 // FlowStatus is the status for a Flow resource

pkg/apis/flows/v1alpha1/register.go

@@ -0,0 +1,53 @@
+/*
+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 v1alpha1
+
+import (
+	"github.com/knative/eventing/pkg/apis/flows"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// SchemeGroupVersion is group version used to register these objects
+var SchemeGroupVersion = schema.GroupVersion{Group: flows.GroupName, Version: "v1alpha1"}
+
+// Kind takes an unqualified kind and returns back a Group qualified GroupKind
+func Kind(kind string) schema.GroupKind {
+	return SchemeGroupVersion.WithKind(kind).GroupKind()
+}
+
+// Resource takes an unqualified resource and returns a Group qualified GroupResource
+func Resource(resource string) schema.GroupResource {
+	return SchemeGroupVersion.WithResource(resource).GroupResource()
+}
+
+var (
+	SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
+	AddToScheme   = SchemeBuilder.AddToScheme
+)
+
+// Adds the list of known types to Scheme.
+func addKnownTypes(scheme *runtime.Scheme) error {
+	scheme.AddKnownTypes(SchemeGroupVersion,
+		&Flow{},
+		&FlowList{},
+	)
+	metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
+	return nil
+}