Merge pull request #2264 from hashicorp/pr-2045-slackpad

Enables stale mode for watchers.

command/watch.go

@@ -37,6 +37,8 @@ Options:
   -http-addr=127.0.0.1:8500  HTTP address of the Consul agent.
   -datacenter=""             Datacenter to query. Defaults to that of agent.
   -token=""                  ACL token to use. Defaults to that of agent.
+  -stale=[true|false]        Specifies if watch data is permitted to be stale.
+                             Defaults to false.
 
 Watch Specification:
 
@@ -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
 	}

watch/funcs.go

@@ -27,6 +27,11 @@ func init() {
 
 // keyWatch is used to return a key watching function
 func keyWatch(params map[string]interface{}) (WatchFunc, error) {
+	stale := false
+	if err := assignValueBool(params, "stale", &stale); err != nil {
+		return nil, err
+	}
+
 	var key string
 	if err := assignValue(params, "key", &key); err != nil {
 		return nil, err
@@ -34,10 +39,9 @@ func keyWatch(params map[string]interface{}) (WatchFunc, error) {
 	if key == "" {
 		return nil, fmt.Errorf("Must specify a single key to watch")
 	}
-
 	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
@@ -52,6 +56,11 @@ func keyWatch(params map[string]interface{}) (WatchFunc, error) {
 
 // keyPrefixWatch is used to return a key prefix watching function
 func keyPrefixWatch(params map[string]interface{}) (WatchFunc, error) {
+	stale := false
+	if err := assignValueBool(params, "stale", &stale); err != nil {
+		return nil, err
+	}
+
 	var prefix string
 	if err := assignValue(params, "prefix", &prefix); err != nil {
 		return nil, err
@@ -59,10 +68,9 @@ func keyPrefixWatch(params map[string]interface{}) (WatchFunc, error) {
 	if prefix == "" {
 		return nil, fmt.Errorf("Must specify a single prefix to watch")
 	}
-
 	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 +82,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 +101,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
@@ -102,6 +120,11 @@ func nodesWatch(params map[string]interface{}) (WatchFunc, error) {
 
 // serviceWatch is used to watch a specific service for changes
 func serviceWatch(params map[string]interface{}) (WatchFunc, error) {
+	stale := false
+	if err := assignValueBool(params, "stale", &stale); err != nil {
+		return nil, err
+	}
+
 	var service, tag string
 	if err := assignValue(params, "service", &service); err != nil {
 		return nil, err
@@ -121,7 +144,7 @@ func serviceWatch(params map[string]interface{}) (WatchFunc, error) {
 
 	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
@@ -133,6 +156,11 @@ func serviceWatch(params map[string]interface{}) (WatchFunc, error) {
 
 // checksWatch is used to watch a specific checks in a given state
 func checksWatch(params map[string]interface{}) (WatchFunc, error) {
+	stale := false
+	if err := assignValueBool(params, "stale", &stale); err != nil {
+		return nil, err
+	}
+
 	var service, state string
 	if err := assignValue(params, "service", &service); err != nil {
 		return nil, err
@@ -149,7 +177,7 @@ func checksWatch(params map[string]interface{}) (WatchFunc, error) {
 
 	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}
 		var checks []*consulapi.HealthCheck
 		var meta *consulapi.QueryMeta
 		var err error
@@ -168,6 +196,8 @@ func checksWatch(params map[string]interface{}) (WatchFunc, error) {
 
 // eventWatch is used to watch for events, optionally filtering on name
 func eventWatch(params map[string]interface{}) (WatchFunc, error) {
+	// The stale setting doesn't apply to events.
+
 	var name string
 	if err := assignValue(params, "name", &name); err != nil {
 		return nil, err

website/source/docs/commands/watch.html.markdown

@@ -37,12 +37,15 @@ The list of available flags are:
 
 * `-token` - ACL token to use. Defaults to that of agent.
 
+* `-stale=[true|false]` - Specifies if watch data is permitted to be stale. Defaults
+  to false.
+
 * `-key` - Key to watch. Only for `key` type.
 
 * `-name`- Event name to watch. Only for `event` type.
 
-* `-passingonly=[true|false]` - Should only passing entries be returned. Default false.
-  only for `service` type.
+* `-passingonly=[true|false]` - Should only passing entries be returned. Defaults to
+   false and only applies for `service` type.
 
 * `-prefix` - Key prefix to watch. Only for `keyprefix` type.