luxolus
/
open-consul
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

watch: Remove DSL in place of JSON

This commit is contained in:
Armon Dadgar 2014-08-21 11:38:44 -07:00
parent d47c60daf4
commit 47cb44683d
6 changed files with 59 additions and 239 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
watch/funcs.go
View File

@ -2,14 +2,13 @@ package watch
import (
"fmt"
"strconv"
"github.com/armon/consul-api"
)
// watchFactory is a function that can create a new WatchFunc
// from a parameter configuration
type watchFactory func(params map[string][]string) (WatchFunc, error)
type watchFactory func(params map[string]interface{}) (WatchFunc, error)
// watchFuncFactory maps each type to a factory function
var watchFuncFactory map[string]watchFactory
@ -26,7 +25,7 @@ func init() {
}
// keyWatch is used to return a key watching function
func keyWatch(params map[string][]string) (WatchFunc, error) {
func keyWatch(params map[string]interface{}) (WatchFunc, error) {
var key string
if err := assignValue(params, "key", &key); err != nil {
return nil, err
@ -51,7 +50,7 @@ func keyWatch(params map[string][]string) (WatchFunc, error) {
}
// keyPrefixWatch is used to return a key prefix watching function
func keyPrefixWatch(params map[string][]string) (WatchFunc, error) {
func keyPrefixWatch(params map[string]interface{}) (WatchFunc, error) {
var prefix string
if err := assignValue(params, "prefix", &prefix); err != nil {
return nil, err
@ -73,7 +72,7 @@ func keyPrefixWatch(params map[string][]string) (WatchFunc, error) {
}
// servicesWatch is used to watch the list of available services
func servicesWatch(params map[string][]string) (WatchFunc, error) {
func servicesWatch(params map[string]interface{}) (WatchFunc, error) {
fn := func(p *WatchPlan) (uint64, interface{}, error) {
catalog := p.client.Catalog()
opts := consulapi.QueryOptions{WaitIndex: p.lastIndex}
@ -87,7 +86,7 @@ func servicesWatch(params map[string][]string) (WatchFunc, error) {
}
// nodesWatch is used to watch the list of available nodes
func nodesWatch(params map[string][]string) (WatchFunc, error) {
func nodesWatch(params map[string]interface{}) (WatchFunc, error) {
fn := func(p *WatchPlan) (uint64, interface{}, error) {
catalog := p.client.Catalog()
opts := consulapi.QueryOptions{WaitIndex: p.lastIndex}
@ -101,8 +100,8 @@ func nodesWatch(params map[string][]string) (WatchFunc, error) {
}
// serviceWatch is used to watch a specific service for changes
func serviceWatch(params map[string][]string) (WatchFunc, error) {
var service, tag, passingRaw string
func serviceWatch(params map[string]interface{}) (WatchFunc, error) {
var service, tag string
if err := assignValue(params, "service", &service); err != nil {
return nil, err
}
@ -114,16 +113,9 @@ func serviceWatch(params map[string][]string) (WatchFunc, error) {
return nil, err
}
if err := assignValue(params, "passingonly", &passingRaw); err != nil {
return nil, err
}
passingOnly := false
if passingRaw != "" {
b, err := strconv.ParseBool(passingRaw)
if err != nil {
return nil, fmt.Errorf("Failed to parse passingonly value: %v", err)
}
passingOnly = b
if err := assignValueBool(params, "passingonly", &passingOnly); err != nil {
return nil, err
}
fn := func(p *WatchPlan) (uint64, interface{}, error) {
@ -139,7 +131,7 @@ func serviceWatch(params map[string][]string) (WatchFunc, error) {
}
// checksWatch is used to watch a specific checks in a given state
func checksWatch(params map[string][]string) (WatchFunc, error) {
func checksWatch(params map[string]interface{}) (WatchFunc, error) {
var service, state string
if err := assignValue(params, "service", &service); err != nil {
return nil, err

14
watch/funcs_test.go
View File

@ -18,7 +18,7 @@ func TestKeyWatch(t *testing.T) {
if consulAddr == "" {
t.Skip()
}
plan := mustParse(t, "type:key key:foo/bar/baz")
plan := mustParse(t, `{"type":"key", "key":"foo/bar/baz"}`)
invoke := 0
plan.Handler = func(idx uint64, raw interface{}) {
if invoke == 0 {
@ -72,7 +72,7 @@ func TestKeyPrefixWatch(t *testing.T) {
if consulAddr == "" {
t.Skip()
}
plan := mustParse(t, "type:keyprefix prefix:foo/")
plan := mustParse(t, `{"type":"keyprefix", "prefix":"foo/"}`)
invoke := 0
plan.Handler = func(idx uint64, raw interface{}) {
if invoke == 0 {
@ -128,7 +128,7 @@ func TestServicesWatch(t *testing.T) {
if consulAddr == "" {
t.Skip()
}
plan := mustParse(t, "type:services")
plan := mustParse(t, `{"type":"services"}`)
invoke := 0
plan.Handler = func(idx uint64, raw interface{}) {
if invoke == 0 {
@ -171,7 +171,7 @@ func TestNodesWatch(t *testing.T) {
if consulAddr == "" {
t.Skip()
}
plan := mustParse(t, "type:nodes")
plan := mustParse(t, `{"type":"nodes"}`)
invoke := 0
plan.Handler = func(idx uint64, raw interface{}) {
if invoke == 0 {
@ -220,7 +220,7 @@ func TestServiceWatch(t *testing.T) {
if consulAddr == "" {
t.Skip()
}
plan := mustParse(t, "type:service service:foo tag:bar passingonly:true")
plan := mustParse(t, `{"type":"service", "service":"foo", "tag":"bar", "passingonly":true}`)
invoke := 0
plan.Handler = func(idx uint64, raw interface{}) {
if invoke == 0 {
@ -269,7 +269,7 @@ func TestChecksWatch_State(t *testing.T) {
if consulAddr == "" {
t.Skip()
}
plan := mustParse(t, "type:checks state:warning")
plan := mustParse(t, `{"type":"checks", "state":"warning"}`)
invoke := 0
plan.Handler = func(idx uint64, raw interface{}) {
if invoke == 0 {
@ -329,7 +329,7 @@ func TestChecksWatch_Service(t *testing.T) {
if consulAddr == "" {
t.Skip()
}
plan := mustParse(t, "type:checks service:foobar")
plan := mustParse(t, `{"type":"checks", "service":"foobar"}`)
invoke := 0
plan.Handler = func(idx uint64, raw interface{}) {
if invoke == 0 {

2
watch/plan.go
View File

@ -47,7 +47,7 @@ OUTER:
// Handle an error in the watch function
if err != nil {
log.Printf("consul.watch: Watch '%s' errored: %v", p.Query, err)
log.Printf("consul.watch: Watch (type: %s) errored: %v", p.Type, err)
// Perform an exponential backoff
failures++

7
watch/plan_test.go
View File

@ -9,7 +9,7 @@ func init() {
watchFuncFactory["noop"] = noopWatch
}
func noopWatch(params map[string][]string) (WatchFunc, error) {
func noopWatch(params map[string]interface{}) (WatchFunc, error) {
fn := func(p *WatchPlan) (uint64, interface{}, error) {
idx := p.lastIndex + 1
return idx, idx, nil
@ -18,7 +18,8 @@ func noopWatch(params map[string][]string) (WatchFunc, error) {
}
func mustParse(t *testing.T, q string) *WatchPlan {
plan, err := Parse(q)
params := makeParams(t, q)
plan, err := Parse(params)
if err != nil {
t.Fatalf("err: %v", err)
}
@ -26,7 +27,7 @@ func mustParse(t *testing.T, q string) *WatchPlan {
}
func TestRun_Stop(t *testing.T) {
plan := mustParse(t, "type:noop")
plan := mustParse(t, `{"type":"noop"}`)
var expect uint64 = 1
plan.Handler = func(idx uint64, val interface{}) {
if idx != expect {

143
watch/watch.go
View File

@ -2,7 +2,6 @@ package watch
import (
"fmt"
"strings"
"sync"
"github.com/armon/consul-api"
@ -13,11 +12,10 @@ import (
// This view is watched for changes and a handler is invoked to take any
// appropriate actions.
type WatchPlan struct {
Query string
Datacenter string
Token string
Type string
Exempt map[string][]string
Exempt map[string]interface{}
Func WatchFunc
Handler HandlerFunc
@ -38,20 +36,14 @@ type WatchFunc func(*WatchPlan) (uint64, interface{}, error)
type HandlerFunc func(uint64, interface{})
// Parse takes a watch query and compiles it into a WatchPlan or an error
func Parse(query string) (*WatchPlan, error) {
return ParseExempt(query, nil)
func Parse(params map[string]interface{}) (*WatchPlan, error) {
return ParseExempt(params, nil)
}
// ParseExempt takes a watch query and compiles it into a WatchPlan or an error
// Any exempt parameters are stored in the Exempt map
func ParseExempt(query string, exempt []string) (*WatchPlan, error) {
tokens, err := tokenize(query)
if err != nil {
return nil, fmt.Errorf("Failed to parse: %v", err)
}
params := collapse(tokens)
func ParseExempt(params map[string]interface{}, exempt []string) (*WatchPlan, error) {
plan := &WatchPlan{
Query: query,
stopCh: make(chan struct{}),
}
@ -86,7 +78,7 @@ func ParseExempt(query string, exempt []string) (*WatchPlan, error) {
// Remove the exempt parameters
if len(exempt) > 0 {
plan.Exempt = make(map[string][]string)
plan.Exempt = make(map[string]interface{})
for _, ex := range exempt {
val, ok := params[ex]
if ok {
@ -107,121 +99,28 @@ func ParseExempt(query string, exempt []string) (*WatchPlan, error) {
return plan, nil
}
// assignValue is used to extract a value ensuring it is only
// defined once
func assignValue(params map[string][]string, name string, out *string) error {
if vals, ok := params[name]; ok {
if len(vals) != 1 {
return fmt.Errorf("Multiple definitions of %s", name)
// assignValue is used to extract a value ensuring it is a string
func assignValue(params map[string]interface{}, name string, out *string) error {
if raw, ok := params[name]; ok {
val, ok := raw.(string)
if !ok {
return fmt.Errorf("Expecting %s to be a string")
}
*out = vals[0]
*out = val
delete(params, name)
}
return nil
}
// token is used to represent a "datacenter:foobar" pair, where
// datacenter is the param and foobar is the value
type token struct {
param string
val string
}
func (t *token) GoString() string {
return fmt.Sprintf("%#v", *t)
}
// tokenize splits a query string into a slice of tokens
func tokenize(query string) ([]*token, error) {
var tokens []*token
for i := 0; i < len(query); i++ {
char := query[i]
// Ignore whitespace
if char == ' ' || char == '\t' || char == '\n' {
continue
// assignValueBool is used to extract a value ensuring it is a bool
func assignValueBool(params map[string]interface{}, name string, out *bool) error {
if raw, ok := params[name]; ok {
val, ok := raw.(bool)
if !ok {
return fmt.Errorf("Expecting %s to be a boolean")
}
// Read the next token
next, offset, err := readToken(query[i:])
if err != nil {
return nil, err
}
// Store the token
tokens = append(tokens, next)
// Increment the offset
i += offset
*out = val
delete(params, name)
}
return tokens, nil
}
// readToken is used to read a single token
func readToken(query string) (*token, int, error) {
// Get the token
param, offset, err := readParameter(query)
if err != nil {
return nil, 0, err
}
// Get the value
query = query[offset:]
val, offset2, err := readValue(query)
if err != nil {
return nil, 0, err
}
// Return the new token
token := &token{
param: param,
val: val,
}
return token, offset + offset2, nil
}
// readParameter scans for the next parameter
func readParameter(query string) (string, int, error) {
for i := 0; i < len(query); i++ {
char := query[i]
if char == ':' {
if i == 0 {
return "", 0, fmt.Errorf("Missing parameter name")
} else {
return query[:i], i + 1, nil
}
}
}
return "", 0, fmt.Errorf("Parameter delimiter not found")
}
// readValue is used to scan for the next value
func readValue(query string) (string, int, error) {
// Handle quoted values
if query[0] == '\'' || query[0] == '"' {
quoteChar := query[0:1]
endChar := strings.Index(query[1:], quoteChar)
if endChar == -1 {
return "", 0, fmt.Errorf("Missing end of quotation")
}
return query[1 : endChar+1], endChar + 2, nil
}
// Look for white space
endChar := strings.IndexAny(query, " \t\n")
if endChar == -1 {
return query, len(query), nil
}
return query[:endChar], endChar, nil
}
// collapse is used to collapse a token stream into a map
// of parameter name to list of values.
func collapse(tokens []*token) map[string][]string {
out := make(map[string][]string)
for _, t := range tokens {
existing := out[t.param]
out[t.param] = append(existing, t.val)
}
return out
return nil
}

104
watch/watch_test.go
View File

@ -1,96 +1,14 @@
package watch
import (
"fmt"
"reflect"
"bytes"
"encoding/json"
"testing"
)
func TestTokenize(t *testing.T) {
type tcase struct {
in string
out []*token
err error
}
cases := []tcase{
tcase{
"",
nil,
nil,
},
tcase{
"foo:bar bar:baz zip:zap",
[]*token{
&token{"foo", "bar"},
&token{"bar", "baz"},
&token{"zip", "zap"},
},
nil,
},
tcase{
"foo:\"long input here\" after:this",
[]*token{
&token{"foo", "long input here"},
&token{"after", "this"},
},
nil,
},
tcase{
"foo:'long input here' after:this",
[]*token{
&token{"foo", "long input here"},
&token{"after", "this"},
},
nil,
},
tcase{
"foo:'long input here after:this",
nil,
fmt.Errorf("Missing end of quotation"),
},
tcase{
"foo",
nil,
fmt.Errorf("Parameter delimiter not found"),
},
tcase{
":val",
nil,
fmt.Errorf("Missing parameter name"),
},
}
for _, tc := range cases {
tokens, err := tokenize(tc.in)
if err != nil && tc.err == nil {
t.Fatalf("%s: err: %v", tc.in, err)
} else if tc.err != nil && (err == nil || err.Error() != tc.err.Error()) {
t.Fatalf("%s: bad err: %v", tc.in, err)
}
if !reflect.DeepEqual(tokens, tc.out) {
t.Fatalf("%s: bad: %#v %#v", tc.in, tokens, tc.out)
}
}
}
func TestCollapse(t *testing.T) {
inp := "type:key key:foo key:bar"
tokens, err := tokenize(inp)
if err != nil {
t.Fatalf("err: %v", err)
}
out := collapse(tokens)
expect := map[string][]string{
"type": []string{"key"},
"key": []string{"foo", "bar"},
}
if !reflect.DeepEqual(out, expect) {
t.Fatalf("bad: %#v", out)
}
}
func TestParseBasic(t *testing.T) {
p, err := Parse("type:key datacenter:dc2 token:12345 key:foo")
params := makeParams(t, `{"type":"key", "datacenter":"dc2", "token":"12345", "key":"foo"}`)
p, err := Parse(params)
if err != nil {
t.Fatalf("err: %v", err)
}
@ -106,7 +24,8 @@ func TestParseBasic(t *testing.T) {
}
func TestParse_exempt(t *testing.T) {
p, err := ParseExempt("type:key key:foo handler:foobar", []string{"handler"})
params := makeParams(t, `{"type":"key", "key":"foo", "handler": "foobar"}`)
p, err := ParseExempt(params, []string{"handler"})
if err != nil {
t.Fatalf("err: %v", err)
}
@ -114,7 +33,16 @@ func TestParse_exempt(t *testing.T) {
t.Fatalf("Bad: %#v", p)
}
ex := p.Exempt["handler"]
if len(ex) != 1 && ex[0] != "foobar" {
if ex != "foobar" {
t.Fatalf("bad: %v", ex)
}
}
func makeParams(t *testing.T, s string) map[string]interface{} {
var out map[string]interface{}
dec := json.NewDecoder(bytes.NewReader([]byte(s)))
if err := dec.Decode(&out); err != nil {
t.Fatalf("err: %v", err)
}
return out
}