From 0a3136519003d3822f8518d022103b3652c237df Mon Sep 17 00:00:00 2001 From: Yousong Zhou Date: Mon, 10 May 2021 13:36:18 +0800 Subject: [PATCH 1/4] cloudproxy: agent: ssh: wait ssh conn state while passing it out --- pkg/cloudproxy/agent/ssh/client.go | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/pkg/cloudproxy/agent/ssh/client.go b/pkg/cloudproxy/agent/ssh/client.go index 75e4560ea0..0e337a80f7 100644 --- a/pkg/cloudproxy/agent/ssh/client.go +++ b/pkg/cloudproxy/agent/ssh/client.go @@ -178,7 +178,7 @@ func (c *Client) Start(ctx context.Context) { } } -func (c *Client) runClientState(ctx context.Context, sshClientC chan *ssh.Client) { +func (c *Client) runClientState(ctx context.Context, sshClientC chan<- *ssh.Client) { for { select { case <-ctx.Done(): @@ -203,12 +203,6 @@ func (c *Client) runClientState(ctx context.Context, sshClientC chan *ssh.Client func() { defer sshc.Conn.Close() - select { - case sshClientC <- sshc: - case <-ctx.Done(): - return - } - closeC := make(chan struct{}) go func() { defer close(closeC) @@ -219,6 +213,13 @@ func (c *Client) runClientState(ctx context.Context, sshClientC chan *ssh.Client } }() + select { + case sshClientC <- sshc: + case <-closeC: + case <-ctx.Done(): + return + } + select { case <-closeC: case <-ctx.Done(): From 96bcefd50b371a789b9b2d326c16918a1126eabf Mon Sep 17 00:00:00 2001 From: Yousong Zhou Date: Mon, 10 May 2021 14:16:30 +0800 Subject: [PATCH 2/4] cloudproxy: agent: refresh every 11 seconds --- pkg/cloudproxy/agent/ssh/client.go | 1 + pkg/cloudproxy/agent/worker/worker.go | 14 ++++++++++++-- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/pkg/cloudproxy/agent/ssh/client.go b/pkg/cloudproxy/agent/ssh/client.go index 0e337a80f7..a4af8c651d 100644 --- a/pkg/cloudproxy/agent/ssh/client.go +++ b/pkg/cloudproxy/agent/ssh/client.go @@ -150,6 +150,7 @@ func (c *Client) Start(ctx context.Context) { sshClientC := make(chan *ssh.Client) var sshClient *ssh.Client go c.runClientState(ctx, sshClientC) + for { select { case sshClient = <-sshClientC: diff --git a/pkg/cloudproxy/agent/worker/worker.go b/pkg/cloudproxy/agent/worker/worker.go index 9ef20731a1..ea663f6d16 100644 --- a/pkg/cloudproxy/agent/worker/worker.go +++ b/pkg/cloudproxy/agent/worker/worker.go @@ -180,14 +180,24 @@ func (w *Worker) Start(ctx context.Context) { } go w.apih.Start(ctx) - var mss *agentmodels.ModelSets + const tickDur = 11 * time.Second + var ( + mss *agentmodels.ModelSets + tick = time.NewTicker(tickDur) + ) for { select { case imss := <-w.apih.ModelSets(): log.Infof("agent: got new data from api helper") mss = imss.(*agentmodels.ModelSets) if err := w.run(ctx, mss); err != nil { - log.Errorf("agent: %v", err) + log.Errorf("agent run: %v", err) + } + case <-tick.C: + if mss != nil { + if err := w.run(ctx, mss); err != nil { + log.Errorf("agent refresh run: %v", err) + } } case <-ctx.Done(): return From cb35bc3067427c750b6fe008e1f8742891dc0a49 Mon Sep 17 00:00:00 2001 From: Yousong Zhou Date: Mon, 10 May 2021 15:26:42 +0800 Subject: [PATCH 3/4] cloudproxy: agent: no collision between ports of ssh conn and forwards --- pkg/cloudproxy/agent/ssh/client.go | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/pkg/cloudproxy/agent/ssh/client.go b/pkg/cloudproxy/agent/ssh/client.go index a4af8c651d..4b9143e3eb 100644 --- a/pkg/cloudproxy/agent/ssh/client.go +++ b/pkg/cloudproxy/agent/ssh/client.go @@ -18,6 +18,7 @@ import ( "context" "fmt" "net" + "strconv" "sync" "time" @@ -153,7 +154,28 @@ func (c *Client) Start(ctx context.Context) { for { select { - case sshClient = <-sshClientC: + case sshc := <-sshClientC: + conn := sshc.Conn + localAddr := conn.LocalAddr() + localAddrStr := localAddr.String() + addr, portStr, err := net.SplitHostPort(localAddrStr) + if err != nil { + log.Errorf("split host port of ssh client local addr: %v", err) + sshc.Close() + break + } + port, err := strconv.ParseUint(portStr, 10, 16) + if err != nil { + log.Errorf("parse ssh client local port: %v", err) + sshc.Close() + break + } + if v := c.localForwards.get(int(port), addr); v != nil { + log.Errorf("ssh client local port %d collides with local forward: %#v", port, v) + sshc.Close() + break + } + sshClient = sshc case req := <-c.lfc: if sshClient != nil { c.localForward(ctx, sshClient, req) From 50f0ca52902827cb6bbebe2cb588912406b12212 Mon Sep 17 00:00:00 2001 From: Yousong Zhou Date: Mon, 10 May 2021 17:03:57 +0800 Subject: [PATCH 4/4] guests: sshable: fix breaking out from wait loop --- pkg/compute/models/guest_sshable.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/compute/models/guest_sshable.go b/pkg/compute/models/guest_sshable.go index 4ce414fb59..6db66db36e 100644 --- a/pkg/compute/models/guest_sshable.go +++ b/pkg/compute/models/guest_sshable.go @@ -213,10 +213,11 @@ func (guest *SGuest) sshableTryEach( tmo := time.NewTimer(13 * time.Second) tick := time.NewTicker(3 * time.Second) + out: for { select { case <-tmo.C: - break + break out case <-tick.C: if ok := guest.sshableTryForward(ctx, tryData, &fwd); ok { return nil