Skip to content

Commit

Permalink
Porting xk6-grpc into core
Browse files Browse the repository at this point in the history
  • Loading branch information
@olegbespalov
olegbespalov committed Dec 5, 2023
1 parent bfdc446 commit f36f517
Show file tree
Hide file tree
Showing 34 changed files with 3,191 additions and 2,111 deletions.
3 changes: 1 addition & 2 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@ require (
github.com/golang/protobuf v1.5.3
github.com/gorilla/websocket v1.5.0
github.com/grafana/xk6-browser v1.2.1
github.com/grafana/xk6-grpc v0.1.4-0.20230919144024-6ed5daf33509
github.com/grafana/xk6-output-prometheus-remote v0.3.1
github.com/grafana/xk6-redis v0.2.0
github.com/grafana/xk6-timers v0.1.2
Expand All @@ -30,6 +29,7 @@ require (
github.com/mccutchen/go-httpbin v1.1.2-0.20190116014521-c5cb2f4802fa
github.com/mstoykov/atlas v0.0.0-20220811071828-388f114305dd
github.com/mstoykov/envconfig v1.4.1-0.20220114105314-765c6d8c76f1
github.com/mstoykov/k6-taskqueue-lib v0.1.0
github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d
github.com/serenize/snaker v0.0.0-20201027110005-a7ad2135616e
github.com/sirupsen/logrus v1.9.3
Expand Down Expand Up @@ -77,7 +77,6 @@ require (
github.com/inconshreveable/mousetrap v1.0.0 // indirect
github.com/josharian/intern v1.0.0 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
github.com/mstoykov/k6-taskqueue-lib v0.1.0 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/prometheus/client_golang v1.16.0 // indirect
github.com/prometheus/client_model v0.4.0 // indirect
Expand Down
2 changes: 0 additions & 2 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,8 +95,6 @@ github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWm
github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
github.com/grafana/xk6-browser v1.2.1 h1:O2fuHHvmmhXvWTPXzD+jsnt1XkVgVjx0+Lj1hsGIWMM=
github.com/grafana/xk6-browser v1.2.1/go.mod h1:D3k9/MQHnNKfyzU3fh32pHlrh3GY2LAlkY4wYt/Vn4Y=
github.com/grafana/xk6-grpc v0.1.4-0.20230919144024-6ed5daf33509 h1:9ujE4S5cA3WDhRJnwNuUDtfk3w9FeWx6PaZ+lb3o46M=
github.com/grafana/xk6-grpc v0.1.4-0.20230919144024-6ed5daf33509/go.mod h1:sFTwAsHAtp2f1PNiq0wPjJ7HrAIKploI7Y5mOYo+zIQ=
github.com/grafana/xk6-output-prometheus-remote v0.3.1 h1:X23rQzlJD8dXWB31DkxR4uPnuRFo8L0Y0H22fSG9xl0=
github.com/grafana/xk6-output-prometheus-remote v0.3.1/go.mod h1:0JLAm4ONsNUlNoxJXAwOCfA6GtDwTPs557OplAvE+3o=
github.com/grafana/xk6-redis v0.2.0 h1:iXmAKVlAxafZ/h8ptuXTFhGu63IFsyDI8QjUgWm66BU=
Expand Down
3 changes: 1 addition & 2 deletions js/jsmodules.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@ import (
"go.k6.io/k6/js/modules/k6/ws"

"github.com/grafana/xk6-browser/browser"
expGrpc "github.com/grafana/xk6-grpc/grpc"
"github.com/grafana/xk6-redis/redis"
"github.com/grafana/xk6-timers/timers"
"github.com/grafana/xk6-webcrypto/webcrypto"
Expand All @@ -35,7 +34,7 @@ func getInternalJSModules() map[string]interface{} {
"k6/experimental/redis": redis.New(),
"k6/experimental/webcrypto": webcrypto.New(),
"k6/experimental/websockets": &expws.RootModule{},
"k6/experimental/grpc": expGrpc.New(),
"k6/experimental/grpc": grpc.New(), // TODO: make warning
"k6/experimental/timers": timers.New(),
"k6/experimental/tracing": tracing.New(),
"k6/experimental/browser": browser.New(),
Expand Down
249 changes: 37 additions & 212 deletions js/modules/k6/grpc/client.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,6 @@ import (
"go.k6.io/k6/js/common"
"go.k6.io/k6/js/modules"
"go.k6.io/k6/lib/netext/grpcext"
"go.k6.io/k6/lib/types"
"go.k6.io/k6/metrics"

"github.com/dop251/goja"
"github.com/jhump/protoreflect/desc"
Expand Down Expand Up @@ -213,7 +211,7 @@ func (c *Client) Connect(addr string, params goja.Value) (bool, error) {
return false, common.NewInitContextError("connecting to a gRPC server in the init context is not supported")
}

p, err := newConnectParams(c.vu.Runtime(), params)
p, err := newConnectParams(c.vu, params)
if err != nil {
return false, fmt.Errorf("invalid grpc.connect() parameters: %w", err)
}
Expand Down Expand Up @@ -299,9 +297,14 @@ func (c *Client) Invoke(
return nil, fmt.Errorf("method %q not found in file descriptors", method)
}

p, err := c.parseInvokeParams(params)
p, err := newCallParams(c.vu, params)
if err != nil {
return nil, fmt.Errorf("invalid grpc.invoke() parameters: %w", err)
return nil, fmt.Errorf("invalid GRPC's client.invoke() parameters: %w", err)
}

// k6 GRPC Invoke's default timeout is 2 minutes
if p.Timeout == time.Duration(0) {
p.Timeout = 2 * time.Minute
}

if req == nil {
Expand All @@ -315,17 +318,7 @@ func (c *Client) Invoke(
ctx, cancel := context.WithTimeout(c.vu.Context(), p.Timeout)
defer cancel()

if state.Options.SystemTags.Has(metrics.TagURL) {
p.TagsAndMeta.SetSystemTagOrMeta(metrics.TagURL, fmt.Sprintf("%s%s", c.addr, method))
}
parts := strings.Split(method[1:], "/")
p.TagsAndMeta.SetSystemTagOrMetaIfEnabled(state.Options.SystemTags, metrics.TagService, parts[0])
p.TagsAndMeta.SetSystemTagOrMetaIfEnabled(state.Options.SystemTags, metrics.TagMethod, parts[1])

// Only set the name system tag if the user didn't explicitly set it beforehand
if _, ok := p.TagsAndMeta.Tags.Get("name"); !ok {
p.TagsAndMeta.SetSystemTagOrMetaIfEnabled(state.Options.SystemTags, metrics.TagName, method)
}
p.SetSystemTags(state, c.addr, method)

reqmsg := grpcext.Request{
MethodDescriptor: methodDesc,
Expand Down Expand Up @@ -394,6 +387,7 @@ func (c *Client) convertToMethodInfo(fdset *descriptorpb.FileDescriptorSet) ([]M
appendMethodInfo(fd, sd, md)
}
}

messages := fd.Messages()

stack := make([]protoreflect.MessageDescriptor, 0, messages.Len())
Expand Down Expand Up @@ -427,218 +421,49 @@ func (c *Client) convertToMethodInfo(fdset *descriptorpb.FileDescriptorSet) ([]M
return rtn, nil
}

type invokeParams struct {
Metadata metadata.MD
TagsAndMeta metrics.TagsAndMeta
Timeout time.Duration
}

func (c *Client) parseInvokeParams(paramsVal goja.Value) (*invokeParams, error) {
result := &invokeParams{
Timeout: 1 * time.Minute,
TagsAndMeta: c.vu.State().Tags.GetCurrentValues(),
Metadata: metadata.New(nil),
}
if paramsVal == nil || goja.IsUndefined(paramsVal) || goja.IsNull(paramsVal) {
return result, nil
}
rt := c.vu.Runtime()
params := paramsVal.ToObject(rt)
for _, k := range params.Keys() {
switch k {
case "metadata":
md, err := newMetadata(params.Get(k))
if err != nil {
return result, fmt.Errorf("invalid metadata param: %w", err)
}

result.Metadata = md
case "tags":
if err := common.ApplyCustomUserTags(rt, &result.TagsAndMeta, params.Get(k)); err != nil {
return result, fmt.Errorf("metric tags: %w", err)
}
case "timeout":
var err error
v := params.Get(k).Export()
result.Timeout, err = types.GetDurationValue(v)
if err != nil {
return result, fmt.Errorf("invalid timeout value: %w", err)
}
case "headers":
return result, errors.New("headers param is not supported anymore. Please, use metadata param instead")
default:
return result, fmt.Errorf("unknown param: %q", k)
}
}
return result, nil
}

// newMetadata constructs a metadata.MD from the input value.
func newMetadata(input goja.Value) (metadata.MD, error) {
md := metadata.New(nil)

if common.IsNullish(input) {
return md, nil
}

v := input.Export()
func walkFileDescriptors(seen map[string]struct{}, fd *desc.FileDescriptor) []*descriptorpb.FileDescriptorProto {
fds := []*descriptorpb.FileDescriptorProto{}

raw, ok := v.(map[string]interface{})
if !ok {
return md, errors.New("must be an object with key-value pairs")
if _, ok := seen[fd.GetName()]; ok {
return fds
}
seen[fd.GetName()] = struct{}{}
fds = append(fds, fd.AsFileDescriptorProto())

for hk, kv := range raw {
var val string
// The gRPC spec defines that Binary-valued keys end in -bin
// https://grpc.io/docs/what-is-grpc/core-concepts/#metadata
if strings.HasSuffix(hk, "-bin") {
var binVal []byte
if binVal, ok = kv.([]byte); !ok {
return md, fmt.Errorf("%q value must be binary", hk)
}

// https://github.com/grpc/grpc-go/blob/v1.57.0/Documentation/grpc-metadata.md#storing-binary-data-in-metadata
val = string(binVal)
} else if val, ok = kv.(string); !ok {
return md, fmt.Errorf("%q value must be a string", hk)
}

md.Append(hk, val)
for _, dep := range fd.GetDependencies() {
deps := walkFileDescriptors(seen, dep)
fds = append(fds, deps...)
}

return md, nil
}

type connectParams struct {
IsPlaintext bool
ReflectionMetadata metadata.MD
UseReflectionProtocol bool
Timeout time.Duration
MaxReceiveSize int64
MaxSendSize int64
TLS map[string]interface{}
return fds
}

func newConnectParams(rt *goja.Runtime, input goja.Value) (connectParams, error) { //nolint:funlen,gocognit,cyclop
params := connectParams{
IsPlaintext: false,
UseReflectionProtocol: false,
ReflectionMetadata: metadata.New(nil),
Timeout: time.Minute,
MaxReceiveSize: 0,
MaxSendSize: 0,
// sanitizeMethodName
func sanitizeMethodName(name string) string {
if name == "" {
return name
}

if common.IsNullish(input) {
return params, nil
if !strings.HasPrefix(name, "/") {
name = "/" + name
}

raw := input.ToObject(rt)

for _, k := range raw.Keys() {
v := raw.Get(k).Export()

switch k {
case "plaintext":
var ok bool
params.IsPlaintext, ok = v.(bool)
if !ok {
return params, fmt.Errorf("invalid plaintext value: '%#v', it needs to be boolean", v)
}
case "timeout":
var err error
params.Timeout, err = types.GetDurationValue(v)
if err != nil {
return params, fmt.Errorf("invalid timeout value: %w", err)
}
case "reflect":
var ok bool
params.UseReflectionProtocol, ok = v.(bool)
if !ok {
return params, fmt.Errorf("invalid reflect value: '%#v', it needs to be boolean", v)
}
case "reflectMetadata":
md, err := newMetadata(raw.Get(k))
if err != nil {
return params, fmt.Errorf("invalid reflectMetadata param: %w", err)
}
params.ReflectionMetadata = md
case "maxReceiveSize":
var ok bool
params.MaxReceiveSize, ok = v.(int64)
if !ok {
return params, fmt.Errorf("invalid maxReceiveSize value: '%#v', it needs to be an integer", v)
}
if params.MaxReceiveSize < 0 {
return params, fmt.Errorf("invalid maxReceiveSize value: '%#v, it needs to be a positive integer", v)
}
case "maxSendSize":
var ok bool
params.MaxSendSize, ok = v.(int64)
if !ok {
return params, fmt.Errorf("invalid maxSendSize value: '%#v', it needs to be an integer", v)
}
if params.MaxSendSize < 0 {
return params, fmt.Errorf("invalid maxSendSize value: '%#v, it needs to be a positive integer", v)
}
case "tls":
var ok bool
params.TLS, ok = v.(map[string]interface{})

if !ok {
return params, fmt.Errorf("invalid tls value: '%#v', expected (optional) keys: cert, key, password, and cacerts", v)
}
// optional map keys below
if cert, certok := params.TLS["cert"]; certok {
if _, ok = cert.(string); !ok {
return params, fmt.Errorf("invalid tls cert value: '%#v', it needs to be a PEM formatted string", v)
}
}
if key, keyok := params.TLS["key"]; keyok {
if _, ok = key.(string); !ok {
return params, fmt.Errorf("invalid tls key value: '%#v', it needs to be a PEM formatted string", v)
}
}
if pass, passok := params.TLS["password"]; passok {
if _, ok = pass.(string); !ok {
return params, fmt.Errorf("invalid tls password value: '%#v', it needs to be a string", v)
}
}
if cacerts, cacertsok := params.TLS["cacerts"]; cacertsok {
var cacertsArray []interface{}
if cacertsArray, ok = cacerts.([]interface{}); ok {
for _, cacertsArrayEntry := range cacertsArray {
if _, ok = cacertsArrayEntry.(string); !ok {
return params, fmt.Errorf("invalid tls cacerts value: '%#v',"+
" it needs to be a string or an array of PEM formatted strings", v)
}
}
} else if _, ok = cacerts.(string); !ok {
return params, fmt.Errorf("invalid tls cacerts value: '%#v',"+
" it needs to be a string or an array of PEM formatted strings", v)
}
}
default:
return params, fmt.Errorf("unknown connect param: %q", k)
}
}
return params, nil
return name
}

func walkFileDescriptors(seen map[string]struct{}, fd *desc.FileDescriptor) []*descriptorpb.FileDescriptorProto {
fds := []*descriptorpb.FileDescriptorProto{}
// getMethodDescriptor sanitize it, and gets GRPC method descriptor or an error if not found
func (c *Client) getMethodDescriptor(method string) (protoreflect.MethodDescriptor, error) {
method = sanitizeMethodName(method)

if _, ok := seen[fd.GetName()]; ok {
return fds
if method == "" {
return nil, errors.New("method to invoke cannot be empty")
}
seen[fd.GetName()] = struct{}{}
fds = append(fds, fd.AsFileDescriptorProto())

for _, dep := range fd.GetDependencies() {
deps := walkFileDescriptors(seen, dep)
fds = append(fds, deps...)
methodDesc := c.mds[method]

if methodDesc == nil {
return nil, fmt.Errorf("method %q not found in file descriptors", method)
}

return fds
return methodDesc, nil
}
Loading

0 comments on commit f36f517

Please sign in to comment.