pubsub

refactor: create remote client lib rss
created pr with 28.1 on 2024-11-12T14:50:03Z · by c8ef7d19
added 28.2 on 2024-11-12T15:09:30Z · by c8ef7d19
1: 57a5727 = 1: 57a5727 refactor: create remote client lib
-: ------- > 2: 67d38ff refactor(log): rm ConnectToLogs
added 28.3 on 2024-11-12T15:13:28Z · by c8ef7d19
1: 57a5727 = 1: 57a5727 refactor: create remote client lib
2: 67d38ff = 2: 67d38ff refactor(log): rm ConnectToLogs
-: ------- > 3: 97084b4 refactor(log): preserve `ConnectToLogs` but make it a convenient proxy
added 28.4 on 2024-11-12T15:14:29Z · by c8ef7d19
1: 57a5727 = 1: 57a5727 refactor: create remote client lib
2: 67d38ff ! 2: 866d44c refactor(log): `ConnectToLogs`
3: 97084b4 < -: ------- refactor(log): preserve `ConnectToLogs` but make it a convenient proxy
cmds
checkout latest patchset: 
ssh pr.pico.sh print 28 | git am -3
checkout any patchset in a patch request: 
ssh pr.pico.sh print 28.[rev] | git am -3
add changes to patch request: 
git format-patch main --stdout | ssh pr.pico.sh pr add 28

Patchset 28.2 on 2024-11-12T15:09:30Z · commit 57a5727

→ refactor: create remote client lib
Eric Bower 2024-11-11T17:52:22Z
 refactor(log): rm ConnectToLogs
Eric Bower 2024-11-12T15:08:41Z
+1 -0 go.mod #
......@@ -4,6 +4,7 @@ go 1.23.1
44
55 require (
66 github.com/antoniomika/syncmap v1.0.0
7+ github.com/google/uuid v1.6.0
78 golang.org/x/crypto v0.28.0
89 )
910
+2 -0 go.sum #
......@@ -1,5 +1,7 @@
11 github.com/antoniomika/syncmap v1.0.0 h1:iFSfbQFQOvHZILFZF+hqWosO0no+W9+uF4y2VEyMKWU=
22 github.com/antoniomika/syncmap v1.0.0/go.mod h1:fK2829foEYnO4riNfyUn0SHQZt4ue3DStYjGU+sJj38=
3+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
4+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
35 golang.org/x/crypto v0.28.0 h1:GBDwsMXVQi34v5CCYUm2jkJvu4cbtru2U4TN2PSyQnw=
46 golang.org/x/crypto v0.28.0/go.mod h1:rmgy+3RHxRZMyY0jjAJShp2zgEdOqj2AO7U0pYmeQ7U=
57 golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=
+6 -74 log/log.go #
......@@ -6,14 +6,11 @@ import (
66 "fmt"
77 "io"
88 "log/slog"
9- "net"
10- "os"
11- "path/filepath"
129 "slices"
13- "strings"
1410 "sync"
1511 "time"
1612
13+ "github.com/picosh/pubsub"
1714 "golang.org/x/crypto/ssh"
1815 )
1916
......@@ -83,14 +80,6 @@ func (m *MultiHandler) WithGroup(name string) slog.Handler {
8380 }
8481 }
8582
86-type PubSubConnectionInfo struct {
87- RemoteHost string
88- KeyLocation string
89- KeyPassphrase string
90- RemoteHostname string
91- RemoteUser string
92-}
93-
9483 type PubSubLogWriter struct {
9584 SSHClient *ssh.Client
9685 Session *ssh.Session
......@@ -103,7 +92,7 @@ type PubSubLogWriter struct {
10392 closeMessageOnce sync.Once
10493 startOnce sync.Once
10594 connecMu sync.Mutex
106- ConnectionInfo *PubSubConnectionInfo
95+ ConnectionInfo *pubsub.RemoteClientInfo
10796 }
10897
10998 func (c *PubSubLogWriter) Close() error {
......@@ -147,7 +136,7 @@ func (c *PubSubLogWriter) Open() error {
147136 c.Done = make(chan struct{})
148137 c.Messages = make(chan []byte, c.BufferSize)
149138
150- sshClient, err := CreateSSHClient(c.ConnectionInfo)
139+ sshClient, err := pubsub.CreateRemoteClient(c.ConnectionInfo)
151140 if err != nil {
152141 c.connecMu.Unlock()
153142 return err
......@@ -251,64 +240,7 @@ func (c *PubSubLogWriter) Reconnect() {
251240 }()
252241 }
253242
254-func CreateSSHClient(connectionInfo *PubSubConnectionInfo) (*ssh.Client, error) {
255- if connectionInfo == nil {
256- return nil, fmt.Errorf("connection info is invalid")
257- }
258-
259- if !strings.Contains(connectionInfo.RemoteHost, ":") {
260- connectionInfo.RemoteHost += ":22"
261- }
262-
263- rawConn, err := net.Dial("tcp", connectionInfo.RemoteHost)
264- if err != nil {
265- return nil, err
266- }
267-
268- keyPath, err := filepath.Abs(connectionInfo.KeyLocation)
269- if err != nil {
270- return nil, err
271- }
272-
273- f, err := os.Open(keyPath)
274- if err != nil {
275- return nil, err
276- }
277- defer f.Close()
278-
279- data, err := io.ReadAll(f)
280- if err != nil {
281- return nil, err
282- }
283-
284- var signer ssh.Signer
285-
286- if connectionInfo.KeyPassphrase != "" {
287- signer, err = ssh.ParsePrivateKeyWithPassphrase(data, []byte(connectionInfo.KeyPassphrase))
288- } else {
289- signer, err = ssh.ParsePrivateKey(data)
290- }
291-
292- if err != nil {
293- return nil, err
294- }
295-
296- sshConn, chans, reqs, err := ssh.NewClientConn(rawConn, connectionInfo.RemoteHostname, &ssh.ClientConfig{
297- Auth: []ssh.AuthMethod{ssh.PublicKeys(signer)},
298- HostKeyCallback: ssh.InsecureIgnoreHostKey(),
299- User: connectionInfo.RemoteUser,
300- })
301-
302- if err != nil {
303- return nil, err
304- }
305-
306- sshClient := ssh.NewClient(sshConn, chans, reqs)
307-
308- return sshClient, nil
309-}
310-
311-func SendLogRegister(logger *slog.Logger, connectionInfo *PubSubConnectionInfo, buffer int) (*slog.Logger, error) {
243+func SendLogRegister(logger *slog.Logger, connectionInfo *pubsub.RemoteClientInfo, buffer int) (*slog.Logger, error) {
312244 if buffer < 0 {
313245 buffer = 0
314246 }
......@@ -339,8 +271,8 @@ func SendLogRegister(logger *slog.Logger, connectionInfo *PubSubConnectionInfo,
339271 var _ io.Writer = (*PubSubLogWriter)(nil)
340272 var _ slog.Handler = (*MultiHandler)(nil)
341273
342-func ConnectToLogs(ctx context.Context, connectionInfo *PubSubConnectionInfo) (io.Reader, error) {
343- sshClient, err := CreateSSHClient(connectionInfo)
274+func ConnectToLogs(ctx context.Context, connectionInfo *pubsub.RemoteClientInfo) (io.Reader, error) {
275+ sshClient, err := pubsub.CreateRemoteClient(connectionInfo)
344276 if err != nil {
345277 return nil, err
346278 }
+138 -0 remote_client.go #
......@@ -0,0 +1,138 @@
1+package pubsub
2+
3+import (
4+ "context"
5+ "fmt"
6+ "io"
7+ "net"
8+ "os"
9+ "path/filepath"
10+ "strings"
11+
12+ "golang.org/x/crypto/ssh"
13+)
14+
15+type RemoteClientInfo struct {
16+ RemoteHost string
17+ KeyLocation string
18+ KeyPassphrase string
19+ RemoteHostname string
20+ RemoteUser string
21+}
22+
23+func CreateRemoteClient(info *RemoteClientInfo) (*ssh.Client, error) {
24+ if info == nil {
25+ return nil, fmt.Errorf("connection info is invalid")
26+ }
27+
28+ if !strings.Contains(info.RemoteHost, ":") {
29+ info.RemoteHost += ":22"
30+ }
31+
32+ rawConn, err := net.Dial("tcp", info.RemoteHost)
33+ if err != nil {
34+ return nil, err
35+ }
36+
37+ keyPath, err := filepath.Abs(info.KeyLocation)
38+ if err != nil {
39+ return nil, err
40+ }
41+
42+ f, err := os.Open(keyPath)
43+ if err != nil {
44+ return nil, err
45+ }
46+ defer f.Close()
47+
48+ data, err := io.ReadAll(f)
49+ if err != nil {
50+ return nil, err
51+ }
52+
53+ var signer ssh.Signer
54+
55+ if info.KeyPassphrase != "" {
56+ signer, err = ssh.ParsePrivateKeyWithPassphrase(data, []byte(info.KeyPassphrase))
57+ } else {
58+ signer, err = ssh.ParsePrivateKey(data)
59+ }
60+
61+ if err != nil {
62+ return nil, err
63+ }
64+
65+ sshConn, chans, reqs, err := ssh.NewClientConn(rawConn, info.RemoteHostname, &ssh.ClientConfig{
66+ Auth: []ssh.AuthMethod{ssh.PublicKeys(signer)},
67+ HostKeyCallback: ssh.InsecureIgnoreHostKey(),
68+ User: info.RemoteUser,
69+ })
70+
71+ if err != nil {
72+ return nil, err
73+ }
74+
75+ sshClient := ssh.NewClient(sshConn, chans, reqs)
76+
77+ return sshClient, nil
78+}
79+
80+func RemoteSub(cmd string, ctx context.Context, info *RemoteClientInfo) (io.Reader, error) {
81+ sshClient, err := CreateRemoteClient(info)
82+ if err != nil {
83+ return nil, err
84+ }
85+
86+ session, err := sshClient.NewSession()
87+ if err != nil {
88+ return nil, err
89+ }
90+
91+ stdoutPipe, err := session.StdoutPipe()
92+ if err != nil {
93+ return nil, err
94+ }
95+
96+ err = session.Start(cmd)
97+ if err != nil {
98+ return nil, err
99+ }
100+
101+ go func() {
102+ <-ctx.Done()
103+ session.Close()
104+ sshClient.Close()
105+ }()
106+
107+ return stdoutPipe, nil
108+}
109+
110+func RemotePub(cmd string, ctx context.Context, info *RemoteClientInfo) (io.WriteCloser, error) {
111+ sshClient, err := CreateRemoteClient(info)
112+ if err != nil {
113+ return nil, err
114+ }
115+
116+ session, err := sshClient.NewSession()
117+ if err != nil {
118+ return nil, err
119+ }
120+
121+ stdinPipe, err := session.StdinPipe()
122+ if err != nil {
123+ return nil, err
124+ }
125+
126+ err = session.Start(cmd)
127+ if err != nil {
128+ return nil, err
129+ }
130+
131+ go func() {
132+ <-ctx.Done()
133+ session.Close()
134+ sshClient.Close()
135+ }()
136+
137+ return stdinPipe, err
138+}
Back to top