mirror of https://github.com/status-im/consul.git
Makes the default protocol 2 and lets 3 interoperate with 2.
This commit is contained in:
parent
80d5a3029f
commit
660da92152
|
@ -545,6 +545,22 @@ func (a *Agent) WANMembers() []serf.Member {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// CanServersUnderstandProtocol checks to see if all the servers understand the
|
||||||
|
// given protocol version.
|
||||||
|
func (a *Agent) CanServersUnderstandProtocol(version uint8) bool {
|
||||||
|
numServers, numWhoGrok := 0, 0
|
||||||
|
members := a.LANMembers()
|
||||||
|
for _, member := range members {
|
||||||
|
if member.Tags["role"] == "consul" {
|
||||||
|
numServers++
|
||||||
|
if member.ProtocolMax >= version {
|
||||||
|
numWhoGrok++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return (numServers > 0) && (numWhoGrok == numServers)
|
||||||
|
}
|
||||||
|
|
||||||
// StartSync is called once Services and Checks are registered.
|
// StartSync is called once Services and Checks are registered.
|
||||||
// This is called to prevent a race between clients and the anti-entropy routines
|
// This is called to prevent a race between clients and the anti-entropy routines
|
||||||
func (a *Agent) StartSync() {
|
func (a *Agent) StartSync() {
|
||||||
|
@ -562,6 +578,16 @@ func (a *Agent) ResumeSync() {
|
||||||
a.state.Resume()
|
a.state.Resume()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Returns the coordinate of this node in the local pool (assumes coordinates
|
||||||
|
// are enabled, so check that before calling).
|
||||||
|
func (a *Agent) GetCoordinate() (*coordinate.Coordinate, error) {
|
||||||
|
if a.config.Server {
|
||||||
|
return a.server.GetLANCoordinate()
|
||||||
|
} else {
|
||||||
|
return a.client.GetCoordinate()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// sendCoordinate is a long-running loop that periodically sends our coordinate
|
// sendCoordinate is a long-running loop that periodically sends our coordinate
|
||||||
// to the server. Closing the agent's shutdownChannel will cause this to exit.
|
// to the server. Closing the agent's shutdownChannel will cause this to exit.
|
||||||
func (a *Agent) sendCoordinate() {
|
func (a *Agent) sendCoordinate() {
|
||||||
|
@ -573,14 +599,13 @@ func (a *Agent) sendCoordinate() {
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-time.After(intv):
|
case <-time.After(intv):
|
||||||
|
if !a.CanServersUnderstandProtocol(3) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
var c *coordinate.Coordinate
|
var c *coordinate.Coordinate
|
||||||
var err error
|
var err error
|
||||||
if a.config.Server {
|
if c, err = a.GetCoordinate(); err != nil {
|
||||||
c, err = a.server.GetLANCoordinate()
|
|
||||||
} else {
|
|
||||||
c, err = a.client.GetCoordinate()
|
|
||||||
}
|
|
||||||
if err != nil {
|
|
||||||
a.logger.Printf("[ERR] agent: failed to get coordinate: %s", err)
|
a.logger.Printf("[ERR] agent: failed to get coordinate: %s", err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
|
@ -17,18 +17,17 @@ type AgentSelf struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *HTTPServer) AgentSelf(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
func (s *HTTPServer) AgentSelf(resp http.ResponseWriter, req *http.Request) (interface{}, error) {
|
||||||
var coord *coordinate.Coordinate
|
var c *coordinate.Coordinate
|
||||||
if !s.agent.config.DisableCoordinates {
|
if !s.agent.config.DisableCoordinates {
|
||||||
var err error
|
var err error
|
||||||
coord, err = s.agent.server.GetLANCoordinate()
|
if c, err = s.agent.GetCoordinate(); err != nil {
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return AgentSelf{
|
return AgentSelf{
|
||||||
Config: s.agent.config,
|
Config: s.agent.config,
|
||||||
Coord: coord,
|
Coord: c,
|
||||||
Member: s.agent.LocalMember(),
|
Member: s.agent.LocalMember(),
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
|
@ -1581,3 +1581,51 @@ func TestAgent_purgeCheckState(t *testing.T) {
|
||||||
t.Fatalf("should have removed file")
|
t.Fatalf("should have removed file")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAgent_GetCoordinate(t *testing.T) {
|
||||||
|
check := func(server bool) {
|
||||||
|
config := nextConfig()
|
||||||
|
config.Server = server
|
||||||
|
dir, agent := makeAgent(t, config)
|
||||||
|
defer os.RemoveAll(dir)
|
||||||
|
defer agent.Shutdown()
|
||||||
|
|
||||||
|
// This doesn't verify the returned coordinate, but it makes
|
||||||
|
// sure that the agent chooses the correct Serf instance,
|
||||||
|
// depending on how it's configured as a client or a server.
|
||||||
|
// If it chooses the wrong one, this will crash.
|
||||||
|
if _, err := agent.GetCoordinate(); err != nil {
|
||||||
|
t.Fatalf("err: %s", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
check(true)
|
||||||
|
check(false)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAgent_CanServersUnderstandProtocol(t *testing.T) {
|
||||||
|
config := nextConfig()
|
||||||
|
dir, agent := makeAgent(t, config)
|
||||||
|
defer os.RemoveAll(dir)
|
||||||
|
defer agent.Shutdown()
|
||||||
|
|
||||||
|
min := uint8(consul.ProtocolVersionMin)
|
||||||
|
if !agent.CanServersUnderstandProtocol(min) {
|
||||||
|
t.Fatalf("should grok %d", min)
|
||||||
|
}
|
||||||
|
|
||||||
|
max := uint8(consul.ProtocolVersionMax)
|
||||||
|
if !agent.CanServersUnderstandProtocol(max) {
|
||||||
|
t.Fatalf("should grok %d", max)
|
||||||
|
}
|
||||||
|
|
||||||
|
current := uint8(config.Protocol)
|
||||||
|
if !agent.CanServersUnderstandProtocol(current) {
|
||||||
|
t.Fatalf("should grok %d", current)
|
||||||
|
}
|
||||||
|
|
||||||
|
future := max + 1
|
||||||
|
if agent.CanServersUnderstandProtocol(future) {
|
||||||
|
t.Fatalf("should not grok %d", future)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
@ -478,7 +478,7 @@ func DefaultConfig() *Config {
|
||||||
},
|
},
|
||||||
StatsitePrefix: "consul",
|
StatsitePrefix: "consul",
|
||||||
SyslogFacility: "LOCAL0",
|
SyslogFacility: "LOCAL0",
|
||||||
Protocol: consul.ProtocolVersionMax,
|
Protocol: consul.ProtocolVersion2Compatible,
|
||||||
CheckUpdateInterval: 5 * time.Minute,
|
CheckUpdateInterval: 5 * time.Minute,
|
||||||
AEInterval: time.Minute,
|
AEInterval: time.Minute,
|
||||||
DisableCoordinates: false,
|
DisableCoordinates: false,
|
||||||
|
|
|
@ -32,6 +32,7 @@ func init() {
|
||||||
protocolVersionMap = map[uint8]uint8{
|
protocolVersionMap = map[uint8]uint8{
|
||||||
1: 4,
|
1: 4,
|
||||||
2: 4,
|
2: 4,
|
||||||
|
3: 4,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -267,7 +268,7 @@ func DefaultConfig() *Config {
|
||||||
SerfLANConfig: serf.DefaultConfig(),
|
SerfLANConfig: serf.DefaultConfig(),
|
||||||
SerfWANConfig: serf.DefaultConfig(),
|
SerfWANConfig: serf.DefaultConfig(),
|
||||||
ReconcileInterval: 60 * time.Second,
|
ReconcileInterval: 60 * time.Second,
|
||||||
ProtocolVersion: ProtocolVersionMax,
|
ProtocolVersion: ProtocolVersion2Compatible,
|
||||||
ACLTTL: 30 * time.Second,
|
ACLTTL: 30 * time.Second,
|
||||||
ACLDefaultPolicy: "allow",
|
ACLDefaultPolicy: "allow",
|
||||||
ACLDownPolicy: "extend-cache",
|
ACLDownPolicy: "extend-cache",
|
||||||
|
|
|
@ -28,7 +28,15 @@ import (
|
||||||
// protocol versions.
|
// protocol versions.
|
||||||
const (
|
const (
|
||||||
ProtocolVersionMin uint8 = 1
|
ProtocolVersionMin uint8 = 1
|
||||||
ProtocolVersionMax = 2
|
|
||||||
|
// Version 3 added support for network coordinates but we kept the
|
||||||
|
// default protocol version at 2 to ease the transition to this new
|
||||||
|
// feature. A Consul agent speaking version 2 of the protocol will
|
||||||
|
// attempt to send its coordinates to a server who understands version
|
||||||
|
// 3 or greater.
|
||||||
|
ProtocolVersion2Compatible = 2
|
||||||
|
|
||||||
|
ProtocolVersionMax = 3
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
|
Loading…
Reference in New Issue