Support gRPC streaming

[joshcarp]

Feature Description

Support gRPC streaming as described here

Suggested Solution (optional)

The grpc Client currently uses the grpc.Invoke command:
https://github.com/k6io/k6/blob/61f464b99a2d052347fee157eb68a024f5cf7430/js/modules/k6/grpc/client.go#L398

Instead of this, possibly using the return value of NewStream godoc:

https://github.com/grpc/grpc-go/blob/87eb5b7502493f758e76c4d09430c0049a81a557/stream.go#L144

This would look similar to the generated stream endpoint grpc code

func (c *routeGuideClient) ListFeatures(ctx context.Context, in *Rectangle, opts ...grpc.CallOption) (RouteGuide_ListFeaturesClient, error) {
	stream, err := c.cc.NewStream(ctx, &_RouteGuide_serviceDesc.Streams[0], "/main.RouteGuide/ListFeatures", opts...)
	if err != nil {
		return nil, err
	}
	x := &routeGuideListFeaturesClient{stream}
	if err := x.ClientStream.SendMsg(in); err != nil {
		return nil, err
	}
	if err := x.ClientStream.CloseSend(); err != nil {
		return nil, err
	}
	return x, nil
}

[na--]

Thanks for making this issue!

We'd like to have gRPC streaming, but at least the server streaming and bi-directional streaming parts are somewhat blocked by our lack of event loops in VUs (#882). For now, before we have this in the k6 core, someone should probably try to make an xk6 extension first...

[kelvinrfr]

Hi @na-- , I would like to raise the same question again. Apparently, if I get it correctly, the event loops in VUs are now supported and released based on PR #2228 ... (as reported here #2383 (comment))

Is there any glimpse of development of this feature?

[na--]

Yes! I can't give any promises for when exactly, depends on how we prioritize against a whole bunch of other things, but it's definitely on our short-term roadmap now that we've cleared out the obstacles.

Edit: I'll very optimistically add it to the v0.39.0 milestone, though I doubt we'll be able to get to it in the next ~6 weeks

[kelvinrfr]

@na-- Thank you for such a quick response! That is great and very optimistic! v0.39.0 is really near ❤️

To understand a little bit better, would be possible/how to help on the development of this feature? I mean, is there any definition already to how the streaming tests setup will be done (like the ones for unary calls here) or this is still something open for definition?

[na--]

Still TBD. We are still figuring out things when it comes to event loops (e.g. see this and this for some recent experiments) and we haven't started planning how the JS APIs for gRPC streaming should look like exactly. We'll probably research what other JS gRPC libraries use before we design them... 🤔

That said, if you want to work on this, go right ahead! As I said, I somewhat doubt we'll manage to get this time in time for v0.39.0, but a good external pull request might speed things up significantly! Just make sure to share your API proposal here for comments, before you've invested a significant amount of time implementing it, otherwise there might be a nasty surprise if we decide not to merge it because of some fundamental misalignment.

[kelvinrfr]

Sure! Thanks for the heads up!

[nsmith5]

👋🏻 Hey @na-- and @olegbespalov, are ya'll still looking for some help on this feature (per the help-wanted label)? I'm happy to help out how ever necessary as this feature would be super helpful to me!

[olegbespalov]

Hey @nsmith5 !

I planned to start work on that one, but for sure, community first 😉 So if you want to pick up the issue and work on it, please go ahead 🙌 ☺️

[na--]

@nsmith5, it's great that you want to work on this! As @olegbespalov said, we probably would have picked it up soon, but we'd love to merge your PR instead! 🎉

Here are some details:

1. Please take a look at the instructions in CONTRIBUTING.md
2. Before you start writing any code, please share the JS API you propose to implement. Having a tentative 👍 for the API proposal will make reviewing the code later much easier and increase our chances of actually merging it.
3. If you can, please let us know roughly when you expect to be able to work on this issue. If it's not going to be in the near future, we might prefer to implement it ourselves sooner.
4. We have a new strategy of how we develop and stabilize new JS APIs, see this part of the v0.39.0 release notes for details. We first try to first implement them as unstable xk6-extensions and iterate on the API, then integrate them in k6 as semi-stable k6/experimental/whatever built-in modules and finally release fully stable APIs as k6/whatever. In this case, since we already have parts of the gRPC API in k6, we should probably start with the new streaming parts in k6/experimental.
5. We still don't have good documentation on how to build asynchronous JS APIs with the new event loops, sorry. But maybe use https://github.com/grafana/xk6-timers and https://github.com/grafana/xk6-websockets as examples.

[nsmith5]

Yeah, I don't want to block on ya'll going ahead and implementing yourself. I might try to hack together something to try this soon, but I'm not sure on timescale and have some other priorities a launch that might interrupt. Thanks a bunch on all the pointers on where to learn more and get started. The async js bits in particular I wasn't sure about so I'll take a look at xk6-websockets for sure

[Dogacel]

I think the proposed API can look similar to WebSockets & node grpc streaming API (https://grpc.io/docs/languages/node/basics/#streaming-rpcs)

var call = client.echoStream(echoRequest);
call.on('data', function(data) {});
call.on('end', function() {});
call.on('error', function(e) {});
call.on('status', function(status) {});

So the k6 code looks as so:

const {responseCallback, requestWriter} = client.stream("/endpoint");

// Server streaming
responseCallback.on('data', (x) => check("x is odd", x % 2 == 0));

// Client streaming
[1,2,3].forEach(x => requestWriter.write(x));
responseCallback.next(); // Callback state is completed, consecutive calls to next are ignored

// Bidirectional streaming
responseCallback.on('data', (x) => check("x is odd", x % 2 == 0));
[1,2,3].forEach(x => requestWriter.write(x));

We will probably need utilities such as .wait() or on('timeout', ...) to capture specific metrics.

[olegbespalov]

Hi @nsmith5 !

It seems we are in a position to prioritize this task and start work on that, but since you were part of the discussion above, I'd like to sync with you. Are you okay if we implement this ourselves? ☺️

[olegbespalov]

So a minor update here and a proposal for the API. I've checked the nodejs eco-system and honestly agree with the @Dogacel that our GRPC's stream could look similar to the NodeJS.

There could be an example of the bidirectional stream, but basically, the result of the newStream will always be an object that has a set of methods that are either client (write, end) or server streams (on)

import grpc from 'k6/experimental/grpc';

const client = new grpc.Client();
client.load([], './proto/foo.proto');

export default () => {
   client.connect('localhost:4500', {
      plaintext: true
   });
   
   // a client opens a bidirectional stream
   // `null` as the second argument is the non-required message
   // and params are the k6's params object that can hold additional things like tags, metadata, and so on
   const stream = client.newStream('foo.BarService/someMethod', null, {tags: {a: 'b'}});

   stream.on('data', (message) => {
      // server sends data, processing...
   });

   stream.on('end', () => {
      // server has finished sending
   });

   stream.on('error', (err) => {
      // process the err
   });

   stream.on('status', (status) => {
      // process the status
   });

   // client writes something to the stream
   stream.write({message: 'foo'});
   stream.write({message: 'bar'});
   
   // client ends the stream
   stream.end();
  
   client.close();
};

What do you think?

[allansson]

A few thoughts on the proposed API.

import grpc from 'k6/experimental/grpc';

const client = new grpc.Client();

My suggestions would be to do:

import GrpcClient from 'k6/experimental/grpc' // Default export is the class constructor
import { Client } from 'k6/experimental/grpc' // Named exports for everything that the module exports
import * as grpc from 'k6/experimental/grpc' // Allow namespacing via wildcard imports

The reason NodeJS has the grpc namespace is that Node (mostly) uses require(i.e. CommonJS). With ES modules there's a lot more freedom.

---

client.load([], './proto/foo.proto');

If passing the client some definitions is a requirement, might as well offer an option to pass them in the constructor
(I'm not sure what the empty array does though, so the example is probably not accurate)

new Client({
  definitions: [
    './proto/foo.proto'
  ]
})

---

client.connect('localhost:4500', {
   plaintext: true
 });

client.close();

open + close or connect + disconnect?

---

const stream = client.newStream('foo.BarService/someMethod', metadata);
stream.end();

newStream feels weird to me when JS has the new keyword: new Stream(client).

An alternative is startStream/beginStream to mirror the end method.

[olegbespalov]

Hey @allansson , thanks for the input! ☺️

Sorry that I wasn't so clear, but the proposal is only about streaming. Things like client.connect or client.load are from the existing k6 GRPC API, which didn't want to touch to keep the scope only about streaming and try to deliver this feature in the closest possible release.

So it's the part:

// a client opens any type of  stream (client, server, bidirectional)
const stream = client.newStream('foo.BarService/someMethod');

// on, represents the interactions with the server stream

stream.on('data', (message) => {
   // server sends data, processing...
});

stream.on('end', () => {
   // server has finished sending
});

stream.on('error', (err) => {
   // process the err
});

stream.on('status', (status) => {
   // process the status
});

// write & end represent interactions with the client stream

// client writes something to the stream
stream.write({message: 'foo'});
stream.write({message: 'bar'});

// client ends the stream
stream.end();

However, even on our docs page, we claim that the current API is subject to change, so it's a perfect moment to reconsider the whole API. Still, I'm going to focus first of all on the streaming part. I'll define it as the stretch goal to implement your suggestions.

Regarding the actual streaming part.

newStream feels weird to me when JS has the new keyword: new Stream(client).

The new Stream looks a bit cleaner to me, but slightly concerned that we should always pass the same client as the same argument 🤔 On the other side, it could become a benefit like the new experimental GRPC could expose only this class Stream and the first argument could accept both types, "old" and a "new" clients.

// minimal version:
const stream = new Stream(client, 'foo.BarService/someMethod');

// Another
const message = {foo: "bar"};
const params = {
   tags: {
     a: "b",
   }
};
const stream = new Stream(client, 'foo.BarService/someMethod', message, params);

Also, just in case there is a simple chat https://github.com/olegbespalov/grpc-streaming-example that I will use as the example of the demo app to test the functionality.

[olegbespalov]

For the convenience extracted the proposed API as the separated messaged

import { Stream } from 'k6/experimental/grpc'

// ...
//initialization of the client, the same way how it currently does in the k6/net/grpc
// ..

// any message that you want to send to the server with the opening of a stream
const message = { message: 'hello world' }

// params represent the additional params that we could configure
// it could be the metadata, tags, and extended in the future
const params = {
  tags: {
    my_tag: 'my_value',
  },
}

// this constructs the stream
// Stream(client, method, [,request], [, params])
// - client - already initialized and connected client with the loaded definitions
//   ideally, it should be the same client that currently exists in k6/net/grpc
//   later, we could also accept the client that constructed using new GRPC api
// - method - a gRPC method URL to invoke.
// - request - a canonical request object, as per the Protobuf JSON Mapping.
// - params - an object with additional params that could be passed to the stream.
const stream = new Stream(
  client,
  'foo.BarService/sayHello',
  message,
  params
)

// an `on` method could be used to register the event handlers
// for the server streams

// a `data` event is triggered when the server sends the data
// you could register multiple handlers for the same event
stream.on('data', message => {
  // server send data, processing...
})

// an `end` event is triggered when the server has finished sending the data
// you could register multiple handlers for the same event
stream.on('end', () => {
  // server has finished sending
})

// an `error` event is triggered when an error occurs
// you could register multiple handlers for the same event
stream.on('error', (err) => {
  // process the err
})

// a `status` event is triggered when status of stream changes
// you could register multiple handlers for the same event
stream.on('status', (s) => {
  // process the status
})

// a `write` and `end` methods could be used for the client streams

// write writes data to the stream
stream.write({ message: 'foo' })
stream.write({ message: 'bar' })

// signal the server that the client has finished sending the data
stream.end()

[andreea-anghel]

Do you have a rough estimate of when this feature will be available? Thanks!

[olegbespalov]

Hi @andreea-anghel

This feature is prioritized to land in the following v0.44.0 release, which should happen in mid of April. For sure, it's a soft commitment, but we're doing our best.

[Dogacel]

@olegbespalov do you need any contribution regarding development?

[olegbespalov]

Hey @Dogacel ! Thanks for offering help ☺️ For now all good 👍

[m-guesnon-pvotal]

Hi!

I see the target milestone has been pushed. Is the milestone a target for GA release or experimental?

[olegbespalov]

Hi @m-guesnon-pvotal and the other topic-followers!

Yeah, it's worth updating here.

Unfortunately, the GRPC streaming won't be part of the v0.44.0 release. There were a few things that affected pace 😢

The whole GPRC logic is extracted as https://github.com/grafana/xk6-grpc. We're working to make GRPC streaming happen. So in the following weeks, it will be available as the k6's extension, so it will be possible to try it by building the custom binary. And it's going to be included in the k6 as an experimental module in v0.45.0.

I'm also going to make the status updates here once the things from above will happen.

Thanks for your patience and sorry for the delay.

[olegbespalov]

Here is a minor status update.

The PR with an experimental GRPC module is there #3107, and we're planning to merge it soon, making a new experimental GRPC module available in the k6 v0.45.

The new experimental module extends the current k6's GRPC module, that way that the end user could just replace the module import:

-import { Client } from 'k6/net/grpc';
+import { Client } from 'k6/experimental/grpc';

However, I must stress that this could also change in a feature since we have some discussions about improving the Client's API #2020 (comment)

We continue improving the extension, and hopefully, it could replace the existing k6 grpc module at some point.

The documentation of the API will be available soon (but definitely before the release), and for now, you can check the straightforward examples:

• https://github.com/grafana/k6/blob/0167ce449c38b745bf8e6448582b84fdb0972c63/examples/experimental/grpc/client_streaming.js
• https://github.com/grafana/k6/blob/0167ce449c38b745bf8e6448582b84fdb0972c63/examples/experimental/grpc/server_streaming.js

If you face any issues or have some improvements suggestions, please open the issue in the extension's repository: https://github.com/grafana/xk6-grpc/issues

[olegbespalov]

Hi there!

So, the experimental GRPC module with the streaming support has landed on the v0.45.0 k6's release I'm closing the issue.

However, I'd like to say that the work on improving GRPC and GRPC streaming, in particular, hasn't been entirely done. We continue to do that via working on the xk6-grpc extension, which at some point should replace an existing k6/net/grpc.

The documentation for the new module you can find here https://k6.io/docs/javascript-api/k6-experimental/grpc/

We're kindly asking you to give the try new experimental module and give us feedback throw the GitHub issues directly in the xk6-grpc repository or community forums.