Provide an internal-only API to report connectivity state changes on the ClientConn

[my4-dev]

RELEASE NOTES:

• Provide an internal-only API to report connectivity state changes on the ClientConn

Motivation:

Please refer to #5818

Modifications:

• Add internal/clientutil/connectivity_state_pubsub.go
• Add publisher to connectivityStateManager in order to publish the connectivity changes on ClientConn to subscribers.

Result:

• We can get lossless updates of connectivity state changes on the ClientConn and use a callback.

type ClientController struct {
    cc *ClientConn
    ...
}

type Callback struct {
    c chan connectivity.State
}

func (callback *Callback) GetStateChannel() chan connectivity.State {
    return callback.c
}

func (callback *Callback)  ClientStateChangeListenOnChannel(state connectivity.State) {
    // process...
}

clientController.cc.csMgr.publisher.Register(Callback{...})

RELEASE NOTES: none

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: my4-dev / name: my4 (029f12c, f76b3b6)

[easwars]

@my4-dev : Thanks for taking a stab at this one. I feel that the changes in this PR don't quite add up to what I had imagined in #5818. Let me give an overview of what I had in mind.

I'm not too opposed to your package internal/clientutil, but I would prefer something like internal/connectivitystate. This package would look something like this:

package connectivitystate

// Watcher is the interface to be implemented by components which wish to receive
// connectivity state updates from the Tracker.
type Watcher interface {
  // Implementations must not block.
  OnStateChange(state connectivity.State)
}

// Tracker is equivalent to the Publisher type that you have. But it does more.
type Tracker struct {
  mu sync.Mutex // Access to the underlying fields needs to be guarded by this mutex.
  state connectivity.State // Caches the most recent state update
  watchers map[Watchers]bool // Set of registered watchers
}

func NewTracker() *Tracker { ... }

func (t *Tracker) AddWatcher(watcher Watcher) func () {
  // Adds the provided watcher to the set of watchers in Tracker.
  // Schedules a callback on the provided watcher with current state.
  // Returns a function to remove the provided watcher from the set of watchers. The caller
  // of this method is responsible for invoking this function when it no longer needs to
  // monitor the connectivity state changes on the channel.
}

func (t *Tracker) SetState(state connectivity.State) {
  // Update the cached state
  // Invoke callbacks on all registered watchers.
}

I would assume that you would also need the CallbackSerializer type defined at https://github.com/grpc/grpc-go/blob/master/xds/internal/xdsclient/callback_serializer.go. You would need to move it out of xds/internal to the top-level internal package (and update existing users).

Then, the ClientConn would need to have a field for the Tracker. This should be similar to what you currently have.

type ClientConn struct {
  ...
  connectivityStateTracker *tracker
  ...
}

The one thing I would want to add is an unexported method on the ClientConn, and a corresponding exported method internal package.

package internal

var AddConnectivityStateWatcher interface{} // The actual type for this would be `func(*grpc.ClientConn, connectivitystate.Watcher)

See examples in the internal package for how this is done for other cases to get an idea.

package grpc

func init() {
  internal.AddConnectivityStateWatcher = func(cc *grpc.ClientConn, w connectivitystate.Watcher) {
    if cc.connectivityStateTracker == nil {
      cc.connectivityStateTracker == connectivitystate.NewTracker()
    }
    cc.connectivityStateTracker.AddWatcher(w)
  }
}

We need to ensure that when there are no watchers, we don't do any wasted work.

@dfawley : What do you think about the above approach?

[my4-dev]

@easwars : Thank you for some comments! I have reflected your advice.

I couldn't understand why Testing piplelines failed.
The error logs shows that undefined: newCallbackSerializer, but I don't define such a function and couldn't find a path displayed in logs.

Could you please tell me how to solve this?

[github-actions]

This PR is labeled as requiring an update from the reporter, and no update has been received after 6 days. If no update is provided in the next 7 days, this issue will be automatically closed.

[my4-dev]

Hi @easwars!
I have corrected based on your advice.
Please take a look at my modification.

[easwars]

The tests are failing to with the following errors:

# google.golang.org/grpc/xds/internal/xdsclient [google.golang.org/grpc/xds/internal/xdsclient.test]
Error: xds/internal/xdsclient/authority_test.go:75:23: undefined: newCallbackSerializer

[my4-dev]

@easwars : I've request PR #6153 . Please take a look at this!

[easwars]

I don't mind doing this. I will.

Thank you very much.

[my4-dev]

While implementing test cases, I noticed that watcher.OnStateChange could be called with an incorrect argument because of characteristics of closure if t.state would be changed very quickly.

e.g.

1. SetState is called with connectivity.Idle.
2. watcher.OnStateChange is scheduled.
3. SetState is called with connectivity.Connecting immediately after step 2.
4. watcher.OnStateChange is scheduled.
5. Scheduled watcher.OnStateChange is called with t.state( = connectivity.Connecting) twice incorrectly.

In order to resolve this problem, I'm going to change t.state to state.
We would need to fix watcher.OnStateChange(t.state) in AddWatcher func as well.

Do you have any other opinions?
If so, I'll adopt yours.

[my4-dev]

I've fixed it from watcher.OnStateChange(t.state) to watcher.OnStateChange(state) for the same reason of SetState.

[my4-dev]

I've moved this code block here, added Lock() and changed to release resources of cc.csMgr.connectivityStateTracker.
It would be better to invoke this code block before cc.csMgr.updateState(connectivity.Shutdown) on Close method because it wasn't stable whether connectivity shutdown state change would be reported to Watcher.

[easwars]

I discussed this PR with @dfawley and we are leaning towards the following approach:

• Address the one outstanding comment in internal/grpcsync: move CallbackSerializer from xdsclient/internal to here #6153 and merge it
• Make a separate PR with only the changes currently in internal/connectivitystate/connectivitystate.go. We also want to make this a generic pubsub type instead of a specific pubsub type only for connectivity state changes. The actual code changes to make that happen will be minimal. Instead of the state field being of type connectivity.State, it will be of type interface {}. Similar changes would be required for OnStateChange, NewTracker and SetState.
  • Since we want to make it a generic pub-sub type, we would also have to bikeshed a little on the actual type and method names.
  • Also bikeshed a little on the actual package name and package path.
• Once we merge the generic pubsub type, we can make changes to the ClientConn.

We are already very close on most of the above, just need some minor massaging.

Please let me know what you think about this approach.

[my4-dev]

I agree with you. Let's go with this idea.
It would be better to open new Merge Request for each approachs based on this MR.

I want to go ahead the first approach.
I've replied to your comment. Please take a look at this.

• Address the one outstanding comment in internal/grpcsync: move CallbackSerializer from xdsclient/internal to here #6153 and merge it

[easwars]

I agree with you. Let's go with this idea. It would be better to open new Merge Request for each approachs based on this MR.

Thanks @my4-dev. I had one comment on #6153 that I forgot to hit the send button on :(
I've sent it now. Hopefully, we can merge that soon and move with a separate PR for pubsub type.

[easwars]

Now that #6153 is merged, we can move forward with the next PR where we define the generic pub-sub type. Thanks.

[my4-dev]

I've submitted the next PR #6167 .

[dfawley]

IIUC this is unblocked now?

[my4-dev]

Now that #6167 is merged, we can move forward with next PR where we make changes to the ClientConn.
It would be better to make changes like this in new PR.
So this PR can be closed.