зеркало из https://github.com/microsoft/docker.git
Merge pull request #29787 from yongtang/29730-multiple-published-port
Fix issues of multiple published ports mapping to the same target port
This commit is contained in:
Коммит
82dfa6563f
|
@ -1572,3 +1572,22 @@ func (s *DockerSwarmSuite) TestSwarmServicePsMultipleServiceIDs(c *check.C) {
|
|||
c.Assert(out, checker.Contains, name2+".2")
|
||||
c.Assert(out, checker.Contains, name2+".3")
|
||||
}
|
||||
|
||||
func (s *DockerSwarmSuite) TestSwarmPublishDuplicatePorts(c *check.C) {
|
||||
d := s.AddDaemon(c, true, true)
|
||||
|
||||
out, err := d.Cmd("service", "create", "--publish", "5000:80", "--publish", "5001:80", "--publish", "80", "--publish", "80", "busybox", "top")
|
||||
c.Assert(err, check.IsNil, check.Commentf(out))
|
||||
id := strings.TrimSpace(out)
|
||||
|
||||
// make sure task has been deployed.
|
||||
waitAndAssert(c, defaultReconciliationTimeout, d.CheckActiveContainerCount, checker.Equals, 1)
|
||||
|
||||
// Total len = 4, with 2 dynamic ports and 2 non-dynamic ports
|
||||
// Dynamic ports are likely to be 30000 and 30001 but doesn't matter
|
||||
out, err = d.Cmd("service", "inspect", "--format", "{{.Endpoint.Ports}} len={{len .Endpoint.Ports}}", id)
|
||||
c.Assert(err, check.IsNil, check.Commentf(out))
|
||||
c.Assert(out, checker.Contains, "len=4")
|
||||
c.Assert(out, checker.Contains, "{ tcp 80 5000 ingress}")
|
||||
c.Assert(out, checker.Contains, "{ tcp 80 5001 ingress}")
|
||||
}
|
||||
|
|
|
@ -103,7 +103,7 @@ github.com/docker/containerd 03e5862ec0d8d3b3f750e19fca3ee367e13c090e
|
|||
github.com/tonistiigi/fifo 1405643975692217d6720f8b54aeee1bf2cd5cf4
|
||||
|
||||
# cluster
|
||||
github.com/docker/swarmkit 4762d92234d286ae7c9e061470485e4d34ef8ebd
|
||||
github.com/docker/swarmkit c97146840a26c9ce8023284d0e9c989586cc1857
|
||||
github.com/golang/mock bd3c8e81be01eef76d4b503f5e687d2d1354d2d9
|
||||
github.com/gogo/protobuf v0.3
|
||||
github.com/cloudflare/cfssl 7fb22c8cba7ecaf98e4082d22d65800cf45e042a
|
||||
|
|
|
@ -836,12 +836,12 @@ func encodeVarintCa(data []byte, offset int, v uint64) int {
|
|||
}
|
||||
|
||||
type raftProxyCAServer struct {
|
||||
local CAServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local CAServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyCAServer(local CAServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) CAServer {
|
||||
func NewRaftProxyCAServer(local CAServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) CAServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -858,18 +858,24 @@ func NewRaftProxyCAServer(local CAServer, connSelector raftselector.ConnProvider
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyCAServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyCAServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyCAServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -906,11 +912,15 @@ func (p *raftProxyCAServer) GetRootCACertificate(ctx context.Context, r *GetRoot
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetRootCACertificate(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -937,11 +947,15 @@ func (p *raftProxyCAServer) GetUnlockKey(ctx context.Context, r *GetUnlockKeyReq
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetUnlockKey(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -964,12 +978,12 @@ func (p *raftProxyCAServer) GetUnlockKey(ctx context.Context, r *GetUnlockKeyReq
|
|||
}
|
||||
|
||||
type raftProxyNodeCAServer struct {
|
||||
local NodeCAServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local NodeCAServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyNodeCAServer(local NodeCAServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) NodeCAServer {
|
||||
func NewRaftProxyNodeCAServer(local NodeCAServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) NodeCAServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -986,18 +1000,24 @@ func NewRaftProxyNodeCAServer(local NodeCAServer, connSelector raftselector.Conn
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyNodeCAServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyNodeCAServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyNodeCAServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -1034,11 +1054,15 @@ func (p *raftProxyNodeCAServer) IssueNodeCertificate(ctx context.Context, r *Iss
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.IssueNodeCertificate(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -1065,11 +1089,15 @@ func (p *raftProxyNodeCAServer) NodeCertificateStatus(ctx context.Context, r *No
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.NodeCertificateStatus(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
@ -5256,12 +5256,12 @@ func encodeVarintControl(data []byte, offset int, v uint64) int {
|
|||
}
|
||||
|
||||
type raftProxyControlServer struct {
|
||||
local ControlServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local ControlServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyControlServer(local ControlServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) ControlServer {
|
||||
func NewRaftProxyControlServer(local ControlServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) ControlServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -5278,18 +5278,24 @@ func NewRaftProxyControlServer(local ControlServer, connSelector raftselector.Co
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyControlServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyControlServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyControlServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -5326,11 +5332,15 @@ func (p *raftProxyControlServer) GetNode(ctx context.Context, r *GetNodeRequest)
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetNode(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5357,11 +5367,15 @@ func (p *raftProxyControlServer) ListNodes(ctx context.Context, r *ListNodesRequ
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ListNodes(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5388,11 +5402,15 @@ func (p *raftProxyControlServer) UpdateNode(ctx context.Context, r *UpdateNodeRe
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.UpdateNode(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5419,11 +5437,15 @@ func (p *raftProxyControlServer) RemoveNode(ctx context.Context, r *RemoveNodeRe
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.RemoveNode(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5450,11 +5472,15 @@ func (p *raftProxyControlServer) GetTask(ctx context.Context, r *GetTaskRequest)
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetTask(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5481,11 +5507,15 @@ func (p *raftProxyControlServer) ListTasks(ctx context.Context, r *ListTasksRequ
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ListTasks(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5512,11 +5542,15 @@ func (p *raftProxyControlServer) RemoveTask(ctx context.Context, r *RemoveTaskRe
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.RemoveTask(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5543,11 +5577,15 @@ func (p *raftProxyControlServer) GetService(ctx context.Context, r *GetServiceRe
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetService(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5574,11 +5612,15 @@ func (p *raftProxyControlServer) ListServices(ctx context.Context, r *ListServic
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ListServices(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5605,11 +5647,15 @@ func (p *raftProxyControlServer) CreateService(ctx context.Context, r *CreateSer
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.CreateService(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5636,11 +5682,15 @@ func (p *raftProxyControlServer) UpdateService(ctx context.Context, r *UpdateSer
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.UpdateService(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5667,11 +5717,15 @@ func (p *raftProxyControlServer) RemoveService(ctx context.Context, r *RemoveSer
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.RemoveService(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5698,11 +5752,15 @@ func (p *raftProxyControlServer) GetNetwork(ctx context.Context, r *GetNetworkRe
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetNetwork(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5729,11 +5787,15 @@ func (p *raftProxyControlServer) ListNetworks(ctx context.Context, r *ListNetwor
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ListNetworks(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5760,11 +5822,15 @@ func (p *raftProxyControlServer) CreateNetwork(ctx context.Context, r *CreateNet
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.CreateNetwork(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5791,11 +5857,15 @@ func (p *raftProxyControlServer) RemoveNetwork(ctx context.Context, r *RemoveNet
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.RemoveNetwork(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5822,11 +5892,15 @@ func (p *raftProxyControlServer) GetCluster(ctx context.Context, r *GetClusterRe
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetCluster(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5853,11 +5927,15 @@ func (p *raftProxyControlServer) ListClusters(ctx context.Context, r *ListCluste
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ListClusters(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5884,11 +5962,15 @@ func (p *raftProxyControlServer) UpdateCluster(ctx context.Context, r *UpdateClu
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.UpdateCluster(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5915,11 +5997,15 @@ func (p *raftProxyControlServer) GetSecret(ctx context.Context, r *GetSecretRequ
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.GetSecret(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5946,11 +6032,15 @@ func (p *raftProxyControlServer) UpdateSecret(ctx context.Context, r *UpdateSecr
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.UpdateSecret(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -5977,11 +6067,15 @@ func (p *raftProxyControlServer) ListSecrets(ctx context.Context, r *ListSecrets
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ListSecrets(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -6008,11 +6102,15 @@ func (p *raftProxyControlServer) CreateSecret(ctx context.Context, r *CreateSecr
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.CreateSecret(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -6039,11 +6137,15 @@ func (p *raftProxyControlServer) RemoveSecret(ctx context.Context, r *RemoveSecr
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.RemoveSecret(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
@ -1670,12 +1670,12 @@ func encodeVarintDispatcher(data []byte, offset int, v uint64) int {
|
|||
}
|
||||
|
||||
type raftProxyDispatcherServer struct {
|
||||
local DispatcherServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local DispatcherServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyDispatcherServer(local DispatcherServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) DispatcherServer {
|
||||
func NewRaftProxyDispatcherServer(local DispatcherServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) DispatcherServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -1692,18 +1692,24 @@ func NewRaftProxyDispatcherServer(local DispatcherServer, connSelector raftselec
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyDispatcherServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyDispatcherServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyDispatcherServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -1735,17 +1741,33 @@ func (p *raftProxyDispatcherServer) pollNewLeaderConn(ctx context.Context) (*grp
|
|||
}
|
||||
}
|
||||
|
||||
func (p *raftProxyDispatcherServer) Session(r *SessionRequest, stream Dispatcher_SessionServer) error {
|
||||
type Dispatcher_SessionServerWrapper struct {
|
||||
Dispatcher_SessionServer
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s Dispatcher_SessionServerWrapper) Context() context.Context {
|
||||
return s.ctx
|
||||
}
|
||||
|
||||
func (p *raftProxyDispatcherServer) Session(r *SessionRequest, stream Dispatcher_SessionServer) error {
|
||||
ctx := stream.Context()
|
||||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
return p.local.Session(r, stream)
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
streamWrapper := Dispatcher_SessionServerWrapper{
|
||||
Dispatcher_SessionServer: stream,
|
||||
ctx: ctx,
|
||||
}
|
||||
return p.local.Session(r, streamWrapper)
|
||||
}
|
||||
return err
|
||||
}
|
||||
ctx, err = p.runCtxMods(ctx)
|
||||
ctx, err = p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -1775,11 +1797,15 @@ func (p *raftProxyDispatcherServer) Heartbeat(ctx context.Context, r *HeartbeatR
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.Heartbeat(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -1806,11 +1832,15 @@ func (p *raftProxyDispatcherServer) UpdateTaskStatus(ctx context.Context, r *Upd
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.UpdateTaskStatus(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -1832,17 +1862,33 @@ func (p *raftProxyDispatcherServer) UpdateTaskStatus(ctx context.Context, r *Upd
|
|||
return resp, err
|
||||
}
|
||||
|
||||
func (p *raftProxyDispatcherServer) Tasks(r *TasksRequest, stream Dispatcher_TasksServer) error {
|
||||
type Dispatcher_TasksServerWrapper struct {
|
||||
Dispatcher_TasksServer
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s Dispatcher_TasksServerWrapper) Context() context.Context {
|
||||
return s.ctx
|
||||
}
|
||||
|
||||
func (p *raftProxyDispatcherServer) Tasks(r *TasksRequest, stream Dispatcher_TasksServer) error {
|
||||
ctx := stream.Context()
|
||||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
return p.local.Tasks(r, stream)
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
streamWrapper := Dispatcher_TasksServerWrapper{
|
||||
Dispatcher_TasksServer: stream,
|
||||
ctx: ctx,
|
||||
}
|
||||
return p.local.Tasks(r, streamWrapper)
|
||||
}
|
||||
return err
|
||||
}
|
||||
ctx, err = p.runCtxMods(ctx)
|
||||
ctx, err = p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -1867,17 +1913,33 @@ func (p *raftProxyDispatcherServer) Tasks(r *TasksRequest, stream Dispatcher_Tas
|
|||
return nil
|
||||
}
|
||||
|
||||
func (p *raftProxyDispatcherServer) Assignments(r *AssignmentsRequest, stream Dispatcher_AssignmentsServer) error {
|
||||
type Dispatcher_AssignmentsServerWrapper struct {
|
||||
Dispatcher_AssignmentsServer
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s Dispatcher_AssignmentsServerWrapper) Context() context.Context {
|
||||
return s.ctx
|
||||
}
|
||||
|
||||
func (p *raftProxyDispatcherServer) Assignments(r *AssignmentsRequest, stream Dispatcher_AssignmentsServer) error {
|
||||
ctx := stream.Context()
|
||||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
return p.local.Assignments(r, stream)
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
streamWrapper := Dispatcher_AssignmentsServerWrapper{
|
||||
Dispatcher_AssignmentsServer: stream,
|
||||
ctx: ctx,
|
||||
}
|
||||
return p.local.Assignments(r, streamWrapper)
|
||||
}
|
||||
return err
|
||||
}
|
||||
ctx, err = p.runCtxMods(ctx)
|
||||
ctx, err = p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
@ -321,12 +321,12 @@ func encodeVarintHealth(data []byte, offset int, v uint64) int {
|
|||
}
|
||||
|
||||
type raftProxyHealthServer struct {
|
||||
local HealthServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local HealthServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyHealthServer(local HealthServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) HealthServer {
|
||||
func NewRaftProxyHealthServer(local HealthServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) HealthServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -343,18 +343,24 @@ func NewRaftProxyHealthServer(local HealthServer, connSelector raftselector.Conn
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyHealthServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyHealthServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyHealthServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -391,11 +397,15 @@ func (p *raftProxyHealthServer) Check(ctx context.Context, r *HealthCheckRequest
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.Check(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
@ -1279,12 +1279,12 @@ func encodeVarintLogbroker(data []byte, offset int, v uint64) int {
|
|||
}
|
||||
|
||||
type raftProxyLogsServer struct {
|
||||
local LogsServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local LogsServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyLogsServer(local LogsServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) LogsServer {
|
||||
func NewRaftProxyLogsServer(local LogsServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) LogsServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -1301,18 +1301,24 @@ func NewRaftProxyLogsServer(local LogsServer, connSelector raftselector.ConnProv
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyLogsServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyLogsServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyLogsServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -1344,17 +1350,33 @@ func (p *raftProxyLogsServer) pollNewLeaderConn(ctx context.Context) (*grpc.Clie
|
|||
}
|
||||
}
|
||||
|
||||
func (p *raftProxyLogsServer) SubscribeLogs(r *SubscribeLogsRequest, stream Logs_SubscribeLogsServer) error {
|
||||
type Logs_SubscribeLogsServerWrapper struct {
|
||||
Logs_SubscribeLogsServer
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s Logs_SubscribeLogsServerWrapper) Context() context.Context {
|
||||
return s.ctx
|
||||
}
|
||||
|
||||
func (p *raftProxyLogsServer) SubscribeLogs(r *SubscribeLogsRequest, stream Logs_SubscribeLogsServer) error {
|
||||
ctx := stream.Context()
|
||||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
return p.local.SubscribeLogs(r, stream)
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
streamWrapper := Logs_SubscribeLogsServerWrapper{
|
||||
Logs_SubscribeLogsServer: stream,
|
||||
ctx: ctx,
|
||||
}
|
||||
return p.local.SubscribeLogs(r, streamWrapper)
|
||||
}
|
||||
return err
|
||||
}
|
||||
ctx, err = p.runCtxMods(ctx)
|
||||
ctx, err = p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -1380,12 +1402,12 @@ func (p *raftProxyLogsServer) SubscribeLogs(r *SubscribeLogsRequest, stream Logs
|
|||
}
|
||||
|
||||
type raftProxyLogBrokerServer struct {
|
||||
local LogBrokerServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local LogBrokerServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyLogBrokerServer(local LogBrokerServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) LogBrokerServer {
|
||||
func NewRaftProxyLogBrokerServer(local LogBrokerServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) LogBrokerServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -1402,18 +1424,24 @@ func NewRaftProxyLogBrokerServer(local LogBrokerServer, connSelector raftselecto
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyLogBrokerServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyLogBrokerServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyLogBrokerServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -1445,17 +1473,33 @@ func (p *raftProxyLogBrokerServer) pollNewLeaderConn(ctx context.Context) (*grpc
|
|||
}
|
||||
}
|
||||
|
||||
func (p *raftProxyLogBrokerServer) ListenSubscriptions(r *ListenSubscriptionsRequest, stream LogBroker_ListenSubscriptionsServer) error {
|
||||
type LogBroker_ListenSubscriptionsServerWrapper struct {
|
||||
LogBroker_ListenSubscriptionsServer
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s LogBroker_ListenSubscriptionsServerWrapper) Context() context.Context {
|
||||
return s.ctx
|
||||
}
|
||||
|
||||
func (p *raftProxyLogBrokerServer) ListenSubscriptions(r *ListenSubscriptionsRequest, stream LogBroker_ListenSubscriptionsServer) error {
|
||||
ctx := stream.Context()
|
||||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
return p.local.ListenSubscriptions(r, stream)
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
streamWrapper := LogBroker_ListenSubscriptionsServerWrapper{
|
||||
LogBroker_ListenSubscriptionsServer: stream,
|
||||
ctx: ctx,
|
||||
}
|
||||
return p.local.ListenSubscriptions(r, streamWrapper)
|
||||
}
|
||||
return err
|
||||
}
|
||||
ctx, err = p.runCtxMods(ctx)
|
||||
ctx, err = p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
@ -1480,17 +1524,33 @@ func (p *raftProxyLogBrokerServer) ListenSubscriptions(r *ListenSubscriptionsReq
|
|||
return nil
|
||||
}
|
||||
|
||||
func (p *raftProxyLogBrokerServer) PublishLogs(stream LogBroker_PublishLogsServer) error {
|
||||
type LogBroker_PublishLogsServerWrapper struct {
|
||||
LogBroker_PublishLogsServer
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s LogBroker_PublishLogsServerWrapper) Context() context.Context {
|
||||
return s.ctx
|
||||
}
|
||||
|
||||
func (p *raftProxyLogBrokerServer) PublishLogs(stream LogBroker_PublishLogsServer) error {
|
||||
ctx := stream.Context()
|
||||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
return p.local.PublishLogs(stream)
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
streamWrapper := LogBroker_PublishLogsServerWrapper{
|
||||
LogBroker_PublishLogsServer: stream,
|
||||
ctx: ctx,
|
||||
}
|
||||
return p.local.PublishLogs(streamWrapper)
|
||||
}
|
||||
return err
|
||||
}
|
||||
ctx, err = p.runCtxMods(ctx)
|
||||
ctx, err = p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
@ -1498,12 +1498,12 @@ func encodeVarintRaft(data []byte, offset int, v uint64) int {
|
|||
}
|
||||
|
||||
type raftProxyRaftServer struct {
|
||||
local RaftServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local RaftServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyRaftServer(local RaftServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) RaftServer {
|
||||
func NewRaftProxyRaftServer(local RaftServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) RaftServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -1520,18 +1520,24 @@ func NewRaftProxyRaftServer(local RaftServer, connSelector raftselector.ConnProv
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyRaftServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyRaftServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyRaftServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -1568,11 +1574,15 @@ func (p *raftProxyRaftServer) ProcessRaftMessage(ctx context.Context, r *Process
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ProcessRaftMessage(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -1599,11 +1609,15 @@ func (p *raftProxyRaftServer) ResolveAddress(ctx context.Context, r *ResolveAddr
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.ResolveAddress(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -1626,12 +1640,12 @@ func (p *raftProxyRaftServer) ResolveAddress(ctx context.Context, r *ResolveAddr
|
|||
}
|
||||
|
||||
type raftProxyRaftMembershipServer struct {
|
||||
local RaftMembershipServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local RaftMembershipServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyRaftMembershipServer(local RaftMembershipServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) RaftMembershipServer {
|
||||
func NewRaftProxyRaftMembershipServer(local RaftMembershipServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) RaftMembershipServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -1648,18 +1662,24 @@ func NewRaftProxyRaftMembershipServer(local RaftMembershipServer, connSelector r
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyRaftMembershipServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyRaftMembershipServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyRaftMembershipServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -1696,11 +1716,15 @@ func (p *raftProxyRaftMembershipServer) Join(ctx context.Context, r *JoinRequest
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.Join(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -1727,11 +1751,15 @@ func (p *raftProxyRaftMembershipServer) Leave(ctx context.Context, r *LeaveReque
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.Leave(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
@ -451,12 +451,12 @@ func encodeVarintResource(data []byte, offset int, v uint64) int {
|
|||
}
|
||||
|
||||
type raftProxyResourceAllocatorServer struct {
|
||||
local ResourceAllocatorServer
|
||||
connSelector raftselector.ConnProvider
|
||||
ctxMods []func(context.Context) (context.Context, error)
|
||||
local ResourceAllocatorServer
|
||||
connSelector raftselector.ConnProvider
|
||||
localCtxMods, remoteCtxMods []func(context.Context) (context.Context, error)
|
||||
}
|
||||
|
||||
func NewRaftProxyResourceAllocatorServer(local ResourceAllocatorServer, connSelector raftselector.ConnProvider, ctxMod func(context.Context) (context.Context, error)) ResourceAllocatorServer {
|
||||
func NewRaftProxyResourceAllocatorServer(local ResourceAllocatorServer, connSelector raftselector.ConnProvider, localCtxMod, remoteCtxMod func(context.Context) (context.Context, error)) ResourceAllocatorServer {
|
||||
redirectChecker := func(ctx context.Context) (context.Context, error) {
|
||||
s, ok := transport.StreamFromContext(ctx)
|
||||
if !ok {
|
||||
|
@ -473,18 +473,24 @@ func NewRaftProxyResourceAllocatorServer(local ResourceAllocatorServer, connSele
|
|||
md["redirect"] = append(md["redirect"], addr)
|
||||
return metadata.NewContext(ctx, md), nil
|
||||
}
|
||||
mods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
mods = append(mods, ctxMod)
|
||||
remoteMods := []func(context.Context) (context.Context, error){redirectChecker}
|
||||
remoteMods = append(remoteMods, remoteCtxMod)
|
||||
|
||||
var localMods []func(context.Context) (context.Context, error)
|
||||
if localCtxMod != nil {
|
||||
localMods = []func(context.Context) (context.Context, error){localCtxMod}
|
||||
}
|
||||
|
||||
return &raftProxyResourceAllocatorServer{
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
ctxMods: mods,
|
||||
local: local,
|
||||
connSelector: connSelector,
|
||||
localCtxMods: localMods,
|
||||
remoteCtxMods: remoteMods,
|
||||
}
|
||||
}
|
||||
func (p *raftProxyResourceAllocatorServer) runCtxMods(ctx context.Context) (context.Context, error) {
|
||||
func (p *raftProxyResourceAllocatorServer) runCtxMods(ctx context.Context, ctxMods []func(context.Context) (context.Context, error)) (context.Context, error) {
|
||||
var err error
|
||||
for _, mod := range p.ctxMods {
|
||||
for _, mod := range ctxMods {
|
||||
ctx, err = mod(ctx)
|
||||
if err != nil {
|
||||
return ctx, err
|
||||
|
@ -521,11 +527,15 @@ func (p *raftProxyResourceAllocatorServer) AttachNetwork(ctx context.Context, r
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.AttachNetwork(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -552,11 +562,15 @@ func (p *raftProxyResourceAllocatorServer) DetachNetwork(ctx context.Context, r
|
|||
conn, err := p.connSelector.LeaderConn(ctx)
|
||||
if err != nil {
|
||||
if err == raftselector.ErrIsLeader {
|
||||
ctx, err = p.runCtxMods(ctx, p.localCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.local.DetachNetwork(ctx, r)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
modCtx, err := p.runCtxMods(ctx)
|
||||
modCtx, err := p.runCtxMods(ctx, p.remoteCtxMods)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
@ -16,6 +16,13 @@ import (
|
|||
"google.golang.org/grpc/peer"
|
||||
)
|
||||
|
||||
type localRequestKeyType struct{}
|
||||
|
||||
// LocalRequestKey is a context key to mark a request that originating on the
|
||||
// local node. The assocated value is a RemoteNodeInfo structure describing the
|
||||
// local node.
|
||||
var LocalRequestKey = localRequestKeyType{}
|
||||
|
||||
// LogTLSState logs information about the TLS connection and remote peers
|
||||
func LogTLSState(ctx context.Context, tlsState *tls.ConnectionState) {
|
||||
if tlsState == nil {
|
||||
|
@ -189,6 +196,17 @@ type RemoteNodeInfo struct {
|
|||
// well as the forwarder's ID. This function does not do authorization checks -
|
||||
// it only looks up the node ID.
|
||||
func RemoteNode(ctx context.Context) (RemoteNodeInfo, error) {
|
||||
// If we have a value on the context that marks this as a local
|
||||
// request, we return the node info from the context.
|
||||
localNodeInfo := ctx.Value(LocalRequestKey)
|
||||
|
||||
if localNodeInfo != nil {
|
||||
nodeInfo, ok := localNodeInfo.(RemoteNodeInfo)
|
||||
if ok {
|
||||
return nodeInfo, nil
|
||||
}
|
||||
}
|
||||
|
||||
certSubj, err := certSubjectFromContext(ctx)
|
||||
if err != nil {
|
||||
return RemoteNodeInfo{}, err
|
||||
|
|
|
@ -28,7 +28,6 @@ const (
|
|||
// breaking it apart doesn't seem worth it.
|
||||
type Server struct {
|
||||
mu sync.Mutex
|
||||
wg sync.WaitGroup
|
||||
ctx context.Context
|
||||
cancel func()
|
||||
store *store.MemoryStore
|
||||
|
@ -102,10 +101,9 @@ func (s *Server) NodeCertificateStatus(ctx context.Context, request *api.NodeCer
|
|||
return nil, grpc.Errorf(codes.InvalidArgument, codes.InvalidArgument.String())
|
||||
}
|
||||
|
||||
if err := s.addTask(); err != nil {
|
||||
if err := s.isRunningLocked(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer s.doneTask()
|
||||
|
||||
var node *api.Node
|
||||
|
||||
|
@ -189,10 +187,9 @@ func (s *Server) IssueNodeCertificate(ctx context.Context, request *api.IssueNod
|
|||
return nil, grpc.Errorf(codes.InvalidArgument, codes.InvalidArgument.String())
|
||||
}
|
||||
|
||||
if err := s.addTask(); err != nil {
|
||||
if err := s.isRunningLocked(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer s.doneTask()
|
||||
|
||||
var (
|
||||
blacklistedCerts map[string]*api.BlacklistedCertificate
|
||||
|
@ -211,6 +208,15 @@ func (s *Server) IssueNodeCertificate(ctx context.Context, request *api.IssueNod
|
|||
blacklistedCerts = clusters[0].BlacklistedCertificates
|
||||
}
|
||||
|
||||
// Renewing the cert with a local (unix socket) is always valid.
|
||||
localNodeInfo := ctx.Value(LocalRequestKey)
|
||||
if localNodeInfo != nil {
|
||||
nodeInfo, ok := localNodeInfo.(RemoteNodeInfo)
|
||||
if ok && nodeInfo.NodeID != "" {
|
||||
return s.issueRenewCertificate(ctx, nodeInfo.NodeID, request.CSR)
|
||||
}
|
||||
}
|
||||
|
||||
// If the remote node is a worker (either forwarded by a manager, or calling directly),
|
||||
// issue a renew worker certificate entry with the correct ID
|
||||
nodeID, err := AuthorizeForwardedRoleAndOrg(ctx, []string{WorkerRole}, []string{ManagerRole}, s.securityConfig.ClientTLSCreds.Organization(), blacklistedCerts)
|
||||
|
@ -365,10 +371,8 @@ func (s *Server) Run(ctx context.Context) error {
|
|||
s.mu.Unlock()
|
||||
return errors.New("CA signer is already running")
|
||||
}
|
||||
s.wg.Add(1)
|
||||
s.mu.Unlock()
|
||||
|
||||
defer s.wg.Done()
|
||||
ctx = log.WithModule(ctx, "ca")
|
||||
|
||||
// Retrieve the channels to keep track of changes in the cluster
|
||||
|
@ -398,8 +402,8 @@ func (s *Server) Run(ctx context.Context) error {
|
|||
// returns true without joinTokens being set correctly.
|
||||
s.mu.Lock()
|
||||
s.ctx, s.cancel = context.WithCancel(ctx)
|
||||
s.mu.Unlock()
|
||||
close(s.started)
|
||||
s.mu.Unlock()
|
||||
|
||||
if err != nil {
|
||||
log.G(ctx).WithFields(logrus.Fields{
|
||||
|
@ -460,38 +464,32 @@ func (s *Server) Run(ctx context.Context) error {
|
|||
// Stop stops the CA and closes all grpc streams.
|
||||
func (s *Server) Stop() error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if !s.isRunning() {
|
||||
s.mu.Unlock()
|
||||
return errors.New("CA signer is already stopped")
|
||||
}
|
||||
s.cancel()
|
||||
s.mu.Unlock()
|
||||
// wait for all handlers to finish their CA deals,
|
||||
s.wg.Wait()
|
||||
s.started = make(chan struct{})
|
||||
return nil
|
||||
}
|
||||
|
||||
// Ready waits on the ready channel and returns when the server is ready to serve.
|
||||
func (s *Server) Ready() <-chan struct{} {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.started
|
||||
}
|
||||
|
||||
func (s *Server) addTask() error {
|
||||
func (s *Server) isRunningLocked() error {
|
||||
s.mu.Lock()
|
||||
if !s.isRunning() {
|
||||
s.mu.Unlock()
|
||||
return grpc.Errorf(codes.Aborted, "CA signer is stopped")
|
||||
}
|
||||
s.wg.Add(1)
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) doneTask() {
|
||||
s.wg.Done()
|
||||
}
|
||||
|
||||
func (s *Server) isRunning() bool {
|
||||
if s.ctx == nil {
|
||||
return false
|
||||
|
|
155
vendor/github.com/docker/swarmkit/manager/allocator/networkallocator/portallocator.go
сгенерированный
поставляемый
155
vendor/github.com/docker/swarmkit/manager/allocator/networkallocator/portallocator.go
сгенерированный
поставляемый
|
@ -38,6 +38,71 @@ type portSpace struct {
|
|||
dynamicPortSpace *idm.Idm
|
||||
}
|
||||
|
||||
type allocatedPorts map[api.PortConfig]map[uint32]*api.PortConfig
|
||||
|
||||
// addState add the state of an allocated port to the collection.
|
||||
// `allocatedPorts` is a map of portKey:publishedPort:portState.
|
||||
// In case the value of the portKey is missing, the map
|
||||
// publishedPort:portState is created automatically
|
||||
func (ps allocatedPorts) addState(p *api.PortConfig) {
|
||||
portKey := getPortConfigKey(p)
|
||||
if _, ok := ps[portKey]; !ok {
|
||||
ps[portKey] = make(map[uint32]*api.PortConfig)
|
||||
}
|
||||
ps[portKey][p.PublishedPort] = p
|
||||
}
|
||||
|
||||
// delState delete the state of an allocated port from the collection.
|
||||
// `allocatedPorts` is a map of portKey:publishedPort:portState.
|
||||
//
|
||||
// If publishedPort is non-zero, then it is user defined. We will try to
|
||||
// remove the portState from `allocatedPorts` directly and return
|
||||
// the portState (or nil if no portState exists)
|
||||
//
|
||||
// If publishedPort is zero, then it is dynamically allocated. We will try
|
||||
// to remove the portState from `allocatedPorts`, as long as there is
|
||||
// a portState associated with a non-zero publishedPort.
|
||||
// Note multiple dynamically allocated ports might exists. In this case,
|
||||
// we will remove only at a time so both allocated ports are tracked.
|
||||
//
|
||||
// Note becasue of the potential co-existence of user-defined and dynamically
|
||||
// allocated ports, delState has to be called for user-defined port first.
|
||||
// dynamically allocated ports should be removed later.
|
||||
func (ps allocatedPorts) delState(p *api.PortConfig) *api.PortConfig {
|
||||
portKey := getPortConfigKey(p)
|
||||
|
||||
portStateMap, ok := ps[portKey]
|
||||
|
||||
// If name, port, protocol values don't match then we
|
||||
// are not allocated.
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
|
||||
if p.PublishedPort != 0 {
|
||||
// If SwarmPort was user defined but the port state
|
||||
// SwarmPort doesn't match we are not allocated.
|
||||
v := portStateMap[p.PublishedPort]
|
||||
|
||||
// Delete state from allocatedPorts
|
||||
delete(portStateMap, p.PublishedPort)
|
||||
|
||||
return v
|
||||
}
|
||||
|
||||
// If PublishedPort == 0 and we don't have non-zero port
|
||||
// then we are not allocated
|
||||
for publishedPort, v := range portStateMap {
|
||||
if publishedPort != 0 {
|
||||
// Delete state from allocatedPorts
|
||||
delete(portStateMap, publishedPort)
|
||||
return v
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func newPortAllocator() (*portAllocator, error) {
|
||||
portSpaces := make(map[api.PortConfig_Protocol]*portSpace)
|
||||
for _, protocol := range []api.PortConfig_Protocol{api.ProtocolTCP, api.ProtocolUDP} {
|
||||
|
@ -91,40 +156,53 @@ func reconcilePortConfigs(s *api.Service) []*api.PortConfig {
|
|||
return s.Spec.Endpoint.Ports
|
||||
}
|
||||
|
||||
allocatedPorts := make(map[api.PortConfig]*api.PortConfig)
|
||||
portStates := allocatedPorts{}
|
||||
for _, portState := range s.Endpoint.Ports {
|
||||
if portState.PublishMode != api.PublishModeIngress {
|
||||
continue
|
||||
if portState.PublishMode == api.PublishModeIngress {
|
||||
portStates.addState(portState)
|
||||
}
|
||||
|
||||
allocatedPorts[getPortConfigKey(portState)] = portState
|
||||
}
|
||||
|
||||
var portConfigs []*api.PortConfig
|
||||
|
||||
// Process the portConfig with portConfig.PublishMode != api.PublishModeIngress
|
||||
// and PublishedPort != 0 (high priority)
|
||||
for _, portConfig := range s.Spec.Endpoint.Ports {
|
||||
// If the PublishMode is not Ingress simply pick up
|
||||
// the port config.
|
||||
if portConfig.PublishMode != api.PublishModeIngress {
|
||||
// If the PublishMode is not Ingress simply pick up the port config.
|
||||
portConfigs = append(portConfigs, portConfig)
|
||||
} else if portConfig.PublishedPort != 0 {
|
||||
// Otherwise we only process PublishedPort != 0 in this round
|
||||
|
||||
// Remove record from portState
|
||||
portStates.delState(portConfig)
|
||||
|
||||
// For PublishedPort != 0 prefer the portConfig
|
||||
portConfigs = append(portConfigs, portConfig)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
portState, ok := allocatedPorts[getPortConfigKey(portConfig)]
|
||||
// Iterate portConfigs with PublishedPort == 0 (low priority)
|
||||
for _, portConfig := range s.Spec.Endpoint.Ports {
|
||||
// Ignore ports which are not PublishModeIngress (already processed)
|
||||
// And we only process PublishedPort == 0 in this round
|
||||
// So the following:
|
||||
// `portConfig.PublishMode == api.PublishModeIngress && portConfig.PublishedPort == 0`
|
||||
if portConfig.PublishMode == api.PublishModeIngress && portConfig.PublishedPort == 0 {
|
||||
// If the portConfig is exactly the same as portState
|
||||
// except if SwarmPort is not user-define then prefer
|
||||
// portState to ensure sticky allocation of the same
|
||||
// port that was allocated before.
|
||||
|
||||
// If the portConfig is exactly the same as portState
|
||||
// except if SwarmPort is not user-define then prefer
|
||||
// portState to ensure sticky allocation of the same
|
||||
// port that was allocated before.
|
||||
if ok && portConfig.Name == portState.Name &&
|
||||
portConfig.TargetPort == portState.TargetPort &&
|
||||
portConfig.Protocol == portState.Protocol &&
|
||||
portConfig.PublishedPort == 0 {
|
||||
portConfigs = append(portConfigs, portState)
|
||||
continue
|
||||
// Remove record from portState
|
||||
if portState := portStates.delState(portConfig); portState != nil {
|
||||
portConfigs = append(portConfigs, portState)
|
||||
continue
|
||||
}
|
||||
|
||||
// For all other cases prefer the portConfig
|
||||
portConfigs = append(portConfigs, portConfig)
|
||||
}
|
||||
|
||||
// For all other cases prefer the portConfig
|
||||
portConfigs = append(portConfigs, portConfig)
|
||||
}
|
||||
|
||||
return portConfigs
|
||||
|
@ -213,40 +291,31 @@ func (pa *portAllocator) isPortsAllocated(s *api.Service) bool {
|
|||
return false
|
||||
}
|
||||
|
||||
allocatedPorts := make(map[api.PortConfig]*api.PortConfig)
|
||||
portStates := allocatedPorts{}
|
||||
for _, portState := range s.Endpoint.Ports {
|
||||
if portState.PublishMode != api.PublishModeIngress {
|
||||
continue
|
||||
if portState.PublishMode == api.PublishModeIngress {
|
||||
portStates.addState(portState)
|
||||
}
|
||||
|
||||
allocatedPorts[getPortConfigKey(portState)] = portState
|
||||
}
|
||||
|
||||
// Iterate portConfigs with PublishedPort != 0 (high priority)
|
||||
for _, portConfig := range s.Spec.Endpoint.Ports {
|
||||
// Ignore ports which are not PublishModeIngress
|
||||
if portConfig.PublishMode != api.PublishModeIngress {
|
||||
continue
|
||||
}
|
||||
|
||||
portState, ok := allocatedPorts[getPortConfigKey(portConfig)]
|
||||
|
||||
// If name, port, protocol values don't match then we
|
||||
// are not allocated.
|
||||
if !ok {
|
||||
if portConfig.PublishedPort != 0 && portStates.delState(portConfig) == nil {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// If SwarmPort was user defined but the port state
|
||||
// SwarmPort doesn't match we are not allocated.
|
||||
if portConfig.PublishedPort != portState.PublishedPort &&
|
||||
portConfig.PublishedPort != 0 {
|
||||
return false
|
||||
// Iterate portConfigs with PublishedPort == 0 (low priority)
|
||||
for _, portConfig := range s.Spec.Endpoint.Ports {
|
||||
// Ignore ports which are not PublishModeIngress
|
||||
if portConfig.PublishMode != api.PublishModeIngress {
|
||||
continue
|
||||
}
|
||||
|
||||
// If SwarmPort was not defined by user and port state
|
||||
// is not initialized with a valid SwarmPort value then
|
||||
// we are not allocated.
|
||||
if portConfig.PublishedPort == 0 && portState.PublishedPort == 0 {
|
||||
if portConfig.PublishedPort == 0 && portStates.delState(portConfig) == nil {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
|
|
@ -133,7 +133,6 @@ type Dispatcher struct {
|
|||
}
|
||||
|
||||
// New returns Dispatcher with cluster interface(usually raft.Node).
|
||||
// NOTE: each handler which does something with raft must add to Dispatcher.wg
|
||||
func New(cluster Cluster, c *Config) *Dispatcher {
|
||||
d := &Dispatcher{
|
||||
nodes: newNodeStore(c.HeartbeatPeriod, c.HeartbeatEpsilon, c.GracePeriodMultiplier, c.RateLimitPeriod),
|
||||
|
|
|
@ -335,23 +335,46 @@ func (m *Manager) Run(parent context.Context) error {
|
|||
authenticatedHealthAPI := api.NewAuthenticatedWrapperHealthServer(healthServer, authorize)
|
||||
authenticatedRaftMembershipAPI := api.NewAuthenticatedWrapperRaftMembershipServer(m.raftNode, authorize)
|
||||
|
||||
proxyDispatcherAPI := api.NewRaftProxyDispatcherServer(authenticatedDispatcherAPI, m.raftNode, ca.WithMetadataForwardTLSInfo)
|
||||
proxyCAAPI := api.NewRaftProxyCAServer(authenticatedCAAPI, m.raftNode, ca.WithMetadataForwardTLSInfo)
|
||||
proxyNodeCAAPI := api.NewRaftProxyNodeCAServer(authenticatedNodeCAAPI, m.raftNode, ca.WithMetadataForwardTLSInfo)
|
||||
proxyRaftMembershipAPI := api.NewRaftProxyRaftMembershipServer(authenticatedRaftMembershipAPI, m.raftNode, ca.WithMetadataForwardTLSInfo)
|
||||
proxyResourceAPI := api.NewRaftProxyResourceAllocatorServer(authenticatedResourceAPI, m.raftNode, ca.WithMetadataForwardTLSInfo)
|
||||
proxyLogBrokerAPI := api.NewRaftProxyLogBrokerServer(authenticatedLogBrokerAPI, m.raftNode, ca.WithMetadataForwardTLSInfo)
|
||||
proxyDispatcherAPI := api.NewRaftProxyDispatcherServer(authenticatedDispatcherAPI, m.raftNode, nil, ca.WithMetadataForwardTLSInfo)
|
||||
proxyCAAPI := api.NewRaftProxyCAServer(authenticatedCAAPI, m.raftNode, nil, ca.WithMetadataForwardTLSInfo)
|
||||
proxyNodeCAAPI := api.NewRaftProxyNodeCAServer(authenticatedNodeCAAPI, m.raftNode, nil, ca.WithMetadataForwardTLSInfo)
|
||||
proxyRaftMembershipAPI := api.NewRaftProxyRaftMembershipServer(authenticatedRaftMembershipAPI, m.raftNode, nil, ca.WithMetadataForwardTLSInfo)
|
||||
proxyResourceAPI := api.NewRaftProxyResourceAllocatorServer(authenticatedResourceAPI, m.raftNode, nil, ca.WithMetadataForwardTLSInfo)
|
||||
proxyLogBrokerAPI := api.NewRaftProxyLogBrokerServer(authenticatedLogBrokerAPI, m.raftNode, nil, ca.WithMetadataForwardTLSInfo)
|
||||
|
||||
// localProxyControlAPI is a special kind of proxy. It is only wired up
|
||||
// to receive requests from a trusted local socket, and these requests
|
||||
// don't use TLS, therefore the requests it handles locally should
|
||||
// bypass authorization. When it proxies, it sends them as requests from
|
||||
// this manager rather than forwarded requests (it has no TLS
|
||||
// information to put in the metadata map).
|
||||
// The following local proxies are only wired up to receive requests
|
||||
// from a trusted local socket, and these requests don't use TLS,
|
||||
// therefore the requests they handle locally should bypass
|
||||
// authorization. When requests are proxied from these servers, they
|
||||
// are sent as requests from this manager rather than forwarded
|
||||
// requests (it has no TLS information to put in the metadata map).
|
||||
forwardAsOwnRequest := func(ctx context.Context) (context.Context, error) { return ctx, nil }
|
||||
localProxyControlAPI := api.NewRaftProxyControlServer(baseControlAPI, m.raftNode, forwardAsOwnRequest)
|
||||
localProxyLogsAPI := api.NewRaftProxyLogsServer(m.logbroker, m.raftNode, forwardAsOwnRequest)
|
||||
localCAAPI := api.NewRaftProxyCAServer(m.caserver, m.raftNode, forwardAsOwnRequest)
|
||||
handleRequestLocally := func(ctx context.Context) (context.Context, error) {
|
||||
var remoteAddr string
|
||||
if m.config.RemoteAPI.AdvertiseAddr != "" {
|
||||
remoteAddr = m.config.RemoteAPI.AdvertiseAddr
|
||||
} else {
|
||||
remoteAddr = m.config.RemoteAPI.ListenAddr
|
||||
}
|
||||
|
||||
creds := m.config.SecurityConfig.ClientTLSCreds
|
||||
|
||||
nodeInfo := ca.RemoteNodeInfo{
|
||||
Roles: []string{creds.Role()},
|
||||
Organization: creds.Organization(),
|
||||
NodeID: creds.NodeID(),
|
||||
RemoteAddr: remoteAddr,
|
||||
}
|
||||
|
||||
return context.WithValue(ctx, ca.LocalRequestKey, nodeInfo), nil
|
||||
}
|
||||
localProxyControlAPI := api.NewRaftProxyControlServer(baseControlAPI, m.raftNode, handleRequestLocally, forwardAsOwnRequest)
|
||||
localProxyLogsAPI := api.NewRaftProxyLogsServer(m.logbroker, m.raftNode, handleRequestLocally, forwardAsOwnRequest)
|
||||
localProxyDispatcherAPI := api.NewRaftProxyDispatcherServer(m.dispatcher, m.raftNode, handleRequestLocally, forwardAsOwnRequest)
|
||||
localProxyCAAPI := api.NewRaftProxyCAServer(m.caserver, m.raftNode, handleRequestLocally, forwardAsOwnRequest)
|
||||
localProxyNodeCAAPI := api.NewRaftProxyNodeCAServer(m.caserver, m.raftNode, handleRequestLocally, forwardAsOwnRequest)
|
||||
localProxyResourceAPI := api.NewRaftProxyResourceAllocatorServer(baseResourceAPI, m.raftNode, handleRequestLocally, forwardAsOwnRequest)
|
||||
localProxyLogBrokerAPI := api.NewRaftProxyLogBrokerServer(m.logbroker, m.raftNode, handleRequestLocally, forwardAsOwnRequest)
|
||||
|
||||
// Everything registered on m.server should be an authenticated
|
||||
// wrapper, or a proxy wrapping an authenticated wrapper!
|
||||
|
@ -369,7 +392,11 @@ func (m *Manager) Run(parent context.Context) error {
|
|||
api.RegisterControlServer(m.localserver, localProxyControlAPI)
|
||||
api.RegisterLogsServer(m.localserver, localProxyLogsAPI)
|
||||
api.RegisterHealthServer(m.localserver, localHealthServer)
|
||||
api.RegisterCAServer(m.localserver, localCAAPI)
|
||||
api.RegisterDispatcherServer(m.localserver, localProxyDispatcherAPI)
|
||||
api.RegisterCAServer(m.localserver, localProxyCAAPI)
|
||||
api.RegisterNodeCAServer(m.localserver, localProxyNodeCAAPI)
|
||||
api.RegisterResourceAllocatorServer(m.localserver, localProxyResourceAPI)
|
||||
api.RegisterLogBrokerServer(m.localserver, localProxyLogBrokerAPI)
|
||||
|
||||
healthServer.SetServingStatus("Raft", api.HealthCheckResponse_NOT_SERVING)
|
||||
localHealthServer.SetServingStatus("ControlAPI", api.HealthCheckResponse_NOT_SERVING)
|
||||
|
|
Загрузка…
Ссылка в новой задаче