kemko/nomad
1
0
Fork 0
mirror of https://github.com/kemko/nomad.git synced 2026-01-03 00:45:43 +03:00
Code Issues Packages Projects Releases Wiki Activity

goling(1) compliance pass (e.g. Rpc* -> RPC)

Browse Source
This commit is contained in:
Sean Chittenden
2016-06-10 23:26:15 -04:00
parent 6819f2b68d
commit 5331ea7bc6
8 changed files with 85 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
client/client.go
View File

@@ -302,15 +302,15 @@ func (c *Client) Region() string {
return c.config.Region
}
// RpcMajorVersion returns the structs.ApiMajorVersion supported by the
// RPCMajorVersion returns the structs.ApiMajorVersion supported by the
// client.
func (c *Client) RpcMajorVersion() int {
func (c *Client) RPCMajorVersion() int {
return structs.ApiMajorVersion
}
// RpcMinorVersion returns the structs.ApiMinorVersion supported by the
// RPCMinorVersion returns the structs.ApiMinorVersion supported by the
// client.
func (c *Client) RpcMinorVersion() int {
func (c *Client) RPCMinorVersion() int {
return structs.ApiMinorVersion
}
@@ -354,7 +354,7 @@ func (c *Client) RPC(method string, args interface{}, reply interface{}) error {
}
// Make the RPC request
if err := c.connPool.RPC(c.Region(), server.Addr, c.RpcMajorVersion(), method, args, reply); err != nil {
if err := c.connPool.RPC(c.Region(), server.Addr, c.RPCMajorVersion(), method, args, reply); err != nil {
c.rpcProxy.NotifyFailedServer(server)
c.logger.Printf("[ERR] client: RPC failed to server %s: %v", server.Addr, err)
return err
@@ -1307,7 +1307,7 @@ func (c *Client) setupConsulSyncer() error {
Near: "_agent",
WaitTime: consul.DefaultQueryWaitDuration,
}
consulServices, _, err := consulCatalog.Service(nomadServerServiceName, consul.ServiceTagRpc, opts)
consulServices, _, err := consulCatalog.Service(nomadServerServiceName, consul.ServiceTagRPC, opts)
if err != nil {
mErr.Errors = append(mErr.Errors, fmt.Errorf("unable to query service %q from Consul datacenter %q: %v", nomadServerServiceName, dc, err))
continue
@@ -1326,7 +1326,7 @@ func (c *Client) setupConsulSyncer() error {
continue
}
var ok bool
if ok, err = c.connPool.PingNomadServer(c.Region(), c.RpcMajorVersion(), serverEndpoint); err != nil {
if ok, err = c.connPool.PingNomadServer(c.Region(), c.RPCMajorVersion(), serverEndpoint); err != nil {
mErr.Errors = append(mErr.Errors, err)
continue
}
@@ -1344,7 +1344,7 @@ func (c *Client) setupConsulSyncer() error {
return mErr.ErrorOrNil()
}
for i, _ := range dcs {
for i := range dcs {
dcs[i] = fmt.Sprintf("%q", dcs[i])
}
return fmt.Errorf("no Nomad Servers advertising service %q in Consul datacenters: %s", nomadServerServiceName, dcs)
@@ -1381,7 +1381,7 @@ func (c *Client) setupConsulSyncer() error {
const estInitialConsulServices = 8
const serviceGroupName = "executor"
services := make([]*structs.ConsulService, 0, estInitialConsulServices)
for allocId, ar := range c.getAllocRunners() {
for allocID, ar := range c.getAllocRunners() {
ar.taskStatusLock.RLock()
taskStates := copyTaskStates(ar.taskStates)
ar.taskStatusLock.RUnlock()
@@ -1390,10 +1390,10 @@ func (c *Client) setupConsulSyncer() error {
if tr, ok := ar.tasks[taskName]; ok {
for _, service := range tr.task.ConsulServices {
if service.Name == "" {
service.Name = fmt.Sprintf("%s-%s", tr.task.Name, allocId)
service.Name = fmt.Sprintf("%s-%s", tr.task.Name, allocID)
}
if service.ServiceID == "" {
service.ServiceID = fmt.Sprintf("%s-%s:%s/%s", c.consulSyncer.GenerateServiceID(serviceGroupName, service), tr.task.Name, allocId)
service.ServiceID = fmt.Sprintf("%s-%s:%s/%s", c.consulSyncer.GenerateServiceID(serviceGroupName, service), tr.task.Name, allocID)
}
services = append(services, service)
}
@@ -1461,6 +1461,7 @@ func (c *Client) emitStats(hStats *stats.HostStats) {
}
}
// RPCProxy returns the Client's RPCProxy instance
func (c *Client) RPCProxy() *rpcproxy.RPCProxy {
return c.rpcProxy
}

22
client/rpcproxy/rpcproxy.go
View File

@@ -63,8 +63,8 @@ const (
// configuration to prevents a cyclic import dependency.
type NomadConfigInfo interface {
Datacenter() string
RpcMajorVersion() int
RpcMinorVersion() int
RPCMajorVersion() int
RPCMinorVersion() int
Region() string
}
@@ -468,7 +468,7 @@ func (p *RPCProxy) RebalanceServers() {
// detect the failed node.
selectedServer := l.L[0]
ok, err := p.connPoolPinger.PingNomadServer(p.configInfo.Region(), p.configInfo.RpcMajorVersion(), selectedServer)
ok, err := p.connPoolPinger.PingNomadServer(p.configInfo.Region(), p.configInfo.RPCMajorVersion(), selectedServer)
if ok {
foundHealthyServer = true
break
@@ -697,23 +697,23 @@ func (p *RPCProxy) RefreshServerLists(servers []*structs.NodeServerInfo, numNode
// TODO(sean@): Move the logging throttle logic into a
// dedicated logging package so RPCProxy does not have to
// perform this accounting.
if int32(p.configInfo.RpcMajorVersion()) < s.RpcMajorVersion ||
(int32(p.configInfo.RpcMajorVersion()) == s.RpcMajorVersion &&
int32(p.configInfo.RpcMinorVersion()) < s.RpcMinorVersion) {
if int32(p.configInfo.RPCMajorVersion()) < s.RPCMajorVersion ||
(int32(p.configInfo.RPCMajorVersion()) == s.RPCMajorVersion &&
int32(p.configInfo.RPCMinorVersion()) < s.RPCMinorVersion) {
now := time.Now()
t, ok := p.rpcAPIMismatchThrottle[s.RpcAdvertiseAddr]
t, ok := p.rpcAPIMismatchThrottle[s.RPCAdvertiseAddr]
if ok && t.After(now) {
continue
}
p.logger.Printf("[WARN] client.rpcproxy: API mismatch between client version (v%d.%d) and server version (v%d.%d), ignoring server %q", p.configInfo.RpcMajorVersion(), p.configInfo.RpcMinorVersion(), s.RpcMajorVersion, s.RpcMinorVersion, s.RpcAdvertiseAddr)
p.rpcAPIMismatchThrottle[s.RpcAdvertiseAddr] = now.Add(rpcAPIMismatchLogRate)
p.logger.Printf("[WARN] client.rpcproxy: API mismatch between client version (v%d.%d) and server version (v%d.%d), ignoring server %q", p.configInfo.RPCMajorVersion(), p.configInfo.RPCMinorVersion(), s.RPCMajorVersion, s.RPCMinorVersion, s.RPCAdvertiseAddr)
p.rpcAPIMismatchThrottle[s.RPCAdvertiseAddr] = now.Add(rpcAPIMismatchLogRate)
continue
}
server, err := NewServerEndpoint(s.RpcAdvertiseAddr)
server, err := NewServerEndpoint(s.RPCAdvertiseAddr)
if err != nil {
p.logger.Printf("[WARN] client.rpcproxy: Unable to create a server from %q: %v", s.RpcAdvertiseAddr, err)
p.logger.Printf("[WARN] client.rpcproxy: Unable to create a server from %q: %v", s.RPCAdvertiseAddr, err)
continue
}

6
client/rpcproxy/rpcproxy_test.go
View File

@@ -82,11 +82,11 @@ func (s *fauxSerf) Datacenter() string {
return s.datacenter
}
func (s *fauxSerf) RpcMajorVersion() int {
func (s *fauxSerf) RPCMajorVersion() int {
return s.rpcMajorVersion
}
func (s *fauxSerf) RpcMinorVersion() int {
func (s *fauxSerf) RPCMinorVersion() int {
return s.rpcMinorVersion
}
@@ -645,7 +645,7 @@ func test_reconcileServerList(maxServers int) (bool, error) {
// failPct of the servers for the reconcile. This
// allows for the selected server to no longer be
// healthy for the reconcile below.
if ok, _ := p.connPoolPinger.PingNomadServer(p.configInfo.Region(), p.configInfo.RpcMajorVersion(), node); ok {
if ok, _ := p.connPoolPinger.PingNomadServer(p.configInfo.Region(), p.configInfo.RPCMajorVersion(), node); ok {
// Will still be present
healthyServers = append(healthyServers, node)
} else {

82
command/agent/agent.go
View File

@@ -33,12 +33,12 @@ type Agent struct {
consulSyncer *consul.Syncer
client *client.Client
clientHttpAddr string
clientRpcAddr string
clientHTTPAddr string
clientRPCAddr string
server *nomad.Server
serverHttpAddr string
serverRpcAddr string
serverHTTPAddr string
serverRPCAddr string
serverSerfAddr string
shutdown bool
@@ -174,35 +174,35 @@ func (a *Agent) serverConfig() (*nomad.Config, error) {
// Resolve the Server's HTTP Address
if a.config.AdvertiseAddrs.HTTP != "" {
a.serverHttpAddr = a.config.AdvertiseAddrs.HTTP
a.serverHTTPAddr = a.config.AdvertiseAddrs.HTTP
} else if a.config.Addresses.HTTP != "" {
a.serverHttpAddr = fmt.Sprintf("%v:%v", a.config.Addresses.HTTP, a.config.Ports.HTTP)
a.serverHTTPAddr = fmt.Sprintf("%v:%v", a.config.Addresses.HTTP, a.config.Ports.HTTP)
} else if a.config.BindAddr != "" {
a.serverHttpAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.HTTP)
a.serverHTTPAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.HTTP)
} else {
a.serverHttpAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.HTTP)
a.serverHTTPAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.HTTP)
}
addr, err := net.ResolveTCPAddr("tcp", a.serverHttpAddr)
addr, err := net.ResolveTCPAddr("tcp", a.serverHTTPAddr)
if err != nil {
return nil, fmt.Errorf("error resolving HTTP addr %q: %v:", a.serverHttpAddr, err)
return nil, fmt.Errorf("error resolving HTTP addr %q: %v", a.serverHTTPAddr, err)
}
a.serverHttpAddr = fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
a.serverHTTPAddr = fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
// Resolve the Server's RPC Address
if a.config.AdvertiseAddrs.RPC != "" {
a.serverRpcAddr = a.config.AdvertiseAddrs.RPC
a.serverRPCAddr = a.config.AdvertiseAddrs.RPC
} else if a.config.Addresses.RPC != "" {
a.serverRpcAddr = fmt.Sprintf("%v:%v", a.config.Addresses.RPC, a.config.Ports.RPC)
a.serverRPCAddr = fmt.Sprintf("%v:%v", a.config.Addresses.RPC, a.config.Ports.RPC)
} else if a.config.BindAddr != "" {
a.serverRpcAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.RPC)
a.serverRPCAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.RPC)
} else {
a.serverRpcAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.RPC)
a.serverRPCAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.RPC)
}
addr, err = net.ResolveTCPAddr("tcp", a.serverRpcAddr)
addr, err = net.ResolveTCPAddr("tcp", a.serverRPCAddr)
if err != nil {
return nil, fmt.Errorf("error resolving RPC addr %q: %v:", a.serverRpcAddr, err)
return nil, fmt.Errorf("error resolving RPC addr %q: %v", a.serverRPCAddr, err)
}
a.serverRpcAddr = fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
a.serverRPCAddr = fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
// Resolve the Server's Serf Address
if a.config.AdvertiseAddrs.Serf != "" {
@@ -216,7 +216,7 @@ func (a *Agent) serverConfig() (*nomad.Config, error) {
}
addr, err = net.ResolveTCPAddr("tcp", a.serverSerfAddr)
if err != nil {
return nil, fmt.Errorf("error resolving Serf addr %q: %v:", a.serverSerfAddr, err)
return nil, fmt.Errorf("error resolving Serf addr %q: %v", a.serverSerfAddr, err)
}
a.serverSerfAddr = fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
@@ -292,37 +292,37 @@ func (a *Agent) clientConfig() (*clientconfig.Config, error) {
// Resolve the Client's HTTP address
if a.config.AdvertiseAddrs.HTTP != "" {
a.clientHttpAddr = a.config.AdvertiseAddrs.HTTP
a.clientHTTPAddr = a.config.AdvertiseAddrs.HTTP
} else if a.config.Addresses.HTTP != "" {
a.clientHttpAddr = fmt.Sprintf("%v:%v", a.config.Addresses.HTTP, a.config.Ports.HTTP)
a.clientHTTPAddr = fmt.Sprintf("%v:%v", a.config.Addresses.HTTP, a.config.Ports.HTTP)
} else if a.config.BindAddr != "" {
a.clientHttpAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.HTTP)
a.clientHTTPAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.HTTP)
} else {
a.clientHttpAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.HTTP)
a.clientHTTPAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.HTTP)
}
addr, err := net.ResolveTCPAddr("tcp", a.clientHttpAddr)
addr, err := net.ResolveTCPAddr("tcp", a.clientHTTPAddr)
if err != nil {
return nil, fmt.Errorf("error resolving HTTP addr %q: %v:", a.clientHttpAddr, err)
return nil, fmt.Errorf("error resolving HTTP addr %q: %v", a.clientHTTPAddr, err)
}
httpAddr := fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
conf.Node.HTTPAddr = httpAddr
a.clientHttpAddr = httpAddr
a.clientHTTPAddr = httpAddr
// Resolve the Client's RPC address
if a.config.AdvertiseAddrs.RPC != "" {
a.clientRpcAddr = a.config.AdvertiseAddrs.RPC
a.clientRPCAddr = a.config.AdvertiseAddrs.RPC
} else if a.config.Addresses.RPC != "" {
a.clientRpcAddr = fmt.Sprintf("%v:%v", a.config.Addresses.RPC, a.config.Ports.RPC)
a.clientRPCAddr = fmt.Sprintf("%v:%v", a.config.Addresses.RPC, a.config.Ports.RPC)
} else if a.config.BindAddr != "" {
a.clientRpcAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.RPC)
a.clientRPCAddr = fmt.Sprintf("%v:%v", a.config.BindAddr, a.config.Ports.RPC)
} else {
a.clientRpcAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.RPC)
a.clientRPCAddr = fmt.Sprintf("%v:%v", "127.0.0.1", a.config.Ports.RPC)
}
addr, err = net.ResolveTCPAddr("tcp", a.clientRpcAddr)
addr, err = net.ResolveTCPAddr("tcp", a.clientRPCAddr)
if err != nil {
return nil, fmt.Errorf("error resolving RPC addr %q: %v:", a.clientRpcAddr, err)
return nil, fmt.Errorf("error resolving RPC addr %q: %v", a.clientRPCAddr, err)
}
a.clientRpcAddr = fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
a.clientRPCAddr = fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port)
// Reserve resources on the node.
r := conf.Node.Reserved
@@ -372,13 +372,13 @@ func (a *Agent) setupServer() error {
a.consulSyncer.SetServices(serviceGroupName, []*structs.ConsulService{
&structs.ConsulService{
Name: a.config.Consul.ServerServiceName,
PortLabel: a.serverHttpAddr,
Tags: []string{consul.ServiceTagHttp},
PortLabel: a.serverHTTPAddr,
Tags: []string{consul.ServiceTagHTTP},
},
&structs.ConsulService{
Name: a.config.Consul.ServerServiceName,
PortLabel: a.serverRpcAddr,
Tags: []string{consul.ServiceTagRpc},
PortLabel: a.serverRPCAddr,
Tags: []string{consul.ServiceTagRPC},
},
&structs.ConsulService{
PortLabel: a.serverSerfAddr,
@@ -423,13 +423,13 @@ func (a *Agent) setupClient() error {
a.consulSyncer.SetServices(serviceGroupName, []*structs.ConsulService{
&structs.ConsulService{
Name: a.config.Consul.ClientServiceName,
PortLabel: a.clientHttpAddr,
Tags: []string{consul.ServiceTagHttp},
PortLabel: a.clientHTTPAddr,
Tags: []string{consul.ServiceTagHTTP},
},
&structs.ConsulService{
Name: a.config.Consul.ClientServiceName,
PortLabel: a.clientRpcAddr,
Tags: []string{consul.ServiceTagRpc},
PortLabel: a.clientRPCAddr,
Tags: []string{consul.ServiceTagRPC},
},
})
}

12
command/agent/agent_test.go
View File

@@ -119,10 +119,10 @@ func TestAgent_ServerConfig(t *testing.T) {
if addr := out.RPCAdvertise; addr.IP.String() != "127.0.0.1" || addr.Port != 4001 {
t.Fatalf("bad rpc advertise addr: %#v", addr)
}
if addr := a.serverHttpAddr; addr != "10.10.11.1:4005" {
if addr := a.serverHTTPAddr; addr != "10.10.11.1:4005" {
t.Fatalf("expect 10.11.11.1:4005, got: %v", addr)
}
if addr := a.serverRpcAddr; addr != "127.0.0.1:4001" {
if addr := a.serverRPCAddr; addr != "127.0.0.1:4001" {
t.Fatalf("expect 127.0.0.1:4001, got: %v", addr)
}
@@ -158,11 +158,11 @@ func TestAgent_ServerConfig(t *testing.T) {
if addr := out.SerfConfig.MemberlistConfig.BindAddr; addr != "127.0.0.2" {
t.Fatalf("expect 127.0.0.2, got: %s", addr)
}
if addr := a.serverHttpAddr; addr != "127.0.0.2:4646" {
if addr := a.serverHTTPAddr; addr != "127.0.0.2:4646" {
t.Fatalf("expect 127.0.0.2:4646, got: %s", addr)
}
// NOTE: AdvertiseAddr > Addresses > BindAddr > Defaults
if addr := a.serverRpcAddr; addr != "127.0.0.1:4001" {
if addr := a.serverRPCAddr; addr != "127.0.0.1:4001" {
t.Fatalf("expect 127.0.0.1:4001, got: %s", addr)
}
if addr := a.serverSerfAddr; addr != "127.0.0.1:4000" {
@@ -211,10 +211,10 @@ func TestAgent_ServerConfig(t *testing.T) {
if addr := out.SerfConfig.MemberlistConfig.BindAddr; addr != "127.0.0.3" {
t.Fatalf("expect 127.0.0.3, got: %s", addr)
}
if addr := a.serverHttpAddr; addr != "127.0.0.3:4646" {
if addr := a.serverHTTPAddr; addr != "127.0.0.3:4646" {
t.Fatalf("expect 127.0.0.3:4646, got: %s", addr)
}
if addr := a.serverRpcAddr; addr != "127.0.0.3:4647" {
if addr := a.serverRPCAddr; addr != "127.0.0.3:4647" {
t.Fatalf("expect 127.0.0.3:4647, got: %s", addr)
}
if addr := a.serverSerfAddr; addr != "127.0.0.3:4648" {

10
command/agent/consul/syncer.go
View File

@@ -47,11 +47,11 @@ const (
// spend waiting for a response from a Consul Query.
DefaultQueryWaitDuration = 2 * time.Second
// ServiceTagHttp is the tag assigned to HTTP services
ServiceTagHttp = "http"
// ServiceTagHTTP is the tag assigned to HTTP services
ServiceTagHTTP = "http"
// ServiceTagRpc is the tag assigned to RPC services
ServiceTagRpc = "rpc"
// ServiceTagRPC is the tag assigned to RPC services
ServiceTagRPC = "rpc"
// ServiceTagSerf is the tag assigned to Serf services
ServiceTagSerf = "serf"
@@ -895,6 +895,7 @@ func (c *Syncer) AddPeriodicHandler(name string, fn types.PeriodicCallback) bool
return true
}
// NumHandlers returns the number of callbacks registered with the syncer
func (c *Syncer) NumHandlers() int {
c.periodicLock.RLock()
defer c.periodicLock.RUnlock()
@@ -908,6 +909,7 @@ func (c *Syncer) RemovePeriodicHandler(name string) {
delete(c.periodicCallbacks, name)
}
// ConsulClient returns the Consul client used by the Syncer.
func (c *Syncer) ConsulClient() *consul.Client {
return c.client
}

6
nomad/node_endpoint.go
View File

@@ -126,9 +126,9 @@ func (n *Node) constructNodeServerInfoResponse(snap *state.StateSnapshot, reply
for k, v := range n.srv.localPeers {
reply.Servers = append(reply.Servers,
&structs.NodeServerInfo{
RpcAdvertiseAddr: k,
RpcMajorVersion: int32(v.MajorVersion),
RpcMinorVersion: int32(v.MinorVersion),
RPCAdvertiseAddr: k,
RPCMajorVersion: int32(v.MajorVersion),
RPCMinorVersion: int32(v.MinorVersion),
Datacenter: v.Datacenter,
})
}

6
nomad/structs/structs.go
View File

@@ -171,15 +171,15 @@ type NodeDeregisterRequest struct {
type NodeServerInfo struct {
// RPCAdvertiseAddr is the IP endpoint that a Nomad Server wishes to
// be contacted at for RPCs.
RpcAdvertiseAddr string
RPCAdvertiseAddr string
// RpcMajorVersion is the major version number the Nomad Server
// supports
RpcMajorVersion int32
RPCMajorVersion int32
// RpcMinorVersion is the minor version number the Nomad Server
// supports
RpcMinorVersion int32
RPCMinorVersion int32
// Datacenter is the datacenter that a Nomad server belongs to
Datacenter string