@@ -27,6 +27,9 @@ type batchFrame struct {
2727}
2828
2929func Run (ctx context.Context , cfg Config ) error {
30+ ctx , cancel := context .WithCancel (ctx )
31+ defer cancel ()
32+
3033 if cfg .ServiceURL == "" {
3134 return fmt .Errorf ("service-url is required" )
3235 }
@@ -38,6 +41,7 @@ func Run(ctx context.Context, cfg Config) error {
3841 cfgPtr := & cfg
3942 watcher := NewConfigWatcher (cfgPtr )
4043 go watcher .Run (ctx )
44+ go walCleanupLoop (ctx , cfg .WALDir , cfg .StateDir )
4145
4246 // Load prior state; if none, start from the oldest index (first logs)
4347 st , _ := loadState (cfg .StateDir )
@@ -137,7 +141,13 @@ func Run(ctx context.Context, cfg Config) error {
137141 st .CurGz = fm .File
138142 }
139143 if cfg .Meta {
140- fmt .Fprintf (os .Stderr , "frame file=%s seq=%d off=%d len=%d recs=%d\n " , fm .File , fm .Frame , fm .Off , fm .Len , fm .Recs )
144+ logger .Info ().
145+ Str ("file" , fm .File ).
146+ Uint64 ("frame" , fm .Frame ).
147+ Uint64 ("off" , fm .Off ).
148+ Uint64 ("len" , fm .Len ).
149+ Uint32 ("recs" , fm .Recs ).
150+ Msg ("frame metadata" )
141151 }
142152 // Read compressed bytes for this frame
143153 b , rerr := preadSection (gz , int64 (fm .Off ), int64 (fm .Len ))
@@ -197,37 +207,37 @@ func trySend(cfg Config, httpClient *http.Client, batch *[]batchFrame, batchByte
197207
198208 manifestJSON , err := json .Marshal (manifest )
199209 if err != nil {
200- fmt . Fprintf ( os . Stderr , "Error marshaling manifest: %v \n " , err )
210+ logger . Error (). Err ( err ). Msg ( "marshal manifest"
201211 back .Sleep ()
202212 return
203213 }
204214 manifestPart , err := writer .CreateFormField ("manifest" )
205215 if err != nil {
206- fmt . Fprintf ( os . Stderr , "Error creating manifest field: %v \n " , err )
216+ logger . Error (). Err ( err ). Msg ( "create manifest field"
207217 back .Sleep ()
208218 return
209219 }
210220 if _ , err := manifestPart .Write (manifestJSON ); err != nil {
211- fmt . Fprintf ( os . Stderr , "Error writing manifest field: %v \n " , err )
221+ logger . Error (). Err ( err ). Msg ( "write manifest field"
212222 back .Sleep ()
213223 return
214224 }
215225
216226 framesPart , err := writer .CreateFormFile ("frames" , curIdxBase )
217227 if err != nil {
218- fmt . Fprintf ( os . Stderr , "Error creating frames field: %v \n " , err )
228+ logger . Error (). Err ( err ). Msg ( "create frames field"
219229 back .Sleep ()
220230 return
221231 }
222232 for _ , fr := range * batch {
223233 if _ , err := framesPart .Write (fr .Compressed ); err != nil {
224- fmt . Fprintf ( os . Stderr , "Error writing frames payload: %v \n " , err )
234+ logger . Error (). Err ( err ). Msg ( "write frames payload"
225235 back .Sleep ()
226236 return
227237 }
228238 }
229239 if err := writer .Close (); err != nil {
230- fmt . Fprintf ( os . Stderr , "Error finalizing multipart payload: %v \n " , err )
240+ logger . Error (). Err ( err ). Msg ( "finalize multipart payload"
231241 back .Sleep ()
232242 return
233243 }
@@ -245,19 +255,25 @@ func trySend(cfg Config, httpClient *http.Client, batch *[]batchFrame, batchByte
245255
246256 resp , err := httpClient .Do (req )
247257 if err != nil {
248- fmt . Fprintf ( os . Stderr , "Error sending batch: %v \n " , err )
258+ logger . Error (). Err ( err ). Msg ( "send batch"
249259 back .Sleep ()
250260 return
251261 }
252262 defer resp .Body .Close ()
253263 if resp .StatusCode / 100 != 2 {
254264 body , _ := io .ReadAll (resp .Body )
255- fmt .Fprintf (os .Stderr , "Server returned error: status=%d body=%s\n " , resp .StatusCode , string (body ))
265+ logger .Error ().
266+ Int ("status" , resp .StatusCode ).
267+ Str ("body" , string (body )).
268+ Msg ("server returned error" )
256269 back .Sleep ()
257270 return
258271 }
259272
260- fmt .Fprintf (os .Stderr , "Successfully sent batch of %d frames (%d bytes)\n " , len (* batch ), * batchBytes )
273+ logger .Info ().
274+ Int ("frames" , len (* batch )).
275+ Int ("bytes" , * batchBytes ).
276+ Msg ("sent batch" )
261277
262278 // Success: commit idx offset
263279 st .IdxOffset += advance
0 commit comments