version 1.3.1: bugfix in etcd connexion: too many goroutines were launched

Author: cyrilst

Makefile

@@ -1,4 +1,4 @@
-SSHPROXY_VERSION ?= 1.3.0
+SSHPROXY_VERSION ?= 1.3.1
 SSHPROXY_GIT_URL ?= github.com/cea-hpc/sshproxy
 
 prefix		?= /usr

cmd/sshproxy/recorder.go

@@ -22,6 +22,8 @@ import (
 
 	"github.com/cea-hpc/sshproxy/pkg/record"
 	"github.com/cea-hpc/sshproxy/pkg/utils"
+
+	"go.etcd.io/etcd/clientv3"
 )
 
 // Dup duplicates a []byte slice.
@@ -100,6 +102,7 @@ type Recorder struct {
 	dumpfile              string             // path to filename where the raw records are dumped
 	dumpLimitSize         uint64             // number of bytes beyond which records are no longer dumped
 	dumpLimitWindow       time.Duration      // time window in which dump size is accounted
+	leaseID               clientv3.LeaseID   // etcd lease ID used for updating stats
 	writer                *record.Writer     // *record.Writer where the raw records are dumped
 }
 
@@ -108,7 +111,7 @@ type Recorder struct {
 // If dumpfile is not empty, the intercepted raw data will be written in this
 // file. Logging of basic statistics will be done every logStatsInterval seconds. Bandwidth will be updated in etcd every etcdStatsInterval seconds.
 // It will stop recording when the context is cancelled.
-func NewRecorder(conninfo *ConnInfo, dumpfile, command string, etcdStatsInterval time.Duration, logStatsInterval time.Duration, dumpLimitSize uint64, dumpLimitWindow time.Duration) *Recorder {
+func NewRecorder(conninfo *ConnInfo, dumpfile, command string, etcdStatsInterval time.Duration, logStatsInterval time.Duration, dumpLimitSize uint64, dumpLimitWindow time.Duration, leaseID clientv3.LeaseID) *Recorder {
 	ch := make(chan record.Record)
 
 	return &Recorder{
@@ -125,27 +128,19 @@ func NewRecorder(conninfo *ConnInfo, dumpfile, command string, etcdStatsInterval
 		dumpfile:          dumpfile,
 		dumpLimitSize:     dumpLimitSize,
 		dumpLimitWindow:   dumpLimitWindow,
+		leaseID:           leaseID,
 		writer:            nil,
 	}
 }
 
 // updateStats writes the bandwidth to etcd
-func (r *Recorder) updateStats(ctx context.Context, cli *utils.Client, etcdPath string) {
+func (r *Recorder) updateStats(cli *utils.Client, etcdPath string) {
 	if cli != nil {
 		if cli.IsAlive() {
-			keepAliveChan, err := cli.UpdateStats(ctx, etcdPath, r.bandwidth)
+			err := cli.UpdateStats(etcdPath, r.bandwidth, r.leaseID)
 			if err != nil {
 				log.Errorf("updating stats: %v", err)
 			}
-			go func() {
-				for {
-					select {
-					case <-keepAliveChan:
-					case <-ctx.Done():
-						return
-					}
-				}
-			}()
 		}
 	}
 }
@@ -245,7 +240,7 @@ func (r *Recorder) Run(ctx context.Context, cli *utils.Client, etcdPath string)
 			for {
 				select {
 				case <-time.After(r.etcdStatsInterval):
-					r.updateStats(ctx, cli, etcdPath)
+					r.updateStats(cli, etcdPath)
 				case <-ctx.Done():
 					return
 				}

cmd/sshproxy/sshproxy.go

@@ -32,6 +32,7 @@ import (
 
 	"github.com/docker/docker/pkg/term"
 	"github.com/op/go-logging"
+	"go.etcd.io/etcd/clientv3"
 )
 
 var (
@@ -351,11 +352,13 @@ func mainExitCode() int {
 	}()
 
 	var etcdPath string
+	var leaseID clientv3.LeaseID
 	// Register destination in etcd and keep it alive while running.
 	if cli != nil && cli.IsAlive() {
 		key := fmt.Sprintf("%s@%s", username, service)
-		keepAliveChan, eP, err := cli.SetDestination(ctx, key, sshInfos.Dst(), hostport)
+		keepAliveChan, eP, lID, err := cli.SetDestination(ctx, key, sshInfos.Dst(), hostport)
 		etcdPath = eP
+		leaseID = lID
 		if err != nil {
 			log.Warningf("setting destination in etcd: %v", err)
 		}
@@ -444,7 +447,7 @@ func mainExitCode() int {
 
 	var recorder *Recorder
 	if config.Dump != "" {
-		recorder = NewRecorder(conninfo, config.Dump, originalCmd, config.EtcdStatsInterval.Duration(), config.LogStatsInterval.Duration(), config.DumpLimitSize, config.DumpLimitWindow.Duration())
+		recorder = NewRecorder(conninfo, config.Dump, originalCmd, config.EtcdStatsInterval.Duration(), config.LogStatsInterval.Duration(), config.DumpLimitSize, config.DumpLimitWindow.Duration(), leaseID)
 
 		wg.Add(1)
 		go func() {

misc/sshproxy.spec

@@ -3,7 +3,7 @@
 %global debug_package   %{nil}
 
 Name:           sshproxy
-Version:        1.3.0
+Version:        1.3.1
 Release:        1%{?dist}
 Summary:        SSH proxy
 License:        CeCILL-B
@@ -51,6 +51,9 @@ install -p -m 0644 config/sshproxy.yaml %{buildroot}%{_sysconfdir}/sshproxy
 %{_mandir}/man8/sshproxy-replay.8*
 
 %changelog
+* Wed Sep 23 2020 Cyril Servant <[email protected]> - 1.3.1-1
+- sshproxy 1.3.1
+
 * Wed Aug 05 2020 Cyril Servant <[email protected]> - 1.3.0-1
 - sshproxy 1.3.0
 

pkg/utils/etcd.go

@@ -205,58 +205,51 @@ func (c *Client) GetDestination(key string) (string, error) {
 }
 
 // SetDestination set current destination in etcd.
-func (c *Client) SetDestination(rootctx context.Context, key, sshdHostport string, dst string) (<-chan *clientv3.LeaseKeepAliveResponse, string, error) {
+func (c *Client) SetDestination(rootctx context.Context, key, sshdHostport string, dst string) (<-chan *clientv3.LeaseKeepAliveResponse, string, clientv3.LeaseID, error) {
 	path := fmt.Sprintf("%s/%s/%s/%s", toConnectionKey(key), dst, sshdHostport, time.Now().Format(time.RFC3339Nano))
 	ctx, cancel := context.WithTimeout(context.Background(), c.requestTimeout)
 	resp, err := c.cli.Grant(ctx, c.keyTTL)
 	cancel()
 	if err != nil {
-		return nil, "", err
+		return nil, "", 0, err
 	}
 
 	bytes, err := json.Marshal(&Bandwidth{
 		In:  0,
 		Out: 0,
 	})
 	if err != nil {
-		return nil, "", err
+		return nil, "", 0, err
 	}
 	ctx, cancel = context.WithTimeout(context.Background(), c.requestTimeout)
 	_, err = c.cli.Put(ctx, path, string(bytes), clientv3.WithLease(resp.ID))
 	cancel()
 	if err != nil {
-		return nil, "", err
+		return nil, "", 0, err
 	}
 
 	k, e := c.cli.KeepAlive(rootctx, resp.ID)
-	return k, path, e
+	return k, path, resp.ID, e
 }
 
 // UpdateStats updates the stats (bandwidth in and out in kB/s) of a connection.
-func (c *Client) UpdateStats(rootctx context.Context, etcdPath string, stats map[int]uint64) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
+func (c *Client) UpdateStats(etcdPath string, stats map[int]uint64, leaseID clientv3.LeaseID) error {
 	bytes, err := json.Marshal(&Bandwidth{
 		In:  int(stats[0] / 1024),
 		Out: int((stats[1] + stats[2]) / 1024),
 	})
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), c.requestTimeout)
-	resp, err := c.cli.Grant(ctx, c.keyTTL)
-	cancel()
-	if err != nil {
-		return nil, err
-	}
-
-	ctx, cancel = context.WithTimeout(context.Background(), c.requestTimeout)
-	_, err = c.cli.Put(ctx, etcdPath, string(bytes), clientv3.WithLease(resp.ID))
+	_, err = c.cli.Put(ctx, etcdPath, string(bytes), clientv3.WithLease(leaseID))
 	cancel()
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	return c.cli.KeepAlive(rootctx, resp.ID)
+	return nil
 }
 
 // GetHost returns the host (passed as "host:port") details. If host is not