add matchtag and fix tests

config.yaml

@@ -11,6 +11,9 @@ agent:
   consul_query_interval: 5m
   # token to authenticate client if consul requires it
   consul_token: 00000000-0000-0000-0000-000000000000
+  # this tag must be present in consul for Gocast to pickup the service
+  # by default its enable_gocast
+  consul_match_tag: enable_gocast
 
 bgp:
   local_as: 12345
@@ -21,6 +24,12 @@ bgp:
     - asn:nnnn
     - asn:nnnn
   origin: igp
+  # family will determine whether we want to peer over an ipv4 or ipv6 session
+  # one gocast instance can only peer over one session. Multiple sessions are not
+  # supported. The default is 4
+  family: 4
+  # router ID is required when family 6 is specified
+  router_id: 1.1.1.1
 
 # optional list of apps to register on startup
 apps:

config/config.go

@@ -15,7 +15,8 @@ type AgentConfig struct {
 	CleanupTimer        time.Duration `yaml:"cleanup_timer"`
 	ConsulAddr          string        `yaml:"consul_addr"`
 	ConsulQueryInterval time.Duration `yaml:"consul_query_interval"`
-	ConsulToken			string		  `yaml:"consul_token"`
+	ConsulToken         string        `yaml:"consul_token"`
+	ConsulMatchTag      string        `yaml:"consul_match_tag"`
 }
 
 type BgpConfig struct {
@@ -25,6 +26,8 @@ type BgpConfig struct {
 	PeerIP      string `yaml:"peer_ip"`
 	Communities []string
 	Origin      string
+	Family      int
+	RouterID    string `yaml:"router_id"`
 }
 
 type VipConfig struct {

controller/bgp.go

@@ -32,10 +32,13 @@ func NewController(config c.BgpConfig) (*Controller, error) {
 	c := &Controller{}
 	var gw net.IP
 	var err error
+	if config.Family == 0 {
+		config.Family = 4
+	}
 	if config.PeerIP == "" {
-		gw, err := gateway()
+		gw, err := gateway(config.Family)
 		if err != nil {
-			return nil, fmt.Errorf("Unable to get gw ip: %v", err)
+			return nil, fmt.Errorf("unable to get gw ip: %v", err)
 		}
 		c.peerIP = gw
 	} else {
@@ -44,7 +47,7 @@ func NewController(config c.BgpConfig) (*Controller, error) {
 	if config.LocalIP == "" {
 		gw, err = via(c.peerIP)
 		if err != nil {
-			return nil, fmt.Errorf("Unable to get gw ip: %v", err)
+			return nil, fmt.Errorf("unable to get gw ip: %v", err)
 		}
 		c.localIP, err = localAddress(gw)
 		if err != nil {
@@ -62,16 +65,23 @@ func NewController(config c.BgpConfig) (*Controller, error) {
 	case "unknown":
 		c.origin = 2
 	}
+	var rid string
+	if config.Family == 4 {
+		rid = c.localIP.String()
+	}
+	if config.RouterID != "" {
+		rid = config.RouterID
+	}
 	s := gobgp.NewBgpServer()
 	go s.Serve()
 	if err := s.StartBgp(context.Background(), &api.StartBgpRequest{
 		Global: &api.Global{
 			As:         uint32(config.LocalAS),
-			RouterId:   c.localIP.String(),
+			RouterId:   rid,
 			ListenPort: -1, // gobgp won't listen on tcp:179
 		},
 	}); err != nil {
-		return nil, fmt.Errorf("Unable to start bgp: %v", err)
+		return nil, fmt.Errorf("unable to start bgp: %v", err)
 	}
 	c.s = s
 	c.peerAS = config.PeerAS

controller/consul.go

@@ -17,7 +17,7 @@ const (
 	consulNodeEnv        = "CONSUL_NODE"
 	consulToken          = "CONSUL_TOKEN"
 	allowStale           = "CONSUL_STALE"
-	matchTag             = "enable_gocast"
+	defaultmatchTag      = "enable_gocast"
 	nodeURL              = "/catalog/node"
 	remoteHealthCheckurl = "/health/checks"
 	localHealthCheckurl  = "/agent/checks"
@@ -32,10 +32,11 @@ type Client struct {
 }
 
 type ConsulMon struct {
-	addr   string
-	token  string
-	node   string
-	client Clienter
+	addr     string
+	token    string
+	node     string
+	matchTag string
+	client   Clienter
 }
 
 type ConsulServiceData struct {
@@ -55,12 +56,15 @@ func contains(inp []string, elem string) bool {
 	return false
 }
 
-func NewConsulMon(addr string, token string) (*ConsulMon, error) {
+func NewConsulMon(addr string, token string, matchTag string) (*ConsulMon, error) {
 	node := os.Getenv(consulNodeEnv)
 	if node == "" {
 		return nil, fmt.Errorf("%s env variable not set", consulNodeEnv)
 	}
-	return &ConsulMon{addr: addr, token: token, node: node, client: &http.Client{Timeout: 10 * time.Second}}, nil
+	if matchTag == "" {
+		matchTag = defaultmatchTag
+	}
+	return &ConsulMon{addr: addr, token: token, node: node, client: &http.Client{Timeout: 10 * time.Second}, matchTag: matchTag}, nil
 }
 
 func getHTTPReq(httpMethod string, addr string, tokenFrmCfg string) (*http.Request, error) {
@@ -95,10 +99,10 @@ func (c *ConsulMon) queryServices() ([]*App, error) {
 	defer resp.Body.Close()
 	var consulData ConsulServiceData
 	if err := json.NewDecoder(resp.Body).Decode(&consulData); err != nil {
-		return apps, fmt.Errorf("Unable to decode consul data: %v", err)
+		return apps, fmt.Errorf("unable to decode consul data: %v", err)
 	}
 	for _, service := range consulData.Services {
-		if !contains(service.Tags, matchTag) {
+		if !contains(service.Tags, c.matchTag) {
 			continue
 		}
 		var (
@@ -170,7 +174,7 @@ func (c *ConsulMon) healthCheckLocal(service string) (bool, error) {
 			return false, nil
 		}
 	}
-	return false, fmt.Errorf("No local healthcheck info found for service %s on node %s in consul", service, c.node)
+	return false, fmt.Errorf("no local healthcheck info found for service %s on node %s in consul", service, c.node)
 }
 
 // healthCheckRemote queries the consul cluster's healthcheck endpoint to perform service healthchecks
@@ -202,7 +206,7 @@ func (c *ConsulMon) healthCheckRemote(service string) (bool, error) {
 			return false, nil
 		}
 	}
-	return false, fmt.Errorf("No healthcheck info found for node %s in consul", c.node)
+	return false, fmt.Errorf("no healthcheck info found for node %s in consul", c.node)
 }
 
 // healthCheck determines if we should use the local agent

controller/consul_test.go

@@ -2,7 +2,7 @@ package controller
 
 import (
 	"bytes"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"os"
 	"testing"
@@ -119,13 +119,13 @@ func TestQueryServices(t *testing.T) {
 	a := assert.New(t)
 	client := &MockClient{}
 	cm := &ConsulMon{
-		addr: "foo", node: "test", client: client,
+		addr: "foo", node: "test", client: client, matchTag: "enable_gocast",
 	}
 
 	// test valid app
 	client.do = func(*http.Request) (*http.Response, error) {
 		b := bytes.NewBuffer([]byte(mockConsulData["single-app"]))
-		return &http.Response{Body: ioutil.NopCloser(b), StatusCode: http.StatusOK}, nil
+		return &http.Response{Body: io.NopCloser(b), StatusCode: http.StatusOK}, nil
 	}
 	apps, err := cm.queryServices()
 	if err != nil {
@@ -140,7 +140,7 @@ func TestQueryServices(t *testing.T) {
 	// test no match
 	client.do = func(*http.Request) (*http.Response, error) {
 		b := bytes.NewBuffer([]byte(mockConsulData["single-app-no-match"]))
-		return &http.Response{Body: ioutil.NopCloser(b), StatusCode: http.StatusOK}, nil
+		return &http.Response{Body: io.NopCloser(b), StatusCode: http.StatusOK}, nil
 	}
 	apps, err = cm.queryServices()
 	if err != nil {
@@ -151,7 +151,7 @@ func TestQueryServices(t *testing.T) {
 	// test missing vip
 	client.do = func(*http.Request) (*http.Response, error) {
 		b := bytes.NewBuffer([]byte(mockConsulData["single-app-no-vip"]))
-		return &http.Response{Body: ioutil.NopCloser(b), StatusCode: http.StatusOK}, nil
+		return &http.Response{Body: io.NopCloser(b), StatusCode: http.StatusOK}, nil
 	}
 	apps, _ = cm.queryServices()
 	a.Equal(0, len(apps))
@@ -160,13 +160,13 @@ func TestQueryServices(t *testing.T) {
 func TestHealthCheck(t *testing.T) {
 	a := assert.New(t)
 	client := &MockClient{}
-	cm := &ConsulMon{node: "test-node1", client: client}
+	cm := &ConsulMon{node: "test-node1", client: client, matchTag: "enable_gocast"}
 
 	// test remote checks
 	cm.addr = "http://remote/check"
 	client.do = func(*http.Request) (*http.Response, error) {
 		b := bytes.NewBuffer([]byte(mockConsulCheckData["remote-pass"]))
-		return &http.Response{Body: ioutil.NopCloser(b), StatusCode: http.StatusOK}, nil
+		return &http.Response{Body: io.NopCloser(b), StatusCode: http.StatusOK}, nil
 	}
 	check, err := cm.healthCheck("test-service")
 	if err != nil {
@@ -175,7 +175,7 @@ func TestHealthCheck(t *testing.T) {
 	a.True(check)
 	client.do = func(*http.Request) (*http.Response, error) {
 		b := bytes.NewBuffer([]byte(mockConsulCheckData["remote-fail"]))
-		return &http.Response{Body: ioutil.NopCloser(b), StatusCode: http.StatusOK}, nil
+		return &http.Response{Body: io.NopCloser(b), StatusCode: http.StatusOK}, nil
 	}
 	check, _ = cm.healthCheck("test-service")
 	a.False(check)
@@ -184,7 +184,7 @@ func TestHealthCheck(t *testing.T) {
 	cm.addr = "http://localhost/check"
 	client.do = func(*http.Request) (*http.Response, error) {
 		b := bytes.NewBuffer([]byte(mockConsulCheckData["local-pass"]))
-		return &http.Response{Body: ioutil.NopCloser(b), StatusCode: http.StatusOK}, nil
+		return &http.Response{Body: io.NopCloser(b), StatusCode: http.StatusOK}, nil
 	}
 	check, _ = cm.healthCheck("test-service")
 	if err != nil {
@@ -194,7 +194,7 @@ func TestHealthCheck(t *testing.T) {
 	cm.addr = "http://127.0.0.1/check"
 	client.do = func(*http.Request) (*http.Response, error) {
 		b := bytes.NewBuffer([]byte(mockConsulCheckData["local-fail"]))
-		return &http.Response{Body: ioutil.NopCloser(b), StatusCode: http.StatusOK}, nil
+		return &http.Response{Body: io.NopCloser(b), StatusCode: http.StatusOK}, nil
 	}
 	check, _ = cm.healthCheck("test-service")
 	a.False(check)

controller/monitor.go

@@ -84,7 +84,7 @@ func NewMonitor(config *c.Config) *MonitorMgr {
 		cleanups: make(map[string]chan bool),
 	}
 	if config.Agent.ConsulAddr != "" {
-		cmon, err := NewConsulMon(config.Agent.ConsulAddr, config.Agent.ConsulToken)
+		cmon, err := NewConsulMon(config.Agent.ConsulAddr, config.Agent.ConsulToken, config.Agent.ConsulMatchTag)
 		if err != nil {
 			glog.Errorf("Failed to start consul monitor: %v", err)
 		} else {
@@ -184,7 +184,7 @@ func (m *MonitorMgr) Add(app *App) {
 }
 
 // Remove removes an app from monitor manager, stops BGP
-/// announcement and cleans up state
+// / announcement and cleans up state
 func (m *MonitorMgr) Remove(appName string) {
 	m.monMu.Lock()
 	defer m.monMu.Unlock()

controller/system.go

@@ -18,22 +18,30 @@ func getCmdList(mainCmd string) []string {
 	return cmdList
 }
 
-func gateway() (net.IP, error) {
-	cmd := `ip route | grep "^default" | cut -d" " -f3`
+func gateway(family int) (net.IP, error) {
+	prefix := "ip"
+	if family == 6 {
+		prefix = "ip -6"
+	}
+	cmd := fmt.Sprintf(`%s route | grep "^default" | cut -d" " -f3`, prefix)
 	cmdList := getCmdList(cmd)
 	out, err := exec.Command(execCmd, cmdList...).Output()
 	if err != nil {
-		return nil, fmt.Errorf("Failed to execute command: %s: %v", cmd, err)
+		return nil, fmt.Errorf("failed to execute command: %s: %v", cmd, err)
 	}
 	return net.ParseIP(strings.TrimSpace(string(out))), nil
 }
 
 func via(dest net.IP) (net.IP, error) {
-	cmd := fmt.Sprintf(`ip route get %s | grep via | cut -d" " -f3`, dest.String())
+	prefix := "ip"
+	if dest.To4() == nil {
+		prefix = "ip -6"
+	}
+	cmd := fmt.Sprintf(`%s route get %s | grep via | cut -d" " -f3`, prefix, dest.String())
 	cmdList := getCmdList(cmd)
 	out, err := exec.Command(execCmd, cmdList...).Output()
 	if err != nil {
-		return nil, fmt.Errorf("Failed to execute command: %s: %v", cmd, err)
+		return nil, fmt.Errorf("failed to execute command: %s: %v", cmd, err)
 	}
 	if string(out) == "" {
 		// assume the provided dest is the next hop
@@ -55,7 +63,7 @@ func localAddress(gw net.IP) (net.IP, error) {
 			}
 		}
 	}
-	return nil, fmt.Errorf("Unable to find local address")
+	return nil, fmt.Errorf("unable to find local address")
 }
 
 func addLoopback(name string, addr *net.IPNet) error {
@@ -66,30 +74,42 @@ func addLoopback(name string, addr *net.IPNet) error {
 	if len(label) > 15 {
 		label = label[:15]
 	}
-	cmd := fmt.Sprintf("ip address add %s/%d dev lo label %s", addr.IP.String(), prefixLen, label)
+	prefix := "ip"
+	if addr.IP.To4() == nil {
+		prefix = "ip -6"
+	}
+	cmd := fmt.Sprintf("%s address add %s/%d dev lo label %s", prefix, addr.IP.String(), prefixLen, label)
 	cmdList := getCmdList(cmd)
 	_, err := exec.Command(execCmd, cmdList...).Output()
 	if err != nil {
-		return fmt.Errorf("Failed to Add loopback command: %s: %v", cmd, err)
+		return fmt.Errorf("failed to Add loopback command: %s: %v", cmd, err)
 	}
 	return nil
 }
 
 func deleteLoopback(addr *net.IPNet) error {
+	prefix := "ip"
+	if addr.IP.To4() == nil {
+		prefix = "ip -6"
+	}
 	prefixLen, _ := addr.Mask.Size()
-	cmd := fmt.Sprintf("ip address delete %s/%d dev lo", addr.IP.String(), prefixLen)
+	cmd := fmt.Sprintf("%s address delete %s/%d dev lo", prefix, addr.IP.String(), prefixLen)
 	cmdList := getCmdList(cmd)
 	_, err := exec.Command(execCmd, cmdList...).Output()
 	if err != nil {
-		return fmt.Errorf("Failed to delete loopback command: %s: %v", cmd, err)
+		return fmt.Errorf("failed to delete loopback command: %s: %v", cmd, err)
 	}
 	return nil
 }
 
 func natRule(op string, vip, localAddr net.IP, protocol, lport, dport string) error {
+	prefix := "iptables"
+	if vip.To4() == nil {
+		prefix = "ip6tables"
+	}
 	cmd := fmt.Sprintf(
-		"iptables -t nat -%s PREROUTING -p %s -d %s --dport %s -j DNAT --to-destination %s:%s",
-		op, protocol, vip.String(), lport, localAddr.String(), dport,
+		"%s -t nat -%s PREROUTING -p %s -d %s --dport %s -j DNAT --to-destination %s:%s",
+		prefix, op, protocol, vip.String(), lport, localAddr.String(), dport,
 	)
 	cmdList := getCmdList(cmd)
 	_, err := exec.Command(execCmd, cmdList...).Output()

controller/system_test.go

@@ -12,9 +12,14 @@ import (
 func TestGateway(t *testing.T) {
 	execCmd = os.Args[0]
 	os.Setenv("test_name", "test_gateway")
-	gw, err := gateway()
+	gw, err := gateway(4)
 	assert.Nil(t, err)
 	assert.Equal(t, "10.1.1.1", gw.String())
+
+	os.Setenv("test_name", "test_gateway_v6")
+	gw, err = gateway(6)
+	assert.Nil(t, err)
+	assert.Equal(t, "2001:dead:beef::1", gw.String())
 }
 
 func TestVia(t *testing.T) {
@@ -28,6 +33,11 @@ func TestVia(t *testing.T) {
 	ip, err = via(net.ParseIP("10.1.4.1"))
 	assert.Nil(t, err)
 	assert.Equal(t, "10.1.4.1", ip.String())
+
+	os.Setenv("test_name", "test_via_v6")
+	ip, err = via(net.ParseIP("2001:dead:beef::100"))
+	assert.Nil(t, err)
+	assert.Equal(t, "2001:dead:beef::1", ip.String())
 }
 
 func TestAddLoopback(t *testing.T) {
@@ -41,14 +51,23 @@ func TestAddLoopback(t *testing.T) {
 	_, ipnet, _ = net.ParseCIDR("1.1.1.1/32")
 	err = addLoopback("test_app", ipnet)
 	assert.NotNil(t, err)
+
+	os.Setenv("test_name", "test_add_v6")
+	_, ipnet, _ = net.ParseCIDR("2001:dead:beef:1001::100/64")
+	err = addLoopback("test_app", ipnet)
+	assert.Nil(t, err)
 }
 
 func TestMain(m *testing.M) {
 	switch os.Getenv("test_name") {
 	case "test_gateway":
 		fmt.Println("10.1.1.1")
+	case "test_gateway_v6":
+		fmt.Println("2001:dead:beef::1")
 	case "test_via":
 		fmt.Println("10.1.2.1")
+	case "test_via_v6":
+		fmt.Println("2001:dead:beef::1")
 	case "test_via_none":
 		break
 	case "test_add_fail":