dashboard / erock/pubsub / refactor: create remote client lib #28 rss

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

diff --git a/go.mod b/go.mod
index 073f332..30b4335 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ go 1.23.1
 
 require (
 	github.com/antoniomika/syncmap v1.0.0
+	github.com/google/uuid v1.6.0
 	golang.org/x/crypto v0.28.0
 )
 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

diff --git a/go.sum b/go.sum
index 3c31bf5..b4364cb 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,7 @@
 github.com/antoniomika/syncmap v1.0.0 h1:iFSfbQFQOvHZILFZF+hqWosO0no+W9+uF4y2VEyMKWU=
 github.com/antoniomika/syncmap v1.0.0/go.mod h1:fK2829foEYnO4riNfyUn0SHQZt4ue3DStYjGU+sJj38=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 golang.org/x/crypto v0.28.0 h1:GBDwsMXVQi34v5CCYUm2jkJvu4cbtru2U4TN2PSyQnw=
 golang.org/x/crypto v0.28.0/go.mod h1:rmgy+3RHxRZMyY0jjAJShp2zgEdOqj2AO7U0pYmeQ7U=
 golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

diff --git a/log/log.go b/log/log.go
index 9dc431c..361f6bd 100644
--- a/log/log.go
+++ b/log/log.go
@@ -6,14 +6,11 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
-	"net"
-	"os"
-	"path/filepath"
 	"slices"
-	"strings"
 	"sync"
 	"time"
 
+	"github.com/picosh/pubsub"
 	"golang.org/x/crypto/ssh"
 )
 
@@ -83,14 +80,6 @@ func (m *MultiHandler) WithGroup(name string) slog.Handler {
 	}
 }
 
-type PubSubConnectionInfo struct {
-	RemoteHost     string
-	KeyLocation    string
-	KeyPassphrase  string
-	RemoteHostname string
-	RemoteUser     string
-}
-
 type PubSubLogWriter struct {
 	SSHClient        *ssh.Client
 	Session          *ssh.Session
@@ -103,7 +92,7 @@ type PubSubLogWriter struct {
 	closeMessageOnce sync.Once
 	startOnce        sync.Once
 	connecMu         sync.Mutex
-	ConnectionInfo   *PubSubConnectionInfo
+	ConnectionInfo   *pubsub.RemoteClientInfo
 }
 
 func (c *PubSubLogWriter) Close() error {
@@ -147,7 +136,7 @@ func (c *PubSubLogWriter) Open() error {
 	c.Done = make(chan struct{})
 	c.Messages = make(chan []byte, c.BufferSize)
 
-	sshClient, err := CreateSSHClient(c.ConnectionInfo)
+	sshClient, err := pubsub.CreateRemoteClient(c.ConnectionInfo)
 	if err != nil {
 		c.connecMu.Unlock()
 		return err
@@ -251,64 +240,7 @@ func (c *PubSubLogWriter) Reconnect() {
 	}()
 }
 
-func CreateSSHClient(connectionInfo *PubSubConnectionInfo) (*ssh.Client, error) {
-	if connectionInfo == nil {
-		return nil, fmt.Errorf("connection info is invalid")
-	}
-
-	if !strings.Contains(connectionInfo.RemoteHost, ":") {
-		connectionInfo.RemoteHost += ":22"
-	}
-
-	rawConn, err := net.Dial("tcp", connectionInfo.RemoteHost)
-	if err != nil {
-		return nil, err
-	}
-
-	keyPath, err := filepath.Abs(connectionInfo.KeyLocation)
-	if err != nil {
-		return nil, err
-	}
-
-	f, err := os.Open(keyPath)
-	if err != nil {
-		return nil, err
-	}
-	defer f.Close()
-
-	data, err := io.ReadAll(f)
-	if err != nil {
-		return nil, err
-	}
-
-	var signer ssh.Signer
-
-	if connectionInfo.KeyPassphrase != "" {
-		signer, err = ssh.ParsePrivateKeyWithPassphrase(data, []byte(connectionInfo.KeyPassphrase))
-	} else {
-		signer, err = ssh.ParsePrivateKey(data)
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	sshConn, chans, reqs, err := ssh.NewClientConn(rawConn, connectionInfo.RemoteHostname, &ssh.ClientConfig{
-		Auth:            []ssh.AuthMethod{ssh.PublicKeys(signer)},
-		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
-		User:            connectionInfo.RemoteUser,
-	})
-
-	if err != nil {
-		return nil, err
-	}
-
-	sshClient := ssh.NewClient(sshConn, chans, reqs)
-
-	return sshClient, nil
-}
-
-func SendLogRegister(logger *slog.Logger, connectionInfo *PubSubConnectionInfo, buffer int) (*slog.Logger, error) {
+func SendLogRegister(logger *slog.Logger, connectionInfo *pubsub.RemoteClientInfo, buffer int) (*slog.Logger, error) {
 	if buffer < 0 {
 		buffer = 0
 	}
@@ -339,8 +271,8 @@ func SendLogRegister(logger *slog.Logger, connectionInfo *PubSubConnectionInfo,
 var _ io.Writer = (*PubSubLogWriter)(nil)
 var _ slog.Handler = (*MultiHandler)(nil)
 
-func ConnectToLogs(ctx context.Context, connectionInfo *PubSubConnectionInfo) (io.Reader, error) {
-	sshClient, err := CreateSSHClient(connectionInfo)
+func ConnectToLogs(ctx context.Context, connectionInfo *pubsub.RemoteClientInfo) (io.Reader, error) {
+	sshClient, err := pubsub.CreateRemoteClient(connectionInfo)
 	if err != nil {
 		return nil, err
 	}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

diff --git a/remote_client.go b/remote_client.go
new file mode 100644
index 0000000..84c48f4
--- /dev/null
+++ b/remote_client.go
@@ -0,0 +1,138 @@
+package pubsub
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"golang.org/x/crypto/ssh"
+)
+
+type RemoteClientInfo struct {
+	RemoteHost     string
+	KeyLocation    string
+	KeyPassphrase  string
+	RemoteHostname string
+	RemoteUser     string
+}
+
+func CreateRemoteClient(info *RemoteClientInfo) (*ssh.Client, error) {
+	if info == nil {
+		return nil, fmt.Errorf("connection info is invalid")
+	}
+
+	if !strings.Contains(info.RemoteHost, ":") {
+		info.RemoteHost += ":22"
+	}
+
+	rawConn, err := net.Dial("tcp", info.RemoteHost)
+	if err != nil {
+		return nil, err
+	}
+
+	keyPath, err := filepath.Abs(info.KeyLocation)
+	if err != nil {
+		return nil, err
+	}
+
+	f, err := os.Open(keyPath)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	data, err := io.ReadAll(f)
+	if err != nil {
+		return nil, err
+	}
+
+	var signer ssh.Signer
+
+	if info.KeyPassphrase != "" {
+		signer, err = ssh.ParsePrivateKeyWithPassphrase(data, []byte(info.KeyPassphrase))
+	} else {
+		signer, err = ssh.ParsePrivateKey(data)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	sshConn, chans, reqs, err := ssh.NewClientConn(rawConn, info.RemoteHostname, &ssh.ClientConfig{
+		Auth:            []ssh.AuthMethod{ssh.PublicKeys(signer)},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		User:            info.RemoteUser,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	sshClient := ssh.NewClient(sshConn, chans, reqs)
+
+	return sshClient, nil
+}
+
+func RemoteSub(cmd string, ctx context.Context, info *RemoteClientInfo) (io.Reader, error) {
+	sshClient, err := CreateRemoteClient(info)
+	if err != nil {
+		return nil, err
+	}
+
+	session, err := sshClient.NewSession()
+	if err != nil {
+		return nil, err
+	}
+
+	stdoutPipe, err := session.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+
+	err = session.Start(cmd)
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		<-ctx.Done()
+		session.Close()
+		sshClient.Close()
+	}()
+
+	return stdoutPipe, nil
+}
+
+func RemotePub(cmd string, ctx context.Context, info *RemoteClientInfo) (io.WriteCloser, error) {
+	sshClient, err := CreateRemoteClient(info)
+	if err != nil {
+		return nil, err
+	}
+
+	session, err := sshClient.NewSession()
+	if err != nil {
+		return nil, err
+	}
+
+	stdinPipe, err := session.StdinPipe()
+	if err != nil {
+		return nil, err
+	}
+
+	err = session.Start(cmd)
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		<-ctx.Done()
+		session.Close()
+		sshClient.Close()
+	}()
+
+	return stdinPipe, err
+}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

diff --git a/log/log.go b/log/log.go
index 361f6bd..d959656 100644
--- a/log/log.go
+++ b/log/log.go
@@ -270,33 +270,3 @@ func SendLogRegister(logger *slog.Logger, connectionInfo *pubsub.RemoteClientInf
 
 var _ io.Writer = (*PubSubLogWriter)(nil)
 var _ slog.Handler = (*MultiHandler)(nil)
-
-func ConnectToLogs(ctx context.Context, connectionInfo *pubsub.RemoteClientInfo) (io.Reader, error) {
-	sshClient, err := pubsub.CreateRemoteClient(connectionInfo)
-	if err != nil {
-		return nil, err
-	}
-
-	session, err := sshClient.NewSession()
-	if err != nil {
-		return nil, err
-	}
-
-	stdoutPipe, err := session.StdoutPipe()
-	if err != nil {
-		return nil, err
-	}
-
-	err = session.Start("sub log-drain -k")
-	if err != nil {
-		return nil, err
-	}
-
-	go func() {
-		<-ctx.Done()
-		session.Close()
-		sshClient.Close()
-	}()
-
-	return stdoutPipe, nil
-}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

diff --git a/log/log.go b/log/log.go
index d959656..98bfddc 100644
--- a/log/log.go
+++ b/log/log.go
@@ -270,3 +270,7 @@ func SendLogRegister(logger *slog.Logger, connectionInfo *pubsub.RemoteClientInf
 
 var _ io.Writer = (*PubSubLogWriter)(nil)
 var _ slog.Handler = (*MultiHandler)(nil)
+
+func ConnectToLogs(ctx context.Context, connectionInfo *pubsub.RemoteClientInfo) (io.Reader, error) {
+	return pubsub.RemoteSub("sub log-drain -k", ctx, connectionInfo)
+}