[TASK] add ssh List for handling result of everywhere run

runtime/nodes.go

@@ -58,7 +58,7 @@ func (nodes *Nodes) LearnNode(n *yanic.Node) {
 	// session := nodes.ssh.ConnectTo(node.Address)
 	result, err := nodes.ssh.RunOn(node.GetAddress(nodes.iface), "uptime")
 	if err != nil {
-		logger.Error("init ssh command not run")
+		logger.Error("init ssh command not run", err)
 		return
 	}
 	uptime := ssh.SSHResultToString(result)

ssh/execute.go

@@ -13,25 +13,27 @@ func (m *Manager) ExecuteEverywhere(cmd string) {
 	}
 }
 
-func (m *Manager) ExecuteOn(addr net.TCPAddr, cmd string) {
-	client := m.ConnectTo(addr)
-	if client != nil {
-		m.execute(addr.IP.String(), client, cmd)
+func (m *Manager) ExecuteOn(addr net.TCPAddr, cmd string) error {
+	client, err := m.ConnectTo(addr)
+	if err != nil {
+		return err
 	}
+	return m.execute(addr.IP.String(), client, cmd)
 }
 
-func (m *Manager) execute(host string, client *ssh.Client, cmd string) {
+func (m *Manager) execute(host string, client *ssh.Client, cmd string) error {
 	session, err := client.NewSession()
 	defer session.Close()
 
 	if err != nil {
 		log.Log.Warnf("can not create session on %s: %s", host, err)
 		delete(m.clients, host)
-		return
+		return err
 	}
 	err = session.Run(cmd)
 	if err != nil {
 		log.Log.Warnf("could not run %s on %s: %s", cmd, host, err)
-		delete(m.clients, host)
+		return err
 	}
+	return nil
 }

ssh/execute_test.go

@@ -10,14 +10,21 @@ import (
 func TestExecute(t *testing.T) {
 	assert := assert.New(t)
 
+	addr := net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22}
+
 	mgmt := NewManager("~/.ssh/id_rsa")
 	assert.NotNil(mgmt, "no new manager created")
 
-	mgmt.ConnectTo(net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22})
+	_, err := mgmt.ConnectTo(addr)
+	assert.NoError(err)
 
 	mgmt.ExecuteEverywhere("echo $HOSTNAME")
-	mgmt.ExecuteOn(net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22}, "uptime")
-	mgmt.ExecuteOn(net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22}, "echo $HOSTNAME")
+	err = mgmt.ExecuteOn(addr, "uptime")
+	assert.NoError(err)
+	err = mgmt.ExecuteOn(addr, "echo $HOSTNAME")
+	assert.NoError(err)
+	err = mgmt.ExecuteOn(addr, "exit 1")
+	assert.Error(err)
 
 	mgmt.Close()
 }

ssh/list.go

@@ -0,0 +1,52 @@
+package ssh
+
+import (
+	"sync"
+
+	"golang.org/x/crypto/ssh"
+)
+
+type List struct {
+	cmd        string
+	Clients    map[string]*ListResult
+	sshManager *Manager
+}
+type ListResult struct {
+	ssh       *ssh.Client
+	Runned    bool
+	WithError bool
+	Result    string
+}
+
+func (m *Manager) CreateList(cmd string) *List {
+	list := &List{
+		cmd:        cmd,
+		sshManager: m,
+		Clients:    make(map[string]*ListResult),
+	}
+	for host, client := range m.clients {
+		list.Clients[host] = &ListResult{Runned: false, ssh: client}
+	}
+	return list
+}
+
+func (l List) Run() {
+	wg := new(sync.WaitGroup)
+	for host, client := range l.Clients {
+		wg.Add(1)
+		go l.runlistelement(host, client, wg)
+
+	}
+	wg.Wait()
+}
+
+func (l List) runlistelement(host string, client *ListResult, wg *sync.WaitGroup) {
+	defer wg.Done()
+	result, err := l.sshManager.run(host, client.ssh, l.cmd)
+	client.Runned = true
+	if err != nil {
+		client.WithError = true
+		return
+	}
+	client.Result = SSHResultToString(result)
+}

ssh/list_test.go

@@ -0,0 +1,36 @@
+package ssh
+
+import (
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestList(t *testing.T) {
+	assert := assert.New(t)
+	addr := net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22}
+
+	mgmt := NewManager("~/.ssh/id_rsa")
+	assert.NotNil(mgmt, "no new manager created")
+	_, err := mgmt.ConnectTo(addr)
+	assert.NoError(err)
+
+	list := mgmt.CreateList("exit 1")
+	assert.Len(list.Clients, 1)
+	client := list.Clients[addr.IP.String()]
+	assert.False(client.Runned)
+	list.Run()
+	assert.True(client.Runned)
+	assert.True(client.WithError)
+	assert.Equal("", client.Result)
+
+	list = mgmt.CreateList("echo 15")
+	assert.Len(list.Clients, 1)
+	client = list.Clients[addr.IP.String()]
+	assert.False(client.Runned)
+	list.Run()
+	assert.True(client.Runned)
+	assert.False(client.WithError)
+	assert.Equal("15", client.Result)
+}

ssh/manager.go

@@ -1,6 +1,7 @@
 package ssh
 
 import (
+	"errors"
 	"net"
 	"strings"
 	"sync"
@@ -40,19 +41,19 @@ func NewManager(file string) *Manager {
 	}
 }
 
-func (m *Manager) ConnectTo(addr net.TCPAddr) *ssh.Client {
+func (m *Manager) ConnectTo(addr net.TCPAddr) (*ssh.Client, error) {
 	m.clientsMUX.Lock()
 	defer m.clientsMUX.Unlock()
 	if t, ok := m.clientsBlacklist[addr.IP.String()]; ok {
 		if time.Now().Add(-time.Hour * 24).Before(t) {
-			return nil
+			return nil, errors.New("node on blacklist")
 		} else {
 			delete(m.clientsBlacklist, addr.IP.String())
 		}
 	}
 
 	if client, ok := m.clients[addr.IP.String()]; ok {
-		return client
+		return client, nil
 	}
 
 	client, err := ssh.Dial("tcp", addr.String(), m.config)
@@ -60,14 +61,13 @@ func (m *Manager) ConnectTo(addr net.TCPAddr) *ssh.Client {
 		if strings.Contains(err.Error(), "no supported methods remain") {
 			m.clientsBlacklist[addr.IP.String()] = time.Now()
 			log.Log.Warnf("node was set on the blacklist: %s", err)
-		} else {
-			log.Log.Error(err)
+			return nil, errors.New("node on blacklist")
 		}
-		return nil
+		return nil, err
 	}
 
 	m.clients[addr.IP.String()] = client
-	return client
+	return client, nil
 }
 
 func (m *Manager) Close() {

ssh/run.go

@@ -2,7 +2,6 @@ package ssh
 
 import (
 	"bytes"
-	"errors"
 	"io"
 	"net"
 
@@ -35,11 +34,11 @@ func (m *Manager) RunEverywhere(cmd string, handler SSHResultHandler) {
 }
 
 func (m *Manager) RunOn(addr net.TCPAddr, cmd string) ([]byte, error) {
-	client := m.ConnectTo(addr)
-	if client != nil {
-		return m.run(addr.IP.String(), client, cmd)
+	client, err := m.ConnectTo(addr)
+	if err != nil {
+		return nil, err
 	}
-	return nil, errors.New("no connection for runOn")
+	return m.run(addr.IP.String(), client, cmd)
 }
 
 func (m *Manager) run(host string, client *ssh.Client, cmd string) ([]byte, error) {
@@ -62,7 +61,6 @@ func (m *Manager) run(host string, client *ssh.Client, cmd string) ([]byte, erro
 	err = session.Run(cmd)
 	if err != nil {
 		log.Log.Warnf("could not run %s on %s: %s", cmd, host, err)
-		delete(m.clients, host)
 		return nil, err
 	}
 	var result []byte

ssh/run_test.go

@@ -10,18 +10,19 @@ import (
 
 func TestRun(t *testing.T) {
 	assert := assert.New(t)
-
+	addr := net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22}
 	mgmt := NewManager("~/.ssh/id_rsa")
 	assert.NotNil(mgmt, "no new manager created")
 
-	mgmt.ConnectTo(net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22})
+	_, err := mgmt.ConnectTo(addr)
+	assert.NoError(err)
 
 	mgmt.RunEverywhere("echo 13", SSHResultToStringHandler(func(result string, err error) {
 		assert.NoError(err)
 
 		assert.Equal("13", result)
 	}))
-	result, err := mgmt.RunOn(net.TCPAddr{IP: net.ParseIP("2a06:8782:ffbb:1337::127"), Port: 22}, "echo 16")
+	result, err := mgmt.RunOn(addr, "echo 16")
 	assert.NoError(err)
 
 	str := SSHResultToString(result)