Adding Watch API and initial implementation #2281

dahuang-purestorage · 2023-06-21T22:41:24Z

What this PR does / why we need it:
This PR implments a Watch gRPC call in libopenstorage on Volumes.
In porx, the driver is expected to implement GetVolumeWatcher which will return a go channel used in libopenstorage server to relay the volumes.
In libopenstorage, there is a list of watch connections that keeps track of the current clients. Also, there is a go routine constantly waiting from a volume from porx. Once a volume is recieved in the go channel, openstorage will relay it to the watch clients.

Which issue(s) this PR fixes (optional)
Closes # PWX-29706

Special notes for your reviewer:
TODO:

Adding Unit test
Figure out the best place to start watcher
Determine if it's possible to have a generic channel for additional event/resources (nodes, disks, etc). So it would be easier to add watch endpoint for them as well.

Testing:

Smoke testing with porx on creating, deleting and attaching volumes. Snippet of working output:

grpcurl -plaintext -d '{"volume_event": {"labels": [{}]}}' 0.0.0.0:9020  openstorage.api.OpenStorageWatch.Watch | jq '.volumeEvent.volume.id + " " + .volumeEvent.volume.status + " " + .volumeEvent.volume.state'
"441214862979454928 VOLUME_STATUS_UP VOLUME_STATE_DETACHED"
"27660964167843585 VOLUME_STATUS_UP VOLUME_STATE_ATTACHED"
"27660964167843585 VOLUME_STATUS_UP VOLUME_STATE_ATTACHED"
"27660964167843585 VOLUME_STATUS_UP VOLUME_STATE_DELETED"
....

lpabon · 2023-06-26T16:39:22Z

api/api.proto

@@ -2512,6 +2512,15 @@ service OpenStorageVolume {
 body: "*"
 };
 }
+
+ // Watch streams a list of volumes exist in the system.
+ rpc Watch(SdkVolumeWatchRequest)


This is good, but I would suggest creating a model of Watch for anything on the system, not in the Volumes service. That way we can watch for Pool, Disk, Volume, or any other type of event. Then the event message could use oneof for each type of storage event.

lpabon · 2023-06-28T15:34:18Z

api/api.proto

+// openstorage API. Right now, we will only support Volumes.
+service OpenStorageWatch{
+
+ // Watch streams of volumes exist in the system.


This comment should be changed since this is a generic event API

Please add more documentation here on how to use. Doc can be written in markdown format and will be used in the website https://libopenstorage.github.io

ggriffiths · 2023-06-28T22:35:17Z

volume/drivers/fake/fake.go

@@ -114,11 +118,35 @@ func newFakeDriver(params map[string]string) (*driver, error) {
 }
 }
 }
-
+ // go volumeGenerator(inst)


nit: remove commented code

ggriffiths · 2023-06-28T22:40:18Z

api/api.proto

@@ -3763,6 +3781,39 @@ message SdkVolumeSnapshotScheduleUpdateRequest {
 message SdkVolumeSnapshotScheduleUpdateResponse {
 }

+// Defines the request to watch a volume
+message SdkWatchRequest {
+ // one of event type


To reiterate Luis's comments, maybe expand on some of these comments as it's autogenerated for API docs

ggriffiths · 2023-06-28T22:41:31Z

api/server/sdk/watcher.go

+ volumeEventType = "Volume"
+)
+
+type WathcerServer struct {


typo: WathcerServer -> WatcherServer

ggriffiths · 2023-06-29T00:20:17Z

api/server/sdk/watcher.go

+ for event := range client.eventChannel {
+ var vol *api.Volume
+ if vol, ok = event.(*api.Volume); !ok {
+ continue


Log invalid type here so we know if there's any issues here

adityadani · 2023-06-29T00:06:31Z

api/api.proto

+}
+
+
+// OpenStorageWatcher is a service used to stream events in the system


nit: OpenStorageWatcher is a service that provides APIs for watching on resources and receive them as a stream of events.

adityadani · 2023-06-29T00:10:27Z

api/api.proto

+// OpenStorageWatcher is a service used to stream events in the system
+service OpenStorageWatch{
+
+ // Watch streams of event exist in the system


nit: Watch on resources managed by the driver and receive them as a stream of events.

adityadani · 2023-06-29T00:11:13Z

api/api.proto

+// Defines the request to watch an openstorage volume event for the given label
+// if the label is empty, it returns all the volume events
+message SdkVolumeWatchRequest {
+ // Options during watch


nit: fix comment - labels to filter out the volumes to watch on

adityadani · 2023-06-29T00:18:48Z

api/api.proto

+ rpc Watch(SdkWatchRequest)
+ returns (stream SdkWatchResponse) {
+ option(google.api.http) = {
+ post: "/v1/watch/watch"


I think this should be just /v1/watch. We are not watching on specific resources anymore so we don't need the extra prefix of service name anymore.

adityadani · 2023-06-29T00:19:26Z

api/api.proto

+ // Name of volume
+ string name = 2;
+ // Volume labels
+ map<string, string> labels = 3;


Do we need to return the labels separately? The labels are already a part of Volume

there could be additional labels that were not part of the request

Yeah I get that, but the volume object you are returning on line 3814, already has all the labels (even the ones that were not requested)

adityadani · 2023-06-29T00:30:34Z

api/server/sdk/watcher.go

+ if req.GetVolumeEvent() != nil {
+ return w.volumeWatch(req.GetVolumeEvent(), stream)
+ }
+ return nil


Should we return an error here instead since we were unable to parse the request and/or the object is unsupported?

adityadani · 2023-06-29T00:33:35Z

api/server/sdk/watcher.go

+ // wait for driver to be initialized with an non-empty volume watcher
+ for {
+ if s.volumeServer.driver(ctx) == nil {
+ continue


This will spin continuously if invoked before driver is initialized. Add a sleep may be?

Please add a sleep here before merging the PR

api/server/sdk/watcher.go

adityadani · 2023-06-29T00:44:07Z

api/server/sdk/watcher.go

+ client := watchConnection{
+ name: uuid.New(),
+ eventType: volumeEventType,
+ eventChannel: make(chan interface{}, 2),


Why create a buffer of only 2? This could lead to issues where for some reason the channel is full (say the stream is blocked) and we keep spinning up go routines on line 132 as part of callback invocation.

It would be better to have an unbounded channel, so that we keep queueing in the events, and avoid goroutines waiting on sending on this channel.

Unbuffered channel will wait on receiving the data too. AFAIK, unbuffered channel is a size 1 channel that waits on single event.

I was referring to an unbounded channel and not unbuffered. Essentially just an unbounded queue that will keep collecting the events.

adityadani · 2023-06-29T00:46:43Z

api/server/sdk/watcher.go

+
+ // wait only as long as context deadline allows
+ select {
+ case err := <-errChan:


Why not just do this here?

Suggested change

case err := <-errChan:

case err := <-group.Wait():

we could but we will need a for loop to wait for it. group.Wait() returns an error instead of a channel. Doing it this way couples well with ctx.Done.

ggriffiths · 2023-06-29T00:47:33Z

api/server/sdk/watcher.go

+ continue
+ }
+
+ volumeChannel, _ := s.volumeServer.driver(ctx).GetVolumeWatcher(&api.VolumeLocator{}, make(map[string]string))


Should we check the error here?

volume/volume.go

dahuang-purestorage · 2023-06-30T23:21:34Z

api/server/sdk/watcher.go

+ continue
+ }
+ resp := convertApiVolumeToSdkReponse(vol)
+ if err := stream.Send(resp); err != nil {


Figure out if we can set timeout whenever stream.Send fails

lpabon

👍🏼 I approve of of the changes to the API. I'll leave the implementation checks to Grant and Aditya.

github-actions · 2023-07-07T02:08:56Z

This PR is stale because it has been in review for 3 days with no activity.

adityadani

overall looks good. I have some pending questions on the PR. Also do you think we should write client stubs for the streaming API as well, so that the callers can simply use it w/ the correct server keepalive timeouts.

adityadani · 2023-07-07T21:51:14Z

api/api.proto

+ // Name of volume
+ string name = 2;
+ // Volume labels
+ map<string, string> labels = 3;


Yeah I get that, but the volume object you are returning on line 3814, already has all the labels (even the ones that were not requested)

adityadani · 2023-07-07T21:57:03Z

api/server/sdk/server.go

@@ -142,6 +142,8 @@ type ServerConfig struct {

 // Server is an implementation of the gRPC SDK interface
 type Server struct {
+ ctx context.Context


nit: can we rename this as watcherCtx and watcherCtxCancel so that it doesn't get used anywhere else?

adityadani · 2023-07-07T23:42:56Z

api/server/sdk/watcher.go

+ for event := range client.eventChannel {
+
+ // create a new context that will return error if execution took more than streamTimeout
+ timeoutCtx, timeoutCancelled := context.WithTimeout(ctx, streamTimeout)


I am not following how this timeoutCtx is being enforced? Isn't steam.Send() a blocking function?

adityadani · 2023-07-07T23:43:17Z

api/server/sdk/watcher.go

+ return err
+ }
+ if timeoutCtx.Err() != nil {
+ logrus.Warnf("context error: %v", timeoutCtx.Err())


nit: add more info in this log line

adityadani · 2023-07-07T23:45:12Z

api/server/sdk/watcher.go

+ // wait for driver to be initialized with an non-empty volume watcher
+ for {
+ if s.volumeServer.driver(ctx) == nil {
+ continue


Please add a sleep here before merging the PR

adityadani · 2023-07-07T23:45:52Z

api/server/sdk/watcher.go

+ }
+ s.RUnlock()
+ case <-ctx.Done():
+ logrus.Infof("exiting volume watcher\n")


Don't need the \n. Also can you change logs to start with uppercase letters?

adityadani · 2023-07-07T23:46:24Z

api/server/sdk/watcher.go

+ continue
+ }
+ resp := convertApiVolumeToSdkReponse(vol)
+ if err := stream.Send(resp); err != nil {


We need the timeoutCtx here too.

volume/volume.go

adityadani

lgtm

adityadani · 2023-07-11T16:28:06Z

api/server/sdk/watcher.go

@@ -40,7 +40,7 @@ func (w *WatcherServer) Watch(req *api.SdkWatchRequest, stream api.OpenStorageWa
 if req.GetVolumeEvent() != nil {
 return w.volumeWatch(req.GetVolumeEvent(), stream)
 }
- return status.Errorf(codes.InvalidArgument, "invalid request type for watcher %v", req)
+ return status.Errorf(codes.InvalidArgument, "Invalid request type for watcher %v", req)


nit: errors should still be lower case

Signed-off-by: dahuang <[email protected]>

approved in comment

Signed-off-by: dahuang <[email protected]>

* Adding Watch API and initial implementation - new API introduced in api.proto - new proto-driver function to get a volume watcher that passes "modified/new/deleted" volume through a golang channel - implement watcher go-routine that constantly wait for volume from the golang channel that passes from porx - implement the pipeline that passes volume info from watcher go-routine to all streaming clients Tests: UT passed Local smoke test with Porx Signed-off-by: dahuang <[email protected]>

github-actions bot added area/api area/sdk-server size/xl labels Jun 21, 2023

lpabon previously requested changes Jun 26, 2023

View reviewed changes

dahuang-purestorage requested review from lpabon and adityadani June 27, 2023 20:07

github-actions bot added the stage/in-review label Jun 27, 2023

dahuang-purestorage requested a review from piyush-nimbalkar June 27, 2023 20:07

dahuang-purestorage marked this pull request as ready for review June 28, 2023 00:44

dahuang-purestorage requested a review from ggriffiths as a code owner June 28, 2023 00:44

lpabon reviewed Jun 28, 2023

View reviewed changes

ggriffiths reviewed Jun 28, 2023

View reviewed changes

ggriffiths reviewed Jun 29, 2023

View reviewed changes

adityadani reviewed Jun 29, 2023

View reviewed changes

ggriffiths reviewed Jun 29, 2023

View reviewed changes

dahuang-purestorage commented Jun 30, 2023

View reviewed changes

volume/volume.go Show resolved Hide resolved

dahuang-purestorage commented Jun 30, 2023

View reviewed changes

dahuang-purestorage requested review from ggriffiths, lpabon and adityadani July 2, 2023 00:21

lpabon reviewed Jul 3, 2023

View reviewed changes

github-actions bot added the stage/in-review-stale label Jul 7, 2023

adityadani reviewed Jul 7, 2023

View reviewed changes

dahuang-purestorage requested a review from adityadani July 10, 2023 16:32

adityadani approved these changes Jul 11, 2023

View reviewed changes

ggriffiths approved these changes Jul 11, 2023

View reviewed changes

Adding Watch API and initial implementation

f6cf994

Signed-off-by: dahuang <[email protected]>

dahuang-purestorage added 22 commits July 11, 2023 14:08

Update versions

1c3a0a8

Signed-off-by: dahuang <[email protected]>

fixup! Update versions

3ce2450

Signed-off-by: dahuang <[email protected]>

update Makefile

a7b890d

Signed-off-by: dahuang <[email protected]>

remove volumegenreator in fake.go

bb48373

Signed-off-by: dahuang <[email protected]>

fix start watcher

f99999d

Signed-off-by: dahuang <[email protected]>

refactor

e4aba62

Signed-off-by: dahuang <[email protected]>

refactoring

c8b6597

Signed-off-by: dahuang <[email protected]>

store watchclient by eventType

e8141d3

Signed-off-by: dahuang <[email protected]>

add in lock and more channel handling

f6a80a1

Signed-off-by: dahuang <[email protected]>

fixup proto comments

1a3e4b5

Signed-off-by: dahuang <[email protected]>

fixup! fixup proto comments

435145d

Signed-off-by: dahuang <[email protected]>

fixup! fixup! fixup proto comments

694d3de

Signed-off-by: dahuang <[email protected]>

fixup comment and add timeout for send

21cf0b7

Signed-off-by: dahuang <[email protected]>

fixup! fixup comment and add timeout for send

c7148c8

Signed-off-by: dahuang <[email protected]>

fixup! fixup! fixup comment and add timeout for send

d54029d

Signed-off-by: dahuang <[email protected]>

check for unavailable in Watch

97c1a19

Signed-off-by: dahuang <[email protected]>

Add simple UT for Watch and use context

593609f

Signed-off-by: dahuang <[email protected]>

fixup! Add simple UT for Watch and use context

71b5f27

Signed-off-by: dahuang <[email protected]>

addresses comments

84fa530

Signed-off-by: dahuang <[email protected]>

fixup! addresses comments

783104f

Signed-off-by: dahuang <[email protected]>

lowercase error

557e15c

Signed-off-by: dahuang <[email protected]>

fixup! lowercase error

1824165

Signed-off-by: dahuang <[email protected]>

dahuang-purestorage force-pushed the PWX-29706-add-watch-volume-api branch from e7718ac to 1824165 Compare July 11, 2023 20:10

dahuang-purestorage added 2 commits July 11, 2023 16:06

fixup! fixup! lowercase error

57e27db

Signed-off-by: dahuang <[email protected]>

fixup! fixup! fixup! lowercase error

fb5af53

Signed-off-by: dahuang <[email protected]>

dahuang-purestorage merged commit 9b86347 into master Jul 12, 2023
2 checks passed

dahuang-purestorage deleted the PWX-29706-add-watch-volume-api branch July 12, 2023 18:03

		}


		// OpenStorageWatcher is a service used to stream events in the system

Adding Watch API and initial implementation #2281

Adding Watch API and initial implementation #2281

Conversation

dahuang-purestorage commented Jun 21, 2023 • edited Loading

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

ggriffiths Jun 28, 2023 • edited Loading

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

dahuang-purestorage Jun 30, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lpabon left a comment

Choose a reason for hiding this comment

github-actions bot commented Jul 7, 2023

adityadani 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

adityadani left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dahuang-purestorage commented Jun 21, 2023 •

edited

Loading

ggriffiths Jun 28, 2023 •

edited

Loading

dahuang-purestorage Jun 30, 2023 •

edited

Loading