[WIP] Heartbeat provisioner

cmd/controller/controller-runtime-main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/knative/eventing/pkg/controller/eventing/subscription"
 	"github.com/knative/eventing/pkg/controller/feed"
 	"github.com/knative/eventing/pkg/controller/flow"
+	containercontroller "github.com/knative/eventing/pkg/provisioners/container/controller"
 	"go.uber.org/zap"
 	"strings"
 
@@ -49,6 +50,7 @@ type ProvideFunc func(manager.Manager) (controller.Controller, error)
 // be added to the default providers list.
 var ExperimentalControllers = map[string]ProvideFunc{
 	"subscription.eventing.knative.dev": subscription.ProvideController,
+	"container-provisioner":             containercontroller.ProvideController,
 }
 
 // controllerRuntimeStart runs controllers written for controller-runtime. It's

config/500-controller.yaml

@@ -31,7 +31,7 @@ spec:
         args: [
           "-logtostderr",
           "-stderrthreshold", "INFO",
-          "--experimentalControllers=subscription.eventing.knative.dev" # comma separated list.
+          "--experimentalControllers=subscription.eventing.knative.dev,container-provisioner" # comma separated list.
         ]
         volumeMounts:
           - name: config-logging

pkg/apis/eventing/v1alpha1/register.go

@@ -49,6 +49,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		&ChannelList{},
 		&ClusterProvisioner{},
 		&ClusterProvisionerList{},
+		&Source{},
+		&SourceList{},
 		&Subscription{},
 		&SubscriptionList{},
 	)

pkg/apis/eventing/v1alpha1/source_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"fmt"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/knative/pkg/apis/duck"
@@ -111,6 +112,12 @@ type SourceStatus struct {
 	// +patchStrategy=merge
 	Conditions duckv1alpha1.Conditions `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type"`
 
+	// Provisioned holds the status of a Provisioned Object at a point in time.
+	// +optional
+	// +patchMergeKey=name
+	// +patchStrategy=merge
+	Provisioned []ProvisionedObjectStatus `json:"provisioned,omitempty" patchStrategy:"merge" patchMergeKey:"name"`
+
 	// ObservedGeneration is the 'Generation' of the Source that
 	// was last reconciled by the controller.
 	// +optional
@@ -120,6 +127,17 @@ type SourceStatus struct {
 	Subscribable duckv1alpha1.Subscribable `json:"subscribable,omitempty"`
 }
 
+type ProvisionedObjectStatus struct {
+	// Name of Object
+	Name string `json:"name,omitempty"`
+	// Type is the fully qualified object type.
+	Type string `json:"type,omitempty"`
+	// Status is the current relationship between Source and Object.
+	Status string `json:"status,omitempty"`
+	// Reason is the detailed description describing current relationship status.
+	Reason string `json:"reason,omitempty"`
+}
+
 // GetCondition returns the condition currently associated with the given type, or nil.
 func (ss *SourceStatus) GetCondition(t duckv1alpha1.ConditionType) *duckv1alpha1.Condition {
 	return sourceCondSet.Manage(ss).GetCondition(t)
@@ -145,6 +163,26 @@ func (ss *SourceStatus) MarkDeprovisioned(reason, messageFormat string, messageA
 	sourceCondSet.Manage(ss).MarkFalse(SourceConditionProvisioned, reason, messageFormat, messageA...)
 }
 
+// MarkProvisioned sets the condition that the source has had its backing resources created.
+func (ss *SourceStatus) SetProvisionedObjectState(name, objType, status, reasonFormat string, reasonA ...interface{}) {
+	reason := fmt.Sprintf(reasonFormat, reasonA...)
+	newP := ProvisionedObjectStatus{
+		Name:   name,
+		Type:   objType,
+		Status: status,
+		Reason: reason,
+	}
+	newProvisioned := make([]ProvisionedObjectStatus, 0, len(ss.Provisioned))
+	for _, p := range ss.Provisioned {
+		if p.Name == newP.Name {
+			newProvisioned = append(newProvisioned, newP)
+		} else {
+			newProvisioned = append(newProvisioned, p)
+		}
+	}
+	ss.Provisioned = newProvisioned
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // SourceList is a list of Source resources

pkg/controller/owner_references.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	channelsv1alpha "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
+	eventingv1alpha "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
 	feedsv1alpha "github.com/knative/eventing/pkg/apis/feeds/v1alpha1"
 	flowsv1alpha "github.com/knative/eventing/pkg/apis/flows/v1alpha1"
 )
@@ -51,6 +52,16 @@ func kind(obj metav1.Object) schema.GroupVersionKind {
 	case *flowsv1alpha.Flow:
 		return flowsv1alpha.SchemeGroupVersion.WithKind("Flow")
 
+	// Eventing
+	case *eventingv1alpha.Source:
+		return eventingv1alpha.SchemeGroupVersion.WithKind("Source")
+	case *eventingv1alpha.Channel:
+		return eventingv1alpha.SchemeGroupVersion.WithKind("Channel")
+	case *eventingv1alpha.ClusterProvisioner:
+		return eventingv1alpha.SchemeGroupVersion.WithKind("ClusterProvisioner")
+	case *eventingv1alpha.Subscription:
+		return eventingv1alpha.SchemeGroupVersion.WithKind("Subscription")
+
 	default:
 		panic(fmt.Sprintf("Unsupported object type %T", obj))
 	}

pkg/provisioners/container/config/container.yaml

@@ -0,0 +1,8 @@
+apiVersion: eventing.knative.dev/v1alpha1
+kind: ClusterProvisioner
+metadata:
+  name: container
+spec:
+  reconciles:
+    group: eventing.knative.dev
+    kind: Source

pkg/provisioners/container/controller/provider.go

@@ -0,0 +1,69 @@
+/*
+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 controller
+
+import (
+	"github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
+	"github.com/knative/eventing/pkg/provisioners/sdk"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+const (
+	// controllerAgentName is the string used by this controller to identify
+	// itself when creating events.
+	controllerAgentName = "container-provisioner-controller"
+)
+
+type reconciler struct {
+	client        client.Client
+	restConfig    *rest.Config
+	dynamicClient dynamic.Interface
+	recorder      record.EventRecorder
+}
+
+// ProvideController returns a Subscription controller.
+func ProvideController(mgr manager.Manager) (controller.Controller, error) {
+	p := &sdk.Provider{
+		AgentName: controllerAgentName,
+		Parent:    &v1alpha1.Source{},
+		Owns:      []runtime.Object{&appsv1.Deployment{}, &v1alpha1.Channel{}},
+		Reconciler: &reconciler{
+			recorder: mgr.GetRecorder(controllerAgentName),
+		},
+	}
+
+	return p.ProvideController(mgr)
+}
+
+func (r *reconciler) InjectClient(c client.Client) error {
+	r.client = c
+	return nil
+}
+
+func (r *reconciler) InjectConfig(c *rest.Config) error {
+	r.restConfig = c
+	var err error
+	r.dynamicClient, err = dynamic.NewForConfig(c)
+	return err
+}