mirror of
https://github.com/status-im/consul.git
synced 2025-01-26 21:51:39 +00:00
streaming: set a default timeout
The blocking query backend sets the default value on the server side. The streaming backend does not using blocking queries, so we must set the timeout on the client.
This commit is contained in:
parent
8cfbc8e7c9
commit
242b3a2dc5
@ -395,6 +395,7 @@ func New(bd BaseDeps) (*Agent, error) {
|
||||
Logger: bd.Logger.Named("rpcclient.health"),
|
||||
},
|
||||
UseStreamingBackend: a.config.UseStreamingBackend,
|
||||
QueryOptionDefaults: config.ApplyDefaultQueryOptions(a.config),
|
||||
}
|
||||
|
||||
a.serviceManager = NewServiceManager(&a)
|
||||
|
@ -1930,3 +1930,16 @@ func isFloat(t reflect.Type) bool { return t.Kind() == reflect.Float32 || t.Kind
|
||||
func isComplex(t reflect.Type) bool {
|
||||
return t.Kind() == reflect.Complex64 || t.Kind() == reflect.Complex128
|
||||
}
|
||||
|
||||
// ApplyDefaultQueryOptions returns a function which will set default values on
|
||||
// the options based on the configuration. The RuntimeConfig must not be nil.
|
||||
func ApplyDefaultQueryOptions(config *RuntimeConfig) func(options *structs.QueryOptions) {
|
||||
return func(options *structs.QueryOptions) {
|
||||
switch {
|
||||
case options.MaxQueryTime > config.MaxQueryTime:
|
||||
options.MaxQueryTime = config.MaxQueryTime
|
||||
case options.MaxQueryTime == 0:
|
||||
options.MaxQueryTime = config.DefaultQueryTime
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -17,6 +17,7 @@ type Client struct {
|
||||
MaterializerDeps MaterializerDeps
|
||||
CacheName string
|
||||
UseStreamingBackend bool
|
||||
QueryOptionDefaults func(options *structs.QueryOptions)
|
||||
}
|
||||
|
||||
type NetRPC interface {
|
||||
@ -38,6 +39,8 @@ func (c *Client) ServiceNodes(
|
||||
req structs.ServiceSpecificRequest,
|
||||
) (structs.IndexedCheckServiceNodes, cache.ResultMeta, error) {
|
||||
if c.useStreaming(req) && (req.QueryOptions.UseCache || req.QueryOptions.MinQueryIndex > 0) {
|
||||
c.QueryOptionDefaults(&req.QueryOptions)
|
||||
|
||||
result, err := c.ViewStore.Get(ctx, c.newServiceRequest(req))
|
||||
if err != nil {
|
||||
return structs.IndexedCheckServiceNodes{}, cache.ResultMeta{}, err
|
||||
|
@ -3,10 +3,12 @@ package health
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/hashicorp/consul/agent/cache"
|
||||
"github.com/hashicorp/consul/agent/config"
|
||||
"github.com/hashicorp/consul/agent/structs"
|
||||
"github.com/hashicorp/consul/agent/submatview"
|
||||
)
|
||||
@ -25,6 +27,7 @@ func TestClient_ServiceNodes_BackendRouting(t *testing.T) {
|
||||
ViewStore: &fakeViewStore{},
|
||||
CacheName: "cache-no-streaming",
|
||||
UseStreamingBackend: true,
|
||||
QueryOptionDefaults: config.ApplyDefaultQueryOptions(&config.RuntimeConfig{}),
|
||||
}
|
||||
|
||||
_, _, err := c.ServiceNodes(context.Background(), tc.req)
|
||||
@ -233,3 +236,28 @@ func TestClient_Notify_BackendRouting(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_ServiceNodes_SetsDefaults(t *testing.T) {
|
||||
store := &fakeViewStore{}
|
||||
c := &Client{
|
||||
ViewStore: store,
|
||||
CacheName: "cache-no-streaming",
|
||||
UseStreamingBackend: true,
|
||||
QueryOptionDefaults: config.ApplyDefaultQueryOptions(&config.RuntimeConfig{
|
||||
MaxQueryTime: 200 * time.Second,
|
||||
DefaultQueryTime: 100 * time.Second,
|
||||
}),
|
||||
}
|
||||
|
||||
req := structs.ServiceSpecificRequest{
|
||||
Datacenter: "dc1",
|
||||
ServiceName: "web1",
|
||||
QueryOptions: structs.QueryOptions{MinQueryIndex: 22},
|
||||
}
|
||||
|
||||
_, _, err := c.ServiceNodes(context.Background(), req)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Len(t, store.calls, 1)
|
||||
require.Equal(t, 100*time.Second, store.calls[0].CacheInfo().Timeout)
|
||||
}
|
||||
|
Loading…
x
Reference in New Issue
Block a user