LogsSender will send agent startup logs to the server. Calls to sendLog are non-blocking and will return an error if flushAndClose has been called. Calling sendLog concurrently is not supported. If the context passed to flushAndClose is canceled, any remaining logs will be discarded. Deprecated: Us
(sourceID uuid.UUID, patchLogs func(ctx context.Context, req PatchLogs) error, logger slog.Logger, opts ...func(*logsSenderOptions))
| 123 | // |
| 124 | // Deprecated: Use NewLogSender instead, based on the v2 Agent API. |
| 125 | func LogsSender(sourceID uuid.UUID, patchLogs func(ctx context.Context, req PatchLogs) error, logger slog.Logger, opts ...func(*logsSenderOptions)) (sendLog func(ctx context.Context, log ...Log) error, flushAndClose func(context.Context) error) { |
| 126 | o := logsSenderOptions{ |
| 127 | flushTimeout: 250 * time.Millisecond, |
| 128 | } |
| 129 | for _, opt := range opts { |
| 130 | opt(&o) |
| 131 | } |
| 132 | |
| 133 | // The main context is used to close the sender goroutine and cancel |
| 134 | // any outbound requests to the API. The shutdown context is used to |
| 135 | // signal the sender goroutine to flush logs and then exit. |
| 136 | ctx, cancel := context.WithCancel(context.Background()) |
| 137 | shutdownCtx, shutdown := context.WithCancel(ctx) |
| 138 | |
| 139 | // Synchronous sender, there can only be one outbound send at a time. |
| 140 | sendDone := make(chan struct{}) |
| 141 | send := make(chan []Log, 1) |
| 142 | go func() { |
| 143 | // Set flushTimeout and backlogLimit so that logs are uploaded |
| 144 | // once every 250ms or when 100 logs have been added to the |
| 145 | // backlog, whichever comes first. |
| 146 | backlogLimit := 100 |
| 147 | |
| 148 | flush := time.NewTicker(o.flushTimeout) |
| 149 | |
| 150 | var backlog []Log |
| 151 | defer func() { |
| 152 | flush.Stop() |
| 153 | if len(backlog) > 0 { |
| 154 | logger.Warn(ctx, "startup logs sender exiting early, discarding logs", slog.F("discarded_logs_count", len(backlog))) |
| 155 | } |
| 156 | logger.Debug(ctx, "startup logs sender exited") |
| 157 | close(sendDone) |
| 158 | }() |
| 159 | |
| 160 | done := false |
| 161 | for { |
| 162 | flushed := false |
| 163 | select { |
| 164 | case <-ctx.Done(): |
| 165 | return |
| 166 | case <-shutdownCtx.Done(): |
| 167 | done = true |
| 168 | |
| 169 | // Check queued logs before flushing. |
| 170 | select { |
| 171 | case logs := <-send: |
| 172 | backlog = append(backlog, logs...) |
| 173 | default: |
| 174 | } |
| 175 | case <-flush.C: |
| 176 | flushed = true |
| 177 | case logs := <-send: |
| 178 | backlog = append(backlog, logs...) |
| 179 | flushed = len(backlog) >= backlogLimit |
| 180 | } |
| 181 | |
| 182 | if (done || flushed) && len(backlog) > 0 { |