Bus monitor polish #382

scothis · 2018-08-17T20:41:43Z

General bus enhancements lifted from #378, since that PR is blocked.

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 synched
addressed all govet and golint reports from pkg/buses

n3wscott · 2018-08-20T15:55:21Z

pkg/buses/gcppubsub/bus.go

@@ -105,6 +112,8 @@ func (b *PubSubBus) CreateOrUpdateSubscription(sub *channelsv1alpha1.Subscriptio
 	return err
 }

+// DeleteSubscription removes a Subscription from Cloud Pub/Sub for a Knative


I bet this is a bug based on the CreateOrUpdate comment: or idempotently updates a Subscription if it already exists. which means it will delete adopted topics where we don't mean to. We should be adding something to the metadata that states if we did real work on the remote topic/subscription or not.

Buses should be in total control of their infrastructure, the subscription is prefixed to avoid collisions with a non bus-created resources. While collisions are possible (need to fix #253), it should be rare and is certainly not a feature.

n3wscott · 2018-08-20T15:56:04Z

pkg/buses/gcppubsub/bus.go

@@ -122,6 +131,8 @@ func (b *PubSubBus) DeleteSubscription(sub *channelsv1alpha1.Subscription) error
 	return subscription.Delete(ctx)
 }

+// SendEventToTopic sends a message to the Cloud Pub/Sub Topic backing the
+// Channel.
 func (b *PubSubBus) SendEventToTopic(channel *channelsv1alpha1.Channel, message *buses.Message) error {


is this suppose to be a public method?

n3wscott · 2018-08-20T15:56:11Z

pkg/buses/gcppubsub/bus.go

+// ReceiveEvents will receive events from a Cloud Pub/Sub Subscription for a
+// Knative Subscription. This method will not block, but will continue to
+// receive events until either this method or StopReceiveEvents is called for
+// the same Subscription.
 func (b *PubSubBus) ReceiveEvents(sub *channelsv1alpha1.Subscription, parameters buses.ResolvedParameters) error {


is this suppose to be a public method?

n3wscott · 2018-08-20T16:01:10Z

/lgtm

Minor comments mostly not related to this change but to the code in general.

evankanderson

(Letting @n3wscott finish reviewing and lgtm)

evankanderson · 2018-08-20T19:17:01Z

pkg/buses/kafka/dispatcher/dispatcher.go

-	sInitial := parameters[InitialOffset]
-	if sInitial == Oldest {
+func resolveInitialOffset(parameters buses.ResolvedParameters) (int64, error) {
+	sInitial := parameters[initialOffset]


Switch or a map would be slightly cleaner here; i.e.

const saramaOffset = map[string]int64{ "newest": saramaNewest, "oldest": saramaOldest, } func resolveInitialOffset(...) (int64, error) { offset, ok := saramaOffset[parameters[initialOffset] if !ok { return 0, fmt.Errorf(...) } return offset, nil }

switched to a switch

evankanderson · 2018-08-20T19:21:37Z

pkg/buses/monitor.go

-			glog.Warningf("Could not update status: %v", errS)
+		if !equality.Semantic.DeepEqual(subscription.Status, subscriptionCopy.Status) {
+			// status has changed, update subscription
+			_, errS := monitor.eventingClient.ChannelsV1alpha1().Subscriptions(subscription.Namespace).Update(subscriptionCopy)


It feels strange to be tracking two errors here.

it is a bit funky. We want always want to update the resource status, if the original err being recorded in the resource status, we want to return that error since it's more meaningful. If the update failed, we still want to return that error to trigger the work queue to process the resource again as the update can fail because of a failed optimistic lock.

Will add a comment in the code

evankanderson · 2018-08-20T19:25:47Z

pkg/buses/references.go

 func (r *BusReference) String() string {
-	return fmt.Sprintf("%s/%s", r.Namespace, r.Name)
+	if len(r.Namespace) != 0 {


I have a slight preference for r.Namespace != ""; it makes bugs like #377 slightly easier to catch, IMO.

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

scothis · 2018-08-22T22:38:57Z

@n3wscott @evankanderson PTAL

I addressed your feedback and also pulled in the other refactoring I've been working on. The diff is rather scary, but mostly is moving items around to create components that are similar, more cleanly defined and testable. There's more to do (especially adding tests and polishing rough spots), but I like the general direction and it's a better foundation.

From the commit message:

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

n3wscott

Looks like there are new directories, can you confirm there are new tests in each new directory?

n3wscott · 2018-08-22T22:40:51Z

pkg/buses/bus.go

+// 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 {
+	bus := &bus{


I would invert this method to have bus := $bus{...} after the opts validation has happened, and then that condenses the calls like bus.cache = opts.Cache into just setting inside the struct creation.

will do, although it's not quite as clean as you'd sugest because there's a cyclic dependency between the bus and the receiver I still need to untangle.

n3wscott · 2018-08-22T22:43:29Z

pkg/buses/bus.go

+		handlerFuncs: handlerFuncs,
+	}
+
+	if opts.Cache == nil {


opts is allowed to be nil, so you should wrap all the defaulting with a nil check on opts.

n3wscott · 2018-08-22T22:48:01Z

pkg/buses/bus.go

+}
+
+func (b *bus) dispatchMessage(subscription *channelsv1alpha1.Subscription, message *Message) error {
+	subscriber := subscription.Spec.Subscriber


can subscription be nil here? should we check?

I can't think of a reason to why it should be allowed to be empty. I'll add a check to the validator.

n3wscott · 2018-08-22T22:50:50Z

pkg/buses/cache.go

+// AddChannel adds, or updates, the provided channel to the cache for later
+// retrieal by its reference.
+func (c *Cache) AddChannel(channel *channelsv1alpha1.Channel) {
+	channelRef := NewChannelReference(channel)


nil check channel? return error?

adding a check and test

n3wscott · 2018-08-22T22:50:56Z

pkg/buses/cache.go

+
+// RemoveChannel removes the provided channel from the cache.
+func (c *Cache) RemoveChannel(channel *channelsv1alpha1.Channel) {
+	channelRef := NewChannelReference(channel)


nil check channel? return error?

adding a check and test

n3wscott · 2018-08-22T23:06:42Z

pkg/buses/gcppubsub/bus.go

+		return nil
+	}
+
+	glog.Infof("Delete topic %q\n", topicID)


I wonder if this refactor is the correct time to move to zap or just plain log?

yes, but in a follow-up PR

n3wscott · 2018-08-22T23:07:25Z

pkg/buses/gcppubsub/bus.go

+		return err
+	} else if exists {
+		// TODO update subscription configuration
+		// _, err := subscription.Update(b.ctx, pubsub.SubscriptionConfigToUpdate{})


this seems important?

It's not important right now because there are no bus params, thus the pub/sub subscription will never need to be mutated.

amending the current comment to make it more clear

n3wscott · 2018-08-22T23:10:27Z

pkg/buses/handler_funcs.go

+		err = h.ProvisionFunc(channelRef, parameters)
+		channelCopy := channel.DeepCopy()
+		var cond *channelsv1alpha1.ChannelCondition
+		if err != nil {


I tend to see the if error != nil block directly after the setting of the involved err variable.

in this case you could use the

if err := h.ProvisionFunc(channelRef, parameters); err != nil { ... } else { ... }

style

oh i see, you use this error much further down as well...

n3wscott · 2018-08-22T23:17:20Z

pkg/buses/handler_funcs.go

+			_, errS := reconciler.eventingClient.ChannelsV1alpha1().Channels(channel.Namespace).Update(channelCopy)
+			if errS != nil {
+				glog.Warningf("Could not update channel status: %v", errS)
+				if err != nil {


this is pretty strange. It feels real weird to have two errors and not report both of them. Maybe smush them together as a string in a new error?

will smoosh them into a single error

n3wscott · 2018-08-22T23:18:55Z

pkg/buses/handler_funcs.go

+func (h EventHandlerFuncs) onUnprovision(channel *channelsv1alpha1.Channel, reconciler *Reconciler) error {
+	if h.UnprovisionFunc != nil {
+		channelRef := NewChannelReference(channel)
+		err := h.UnprovisionFunc(channelRef)


general style would have me changing this to:

if err := h.UnprovisionFunc(channelRef); err != nil { reconciler.RecordChannelEventf(channelRef, corev1.EventTypeWarning, errResourceSync, "Error unprovisioning channel: %s", err) // skip updating status conditions since the channel was deleted return err } else { reconciler.RecordChannelEventf(channelRef, corev1.EventTypeNormal, successSynced, "Channel unprovisioned successfully") }

scothis · 2018-08-23T14:49:56Z

@n3wscott PTAL

n3wscott · 2018-08-23T22:12:03Z

/lgtm

I think this is a really nice refactoring.

scothis · 2018-08-24T00:54:31Z

for approval:
/assign @evankanderson

evankanderson

/lgtm
/approve

evankanderson · 2018-08-27T23:21:50Z

pkg/buses/handler_funcs.go

+}
+
+func (h EventHandlerFuncs) onSubscribe(subscription *channelsv1alpha1.Subscription, reconciler *Reconciler) error {
+	if h.SubscribeFunc != nil {


In the future, convert this to early-return:

if h.SubscribeFunc == nil { return nil }

knative-prow-robot · 2018-08-27T23:34:58Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: evankanderson, scothis

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~OWNERS~~ [evankanderson]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

- return early from handlerfuncs if func is nil

scothis · 2018-08-28T00:09:06Z

@evankanderson PTAL. I resolved the conflict with master and addressed your comment

knative-metrics-robot · 2018-08-28T00:10:30Z

The following is the coverage report on pkg/.
Say /test pull-knative-eventing-go-coverage to run the coverage report again

File	Old Coverage	New Coverage	Delta
pkg/apis/channels/v1alpha1/subscription_validation.go	80.0%	84.2%	4.2
pkg/buses/bus.go	Do not exist	0.0%
pkg/buses/cache.go	Do not exist	100.0%
pkg/buses/gcppubsub/bus.go	Do not exist	0.0%
pkg/buses/handler_funcs.go	Do not exist	0.0%
pkg/buses/kafka/bus.go	Do not exist	0.0%
pkg/buses/reconciler.go	Do not exist	0.0%
pkg/buses/references.go	0.0%	100.0%	100.0
pkg/buses/stub/bus.go	Do not exist	0.0%

n3wscott · 2018-08-28T14:23:38Z

/lgtm

Follow-up from knative#382 (comment)

Follow-up from #382 (comment)

Adding back the incorrectly removed serving/servrless content

* Apply vendor patches on make generate-release Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com> * Align create-release-branch Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com> --------- Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

…native#394) * Apply vendor patches on make generate-release * Align create-release-branch --------- Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com> Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Apply vendor patches on make generate-release Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com> * Align create-release-branch Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com> --------- Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

knative-prow-robot assigned n3wscott Aug 17, 2018

knative-prow-robot added the size/L Denotes a PR that changes 100-499 lines, ignoring generated files. label Aug 17, 2018

knative-prow-robot requested review from evankanderson and vaikas August 17, 2018 20:41

n3wscott approved these changes Aug 20, 2018

View reviewed changes

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Aug 20, 2018

evankanderson reviewed Aug 20, 2018

View reviewed changes

scothis force-pushed the monitor-polish branch from a4eba78 to 3be8c55 Compare August 22, 2018 22:31

knative-prow-robot added size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. and removed lgtm Indicates that a PR is ready to be merged. size/L Denotes a PR that changes 100-499 lines, ignoring generated files. labels Aug 22, 2018

Update path to stub bus dispatcher for e2e tests

d08fcaf

n3wscott reviewed Aug 22, 2018

View reviewed changes

scothis added 3 commits August 22, 2018 19:46

Properly update existing channels in stub bus

cb8c668

Review feedback

fce72f2

Fix kafka bus

26605a1

scothis mentioned this pull request Aug 23, 2018

Log dispatch errors in stub bus #386

Merged

n3wscott approved these changes Aug 23, 2018

View reviewed changes

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Aug 23, 2018

knative-prow-robot assigned evankanderson Aug 24, 2018

evankanderson reviewed Aug 27, 2018

View reviewed changes

knative-prow-robot added the approved Indicates a PR has been approved by an approver from all required OWNERS files. label Aug 27, 2018

Merge remote-tracking branch 'upstream/master' into monitor-polish

928f10d

Review feedback

b0ece00

- return early from handlerfuncs if func is nil

knative-prow-robot removed the lgtm Indicates that a PR is ready to be merged. label Aug 28, 2018

knative-prow-robot added the lgtm Indicates that a PR is ready to be merged. label Aug 28, 2018

knative-prow-robot merged commit e9f182a into knative:master Aug 28, 2018

scothis added a commit to scothis/eventing that referenced this pull request Aug 30, 2018

Rename [bus|channel|subscription]Ref

971db43

Follow-up from knative#382 (comment)

scothis mentioned this pull request Aug 30, 2018

Rename [bus|channel|subscription]Ref #406

Merged

knative-prow-robot pushed a commit that referenced this pull request Sep 11, 2018

Rename [bus|channel|subscription]Ref (#406)

56542fa

Follow-up from #382 (comment)

n3wscott mentioned this pull request Sep 25, 2018

adding n3wscott to approvers. #465

Merged

matzew added a commit to matzew/eventing that referenced this pull request Jan 9, 2020

Merge pull request knative#382 from matzew/adding_back_serverless

e624b4d

Adding back the incorrectly removed serving/servrless content

Bus monitor polish #382

Bus monitor polish #382

Conversation

scothis commented Aug 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

n3wscott commented Aug 20, 2018

evankanderson left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scothis commented Aug 22, 2018

n3wscott left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

scothis commented Aug 23, 2018

n3wscott commented Aug 23, 2018

scothis commented Aug 24, 2018

evankanderson left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

knative-prow-robot commented Aug 27, 2018

scothis commented Aug 28, 2018

knative-metrics-robot commented Aug 28, 2018

n3wscott commented Aug 28, 2018