@@ -3,10 +3,19 @@ package buildlog
33import (
44 "context"
55 "fmt"
6+ "io"
7+ "net/url"
68 "time"
79
10+ "github.com/google/uuid"
11+ "golang.org/x/xerrors"
12+ "storj.io/drpc"
13+
814 "cdr.dev/slog"
9- "github.com/coder/coder/codersdk/agentsdk"
15+ "github.com/coder/coder/v2/agent/proto"
16+ "github.com/coder/coder/v2/codersdk"
17+ "github.com/coder/coder/v2/codersdk/agentsdk"
18+ "github.com/coder/retry"
1019)
1120
1221const (
@@ -22,30 +31,94 @@ type StartupLog struct {
2231}
2332
2433type CoderClient interface {
25- PatchStartupLogs (ctx context.Context , req agentsdk.PatchStartupLogs ) error
34+ Send (level codersdk.LogLevel , log string ) error
35+ io.Closer
2636}
2737
28- type CoderLogger struct {
29- ctx context.Context
30- cancel context.CancelFunc
31- client CoderClient
32- logger slog. Logger
33- logChan chan string
34- err error
38+ type coderClient struct {
39+ ctx context.Context
40+ cancel context.CancelFunc
41+ source uuid. UUID
42+ ls * agentsdk. LogSender
43+ sl agentsdk. ScriptLogger
44+ log slog. Logger
3545}
3646
37- func OpenCoderLogger (ctx context.Context , client CoderClient , log slog.Logger ) Logger {
38- ctx , cancel := context .WithCancel (ctx )
47+ func (c * coderClient ) Send (level codersdk.LogLevel , log string ) error {
48+ err := c .sl .Send (c .ctx , agentsdk.Log {
49+ CreatedAt : time .Now (),
50+ Output : log ,
51+ Level : level ,
52+ })
53+ if err != nil {
54+ return xerrors .Errorf ("send build log: %w" , err )
55+ }
56+ return nil
57+ }
3958
40- coder := & CoderLogger {
41- ctx : ctx ,
42- cancel : cancel ,
43- client : client ,
44- logger : log ,
45- logChan : make ( chan string ),
59+ func ( c * coderClient ) Close () error {
60+ defer c . cancel ()
61+ c . ls . Flush ( c . source )
62+ err := c . ls . WaitUntilEmpty ( c . ctx )
63+ if err != nil {
64+ return xerrors . Errorf ( "wait until empty: %w" , err )
4665 }
66+ return nil
67+ }
4768
48- go coder .processLogs ()
69+ func OpenCoderClient (ctx context.Context , accessURL * url.URL , logger slog.Logger , token string ) (CoderClient , error ) {
70+ client := agentsdk .New (accessURL )
71+ client .SetSessionToken (token )
72+
73+ cctx , cancel := context .WithCancel (ctx )
74+ uid := uuid .New ()
75+ ls := agentsdk .NewLogSender (logger )
76+ sl := ls .GetScriptLogger (uid )
77+
78+ var conn drpc.Conn
79+ var err error
80+ for r := retry .New (10 * time .Millisecond , time .Second ); r .Wait (ctx ); {
81+ conn , err = client .ConnectRPC (ctx )
82+ if err != nil {
83+ logger .Error (ctx , "connect err" , slog .Error (err ))
84+ continue
85+ }
86+ break
87+ }
88+ if conn == nil {
89+ cancel ()
90+ return nil , xerrors .Errorf ("connect rpc: %w" , err )
91+ }
92+
93+ arpc := proto .NewDRPCAgentClient (conn )
94+ go func () {
95+ err := ls .SendLoop (ctx , arpc )
96+ if err != nil {
97+ logger .Error (ctx , "send loop" , slog .Error (err ))
98+ }
99+ }()
100+
101+ return & coderClient {
102+ ctx : cctx ,
103+ cancel : cancel ,
104+ source : uid ,
105+ ls : ls ,
106+ sl : sl ,
107+ log : logger ,
108+ }, nil
109+ }
110+
111+ type CoderLogger struct {
112+ ctx context.Context
113+ client CoderClient
114+ logger slog.Logger
115+ }
116+
117+ func OpenCoderLogger (client CoderClient , log slog.Logger ) Logger {
118+ coder := & CoderLogger {
119+ client : client ,
120+ logger : log ,
121+ }
49122
50123 return coder
51124}
@@ -55,84 +128,31 @@ func (c *CoderLogger) Infof(format string, a ...any) {
55128}
56129
57130func (c * CoderLogger ) Info (output string ) {
58- c .log (output )
131+ c .log (codersdk . LogLevelInfo , output )
59132}
60133
61134func (c * CoderLogger ) Errorf (format string , a ... any ) {
62135 c .Error (fmt .Sprintf (format , a ... ))
63136}
64137
65138func (c * CoderLogger ) Error (output string ) {
66- c .log ("ERROR: " + output )
139+ c .log (codersdk . LogLevelError , output )
67140}
68141
69- func (c * CoderLogger ) log (output string ) {
70- if c .err != nil {
71- return
142+ func (c * CoderLogger ) log (level codersdk.LogLevel , output string ) {
143+ if err := c .client .Send (level , output ); err != nil {
144+ c .logger .Error (c .ctx , "send build log" ,
145+ slog .F ("log" , output ),
146+ slog .Error (err ),
147+ )
72148 }
73- c .logChan <- output
74149}
75150
76151func (c * CoderLogger ) Write (p []byte ) (int , error ) {
77152 c .Info (string (p ))
78153 return len (p ), nil
79154}
80155
81- func (c * CoderLogger ) Close () {
82- c .cancel ()
83- }
84-
85- func (c * CoderLogger ) processLogs () {
86- for {
87- var (
88- line string
89- logs = make ([]agentsdk.StartupLog , 0 , CoderLoggerMaxLogs )
90- )
91-
92- select {
93- case line = <- c .logChan :
94- lines := cutString (line , MaxCoderLogSize )
95-
96- for _ , output := range lines {
97- logs = append (logs , agentsdk.StartupLog {
98- CreatedAt : time .Now (),
99- Output : output ,
100- })
101- }
102-
103- case <- c .ctx .Done ():
104- close (c .logChan )
105- return
106- }
107-
108- // Send the logs in a goroutine so that we can avoid blocking
109- // too long on the channel.
110- cpLogs := logs
111- go func (startupLogs []agentsdk.StartupLog ) {
112- err := c .client .PatchStartupLogs (c .ctx , agentsdk.PatchStartupLogs {
113- Logs : startupLogs ,
114- })
115- if err != nil {
116- c .logger .Error (c .ctx , "send startup logs" , slog .Error (err ))
117- }
118- }(cpLogs )
119- }
120- }
121-
122- // cutString cuts a string up into smaller strings that have a len no greater
123- // than the provided max size.
124- // If the string is less than the max size the return slice has one
125- // element with a value of the provided string.
126- func cutString (s string , maxSize int ) []string {
127- if len (s ) <= maxSize {
128- return []string {s }
129- }
130-
131- toks := []string {}
132- for len (s ) > maxSize {
133- toks = append (toks , s [:maxSize ])
134- s = s [maxSize :]
135- }
136-
137- return append (toks , s )
156+ func (c * CoderLogger ) Close () error {
157+ return c .client .Close ()
138158}
0 commit comments