From 965fcf9d6b7e748298476ffd5b4793ac63f7968d Mon Sep 17 00:00:00 2001 From: Evan Gilman Date: Tue, 10 May 2016 00:09:15 -0700 Subject: [PATCH] Enable Stale mode for watchers Solves https://github.com/hashicorp/consul/issues/917 by giving consul watch a `-stale` flag --- command/watch.go | 13 ++++++++++++- watch/funcs.go | 35 ++++++++++++++++++++++++++++++----- 2 files changed, 42 insertions(+), 6 deletions(-) diff --git a/command/watch.go b/command/watch.go index d5caf325a4..69939f72cf 100644 --- a/command/watch.go +++ b/command/watch.go @@ -47,6 +47,8 @@ Watch Specification: -prefix=val Specifies the key prefix to watch. Only for 'keyprefix' type. -service=val Specifies the service to watch. Required for 'service' type, optional for 'checks' type. + -stale=[true|false] Specefies if watch data is permitted to be stale. Defaults + false. -state=val Specifies the states to watch. Optional for 'checks' type. -tag=val Specifies the service tag to filter on. Optional for 'service' type. @@ -57,7 +59,7 @@ Watch Specification: } func (c *WatchCommand) Run(args []string) int { - var watchType, datacenter, token, key, prefix, service, tag, passingOnly, state, name string + var watchType, datacenter, token, key, prefix, service, tag, passingOnly, stale, state, name string cmdFlags := flag.NewFlagSet("watch", flag.ContinueOnError) cmdFlags.Usage = func() { c.Ui.Output(c.Help()) } cmdFlags.StringVar(&watchType, "type", "", "") @@ -68,6 +70,7 @@ func (c *WatchCommand) Run(args []string) int { cmdFlags.StringVar(&service, "service", "", "") cmdFlags.StringVar(&tag, "tag", "", "") cmdFlags.StringVar(&passingOnly, "passingonly", "", "") + cmdFlags.StringVar(&stale, "stale", "", "") cmdFlags.StringVar(&state, "state", "", "") cmdFlags.StringVar(&name, "name", "", "") httpAddr := HTTPAddrFlag(cmdFlags) @@ -109,6 +112,14 @@ func (c *WatchCommand) Run(args []string) int { if tag != "" { params["tag"] = tag } + if stale != "" { + b, err := strconv.ParseBool(stale) + if err != nil { + c.Ui.Error(fmt.Sprintf("Failed to parse stale flag: %s", err)) + return 1 + } + params["stale"] = b + } if state != "" { params["state"] = state } diff --git a/watch/funcs.go b/watch/funcs.go index 9308e7c633..874aba5558 100644 --- a/watch/funcs.go +++ b/watch/funcs.go @@ -35,9 +35,14 @@ func keyWatch(params map[string]interface{}) (WatchFunc, error) { return nil, fmt.Errorf("Must specify a single key to watch") } + stale := false + if err := assignValueBool(params, "stale", &stale); err != nil { + return nil, err + } + fn := func(p *WatchPlan) (uint64, interface{}, error) { kv := p.client.KV() - opts := consulapi.QueryOptions{WaitIndex: p.lastIndex} + opts := consulapi.QueryOptions{AllowStale: stale, WaitIndex: p.lastIndex} pair, meta, err := kv.Get(key, &opts) if err != nil { return 0, nil, err @@ -60,9 +65,14 @@ func keyPrefixWatch(params map[string]interface{}) (WatchFunc, error) { return nil, fmt.Errorf("Must specify a single prefix to watch") } + stale := false + if err := assignValueBool(params, "stale", &stale); err != nil { + return nil, err + } + fn := func(p *WatchPlan) (uint64, interface{}, error) { kv := p.client.KV() - opts := consulapi.QueryOptions{WaitIndex: p.lastIndex} + opts := consulapi.QueryOptions{AllowStale: stale, WaitIndex: p.lastIndex} pairs, meta, err := kv.List(prefix, &opts) if err != nil { return 0, nil, err @@ -74,9 +84,14 @@ func keyPrefixWatch(params map[string]interface{}) (WatchFunc, error) { // servicesWatch is used to watch the list of available services func servicesWatch(params map[string]interface{}) (WatchFunc, error) { + stale := false + if err := assignValueBool(params, "stale", &stale); err != nil { + return nil, err + } + fn := func(p *WatchPlan) (uint64, interface{}, error) { catalog := p.client.Catalog() - opts := consulapi.QueryOptions{WaitIndex: p.lastIndex} + opts := consulapi.QueryOptions{AllowStale: stale, WaitIndex: p.lastIndex} services, meta, err := catalog.Services(&opts) if err != nil { return 0, nil, err @@ -88,9 +103,14 @@ func servicesWatch(params map[string]interface{}) (WatchFunc, error) { // nodesWatch is used to watch the list of available nodes func nodesWatch(params map[string]interface{}) (WatchFunc, error) { + stale := false + if err := assignValueBool(params, "stale", &stale); err != nil { + return nil, err + } + fn := func(p *WatchPlan) (uint64, interface{}, error) { catalog := p.client.Catalog() - opts := consulapi.QueryOptions{WaitIndex: p.lastIndex} + opts := consulapi.QueryOptions{AllowStale: stale, WaitIndex: p.lastIndex} nodes, meta, err := catalog.Nodes(&opts) if err != nil { return 0, nil, err @@ -119,9 +139,14 @@ func serviceWatch(params map[string]interface{}) (WatchFunc, error) { return nil, err } + stale := false + if err := assignValueBool(params, "stale", &stale); err != nil { + return nil, err + } + fn := func(p *WatchPlan) (uint64, interface{}, error) { health := p.client.Health() - opts := consulapi.QueryOptions{WaitIndex: p.lastIndex} + opts := consulapi.QueryOptions{AllowStale: stale, WaitIndex: p.lastIndex} nodes, meta, err := health.Service(service, tag, passingOnly, &opts) if err != nil { return 0, nil, err