Bus monitor polish (#382)

* Bus Refactor

Clarify and solidify the public interfaces for a Bus. Common behavior
has moved into composable units that can individually tests or injected
into the bus for testing.

Bus implementors generally only need to provide custom implementations
for up to six methods defined by EventHandlerFuncs.

Core components:
- [Bus|Channel|Subscription]Reference: decouples bus from the raw
    k8s resources
- Cache: lookup and presist raw resource types for a reference
- EventHandlerFuncs: the contract for bus implementations
- Reconciler: reads resource changes from the API server, saves the
    resource in the Cache and emits events to the EventHandlerFuncs
- MessageDispatcher: sends messages to a destination over the event
    delivery protocol
- MessageReceiver: receive messages for a channel on the bus over the
    event delivery protocol

Behavior changes:
- avoid unnessesary status updates if the status has not changed
- avoid supressing errors when updating status
- avoid supressing updates for unchanged resources as the local copy may
  not be fully reconciled

* Update path to stub bus dispatcher for e2e tests

* Properly update existing channels in stub bus

* Review feedback

* Fix kafka bus

* Review feedback

- return early from handlerfuncs if func is nil

config/buses/stub/stub-bus.yaml

@@ -18,7 +18,7 @@ metadata:
 spec:
   dispatcher:
     name: dispatcher
-    image: github.com/knative/eventing/pkg/buses/stub
+    image: github.com/knative/eventing/pkg/buses/stub/dispatcher
     args: [
       "-logtostderr",
       "-stderrthreshold", "INFO",

pkg/apis/channels/v1alpha1/subscription_validation.go

@@ -27,12 +27,16 @@ func (s *Subscription) Validate() *apis.FieldError {
 }
 
 func (ss *SubscriptionSpec) Validate() *apis.FieldError {
-	if len(ss.Channel) == 0 {
+	if ss.Channel == "" {
 		fe := apis.ErrMissingField("channel")
 		fe.Details = "the Subscription must reference a Channel"
 		return fe
 	}
-	// TODO: should subscriptions have a subscriber?
+	if ss.Subscriber == "" {
+		fe := apis.ErrMissingField("subscriber")
+		fe.Details = "the Subscription must reference a Subscriber"
+		return fe
+	}
 	return nil
 }
 

pkg/apis/channels/v1alpha1/subscription_validation_test.go

@@ -16,9 +16,10 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"testing"
+
 	"github.com/google/go-cmp/cmp"
 	"github.com/knative/pkg/apis"
-	"testing"
 )
 
 func TestSubscriptionSpecValidation(t *testing.T) {
@@ -28,25 +29,29 @@ func TestSubscriptionSpecValidation(t *testing.T) {
 		want *apis.FieldError
 	}{{
 		name: "valid",
-		c: &SubscriptionSpec{
-			Channel: "foo",
-		},
-		want: nil,
-	}, {
-		name: "valid with subscriber",
 		c: &SubscriptionSpec{
 			Channel:    "bar",
 			Subscriber: "foo",
 		},
 		want: nil,
 	}, {
-		name: "valid with subscriber and arguments",
+		name: "valid with arguments",
 		c: &SubscriptionSpec{
 			Channel:    "bar",
 			Subscriber: "foo",
 			Arguments:  &[]Argument{{Name: "foo", Value: "bar"}},
 		},
 		want: nil,
+	}, {
+		name: "missing subscriber",
+		c: &SubscriptionSpec{
+			Channel: "foo",
+		},
+		want: func() *apis.FieldError {
+			fe := apis.ErrMissingField("subscriber")
+			fe.Details = "the Subscription must reference a Subscriber"
+			return fe
+		}(),
 	}, {
 		name: "empty",
 		c:    &SubscriptionSpec{},

pkg/buses/bus.go

@@ -0,0 +1,169 @@
+/*
+ * 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 buses
+
+import (
+	"fmt"
+
+	channelsv1alpha1 "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
+)
+
+type bus struct {
+	busRef       BusReference
+	handlerFuncs EventHandlerFuncs
+
+	reconciler *Reconciler
+	cache      *Cache
+	dispatcher *MessageDispatcher
+	receiver   *MessageReceiver
+}
+
+// BusOpts holds configuration options for new buses. These options are not
+// required for proper operation of the bus, but are useful to override the
+// default behavior and for testing.
+type BusOpts struct {
+	// MasterURL is the address of the Kubernetes API server. Overrides any
+	// value in kubeconfig. Only required if out-of-cluster.
+	MasterURL string
+	// KubeConfig is the path to a kubeconfig. Only required if out-of-cluster.
+	KubeConfig string
+
+	// Cache to use for this bus. A cache will be created for the bus if not
+	// specified.
+	Cache *Cache
+	// Reconciler to use for this bus. A reconciler wil be created for the bus
+	// if not specified.
+	Reconciler *Reconciler
+	// MessageDispatcher to use for this bus. The message dispatcher is used to
+	// send messages from the bus to a subscriber. A message dispatcher will be
+	// created for the bus if needed and not specified.
+	MessageDispatcher *MessageDispatcher
+	// MessageReceiver to use for this bus. The message receiver is used to
+	// receive message sent to a channel. A message receiver will be created
+	// for the bus if needed and not specified.
+	MessageReceiver *MessageReceiver
+}
+
+// BusProvisioner provisions channels and subscriptions for a bus on backing
+// infrastructure.
+type BusProvisioner interface {
+	Run(threadiness int, stopCh <-chan struct{})
+}
+
+// NewBusProvisioner creates a new provisioner for a specific bus.
+// EventHandlerFuncs are used to be notified when a channel or subscription is
+// created, updated or removed.
+func NewBusProvisioner(busRef BusReference, handlerFuncs EventHandlerFuncs, opts *BusOpts) BusProvisioner {
+	if opts == nil {
+		opts = &BusOpts{}
+	}
+	if opts.Cache == nil {
+		opts.Cache = NewCache()
+	}
+	if opts.Reconciler == nil {
+		opts.Reconciler = NewReconciler(Provisioner, opts.MasterURL, opts.KubeConfig, opts.Cache, handlerFuncs)
+	}
+
+	return &bus{
+		busRef:       busRef,
+		handlerFuncs: handlerFuncs,
+		cache:        opts.Cache,
+		reconciler:   opts.Reconciler,
+	}
+}
+
+// BusDispatcher dispatches messages from channels to subscribers via backing
+// infrastructure.
+type BusDispatcher interface {
+	Run(threadiness int, stopCh <-chan struct{})
+	DispatchMessage(subscriptionRef SubscriptionReference, message *Message) error
+}
+
+// NewBusDispatcher creates a new dispatcher for a specific bus.
+// EventHandlerFuncs are used to be notified when a subscription is created,
+// updated or removed, or a message is received.
+func NewBusDispatcher(busRef BusReference, handlerFuncs EventHandlerFuncs, opts *BusOpts) BusDispatcher {
+	var b *bus
+
+	if opts == nil {
+		opts = &BusOpts{}
+	}
+	if opts.Cache == nil {
+		opts.Cache = NewCache()
+	}
+	if opts.Reconciler == nil {
+		opts.Reconciler = NewReconciler(Dispatcher, opts.MasterURL, opts.KubeConfig, opts.Cache, handlerFuncs)
+	}
+	if opts.MessageDispatcher == nil {
+		opts.MessageDispatcher = NewMessageDispatcher()
+	}
+	if opts.MessageReceiver == nil {
+		opts.MessageReceiver = NewMessageReceiver(func(channelRef ChannelReference, message *Message) error {
+			return b.receiveMessage(channelRef, message)
+		})
+	}
+
+	b = &bus{
+		busRef:       busRef,
+		handlerFuncs: handlerFuncs,
+
+		cache:      opts.Cache,
+		reconciler: opts.Reconciler,
+		dispatcher: opts.MessageDispatcher,
+		receiver:   opts.MessageReceiver,
+	}
+
+	return b
+}
+
+// Run starts the bus's processing.
+func (b bus) Run(threadiness int, stopCh <-chan struct{}) {
+	go b.reconciler.Run(b.busRef, threadiness, stopCh)
+	b.reconciler.WaitForCacheSync(stopCh)
+	if b.receiver != nil {
+		go b.receiver.Run(stopCh)
+	}
+
+	<-stopCh
+}
+
+func (b *bus) receiveMessage(channelRef ChannelReference, message *Message) error {
+	_, err := b.cache.Channel(channelRef)
+	if err != nil {
+		return ErrUnknownChannel
+	}
+	return b.handlerFuncs.onReceiveMessage(channelRef, message)
+}
+
+// DispatchMessage sends a message to a subscriber. This function is only
+// avilable for bus dispatchers.
+func (b *bus) DispatchMessage(subscriptionRef SubscriptionReference, message *Message) error {
+	subscription, err := b.cache.Subscription(subscriptionRef)
+	if err != nil {
+		return fmt.Errorf("unable to dispatch to unknown subscription %q", subscriptionRef.String())
+	}
+	return b.dispatchMessage(subscription, message)
+}
+
+func (b *bus) dispatchMessage(subscription *channelsv1alpha1.Subscription, message *Message) error {
+	subscriber := subscription.Spec.Subscriber
+	defaults := DispatchDefaults{
+		Namespace: subscription.Namespace,
+		ReplyTo:   subscription.Spec.ReplyTo,
+	}
+	return b.dispatcher.DispatchMessage(message, subscriber, defaults)
+}

pkg/buses/cache.go

@@ -0,0 +1,98 @@
+/*
+ * 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 buses
+
+import (
+	"fmt"
+
+	channelsv1alpha1 "github.com/knative/eventing/pkg/apis/channels/v1alpha1"
+)
+
+// NewCache create a cache that is able to save and retrive Channels and
+// Subscriptions by their reference.
+func NewCache() *Cache {
+	return &Cache{
+		channels:      make(map[ChannelReference]*channelsv1alpha1.Channel),
+		subscriptions: make(map[SubscriptionReference]*channelsv1alpha1.Subscription),
+	}
+}
+
+// Cache able to save and retrive Channels and Subscriptions by their
+// reference. It is used by the reconciler to track which resources have been
+// provisioned and comparing updated resources to the provisioned version.
+type Cache struct {
+	channels      map[ChannelReference]*channelsv1alpha1.Channel
+	subscriptions map[SubscriptionReference]*channelsv1alpha1.Subscription
+}
+
+// Channel returns a cached channel for provided reference or an error if the
+// channel is not in the cache.
+func (c *Cache) Channel(channelRef ChannelReference) (*channelsv1alpha1.Channel, error) {
+	channel, ok := c.channels[channelRef]
+	if !ok {
+		return nil, fmt.Errorf("unknown channel %q", channelRef.String())
+	}
+	return channel, nil
+}
+
+// Subscription returns a cached subscription for provided reference or an
+// error if the subscription is not in the cache.
+func (c *Cache) Subscription(subscriptionRef SubscriptionReference) (*channelsv1alpha1.Subscription, error) {
+	subscription, ok := c.subscriptions[subscriptionRef]
+	if !ok {
+		return nil, fmt.Errorf("unknown subscription %q", subscriptionRef.String())
+	}
+	return subscription, nil
+}
+
+// AddChannel adds, or updates, the provided channel to the cache for later
+// retrieal by its reference.
+func (c *Cache) AddChannel(channel *channelsv1alpha1.Channel) {
+	if channel == nil {
+		return
+	}
+	channelRef := NewChannelReference(channel)
+	c.channels[channelRef] = channel
+}
+
+// RemoveChannel removes the provided channel from the cache.
+func (c *Cache) RemoveChannel(channel *channelsv1alpha1.Channel) {
+	if channel == nil {
+		return
+	}
+	channelRef := NewChannelReference(channel)
+	delete(c.channels, channelRef)
+}
+
+// AddSubscription adds, or updates, the provided subscription to the cache for
+// later retrieal by its reference.
+func (c *Cache) AddSubscription(subscription *channelsv1alpha1.Subscription) {
+	if subscription == nil {
+		return
+	}
+	subscriptionRef := NewSubscriptionReference(subscription)
+	c.subscriptions[subscriptionRef] = subscription
+}
+
+// RemoveSubscription removes the provided subscription from the cache.
+func (c *Cache) RemoveSubscription(subscription *channelsv1alpha1.Subscription) {
+	if subscription == nil {
+		return
+	}
+	subscriptionRef := NewSubscriptionReference(subscription)
+	delete(c.subscriptions, subscriptionRef)
+}