@@ -180,8 +180,13 @@ func buildURL(endpoint logconfig.Endpoint) string {
180180 address = endpoint .Host
181181 }
182182
183+ scheme := "https"
184+ if ! endpoint .UseSSL () {
185+ scheme = "http"
186+ }
187+
183188 url := url.URL {
184- Scheme : "https" ,
189+ Scheme : scheme ,
185190 Host : address ,
186191 Path : endpoint .PathPrefix + telemetryPath ,
187192 }
@@ -410,71 +415,117 @@ func (ss *senderSession) flush() Payload {
410415 return payload
411416}
412417
413- func (s * senderImpl ) flushSession (ss * senderSession ) error {
414- // There is nothing to do if there are no payloads
415- if ss .payloadCount () == 0 {
416- return nil
418+ // sendPayloadBody POSTs body to a single endpoint with the given request
419+ // type and API key, and returns the HTTP status code plus any transport
420+ // error.
421+ //
422+ // Status-code interpretation (2xx success, 4xx terminal, 5xx retryable) is
423+ // the caller's responsibility — flushSession and the errortracking path
424+ // want different policies despite sharing the marshal / scrub / compress
425+ // / transport stack. Centralising the per-endpoint POST here means there
426+ // is exactly one place to audit headers, compression handling, response
427+ // body lifecycle and error wrapping; addresses review comment C3 on
428+ // PR #49946.
429+ //
430+ // Return contract:
431+ // - (statusCode, nil) on a successful round trip (caller inspects status).
432+ // - (0, err) on request-build or transport failure.
433+ //
434+ // The response body is always closed before this function returns.
435+ func (s * senderImpl ) sendPayloadBody (ctx context.Context , body []byte , reqType , apiKey , url string , compressed bool ) (int , error ) {
436+ req , err := http .NewRequest ("POST" , url , bytes .NewReader (body ))
437+ if err != nil {
438+ return 0 , fmt .Errorf ("new %s request to %s: %w" , reqType , url , err )
417439 }
440+ s .addHeaders (req , reqType , apiKey , strconv .Itoa (len (body )), compressed )
441+ resp , err := s .client .Do (req .WithContext (ctx ))
442+ if err != nil {
443+ return 0 , fmt .Errorf ("post %s to %s: %w" , reqType , url , err )
444+ }
445+ if resp .Body != nil {
446+ resp .Body .Close ()
447+ }
448+ return resp .StatusCode , nil
449+ }
418450
419- s .logComp .Debugf ("Flushing Agent Telemetery session with %d payloads" , ss .payloadCount ())
420-
421- payloads := ss .flush ()
422- payloadJSON , err := json .Marshal (payloads )
451+ // sendPayload marshals v, scrubs sensitive fields, optionally
452+ // zstd-compresses, and POSTs to every configured endpoint via the shared
453+ // sendPayloadBody helper. The four steps (marshal -> scrub -> compress ->
454+ // endpoint iteration) are identical for both the metrics-style payload
455+ // flushed by flushSession and the logs-style payload sent by
456+ // sendLogsBatch -- this is the single home for them.
457+ //
458+ // Returns a joined error containing every endpoint's transport failure
459+ // and every non-2xx status, with the URL embedded. Callers may log the
460+ // error at Debug (telemetry is opportunistic) and move on.
461+ func (s * senderImpl ) sendPayload (ctx context.Context , v any , reqType string ) error {
462+ payloadJSON , err := json .Marshal (v )
423463 if err != nil {
424- return fmt .Errorf ("failed to marshal agent telemetry payload: %w" , err )
464+ return fmt .Errorf ("marshal %s payload: %w" , reqType , err )
425465 }
426466
427467 reqBodyRaw , err := scrubber .ScrubJSON (payloadJSON )
428468 if err != nil {
429- return fmt .Errorf ("failed to scrubl agent telemetry payload: %w" , err )
469+ return fmt .Errorf ("scrub %s payload: %w" , reqType , err )
430470 }
431471
432- // Try to compress the payload if needed
472+ return s .sendPayloadBytes (ctx , reqBodyRaw , reqType )
473+ }
474+
475+ // sendPayloadBytes compresses (if configured) and POSTs reqBodyRaw to every
476+ // configured endpoint. Callers handling scrubbing directly call
477+ // this directly instead of going through sendPayload.
478+ //
479+ // Returns a joined error of every transport failure. Non-2xx HTTP statuses are
480+ // logged at Debug and do not surface in the error (opportunistic telemetry
481+ // contract, see PR #50607 F1/F6).
482+ func (s * senderImpl ) sendPayloadBytes (ctx context.Context , reqBodyRaw []byte , reqType string ) error {
483+ // Try to compress the payload if needed. On compression failure we
484+ // fall back to the uncompressed body and emit a Debug log -- this
485+ // flush path is opportunistic and must never log at Error (an Error
486+ // here would re-enter the errortracking handler in the worst case;
487+ // see review comment F6 on PR #50607).
433488 reqBody := reqBodyRaw
434489 compressed := false
435490 if s .compress {
436- // In case of failed to compress continue with uncompress body
437491 reqBodyCompressed , errTemp := zstd .CompressLevel (nil , reqBodyRaw , s .compressionLevel )
438492 if errTemp == nil {
439493 compressed = true
440494 reqBody = reqBodyCompressed
441495 } else {
442- s .logComp .Errorf ("Failed to compress agent telemetry payload: %v" , errTemp )
496+ s .logComp .Debugf ("Failed to compress %s payload: %v" , reqType , errTemp )
443497 }
444498 }
445499
446- // Send the payload to all endpoints
500+ // Send to every configured endpoint. Status codes -- both 2xx and
501+ // non-2xx -- are logged at Debug; only transport failures surface in
502+ // the returned joined error. This preserves the pre-F1 flushSession
503+ // contract (non-2xx is not an error to the caller) and keeps the
504+ // errortracking flush path free of any Errorf-level log that would
505+ // re-enter the slog handler (see F1/F6 on PR #50607).
447506 var errs error
448- reqType := payloads .RequestType
449- bodyLen := strconv .Itoa (len (reqBody ))
450507 for _ , ep := range s .endpoints .Endpoints {
451508 url := buildURL (ep )
452- req , err := http . NewRequest ( "POST" , url , bytes . NewReader ( reqBody ) )
453- if err != nil {
454- errs = errors .Join (errs , err )
509+ status , sendErr := s . sendPayloadBody ( ctx , reqBody , reqType , ep . GetAPIKey (), url , compressed )
510+ if sendErr != nil {
511+ errs = errors .Join (errs , sendErr )
455512 continue
456513 }
457- s .addHeaders (req , reqType , ep .GetAPIKey (), bodyLen , compressed )
458- resp , err := s .client .Do (req .WithContext (ss .cancelCtx ))
459- if err != nil {
460- errs = errors .Join (errs , err )
461- continue
462- }
463- defer func () {
464- if resp != nil && resp .Body != nil {
465- resp .Body .Close ()
466- }
467- }()
514+ s .logComp .Debugf ("Telemetry endpoint response status code:%d, request type:%s, url:%s" , status , reqType , url )
515+ }
516+ return errs
517+ }
468518
469- // Log return status (and URL if unsuccessful)
470- if resp .StatusCode >= 200 && resp .StatusCode < 300 {
471- s .logComp .Debugf ("Telemetry endpoint response status:%s, request type:%s, status code:%d" , resp .Status , reqType , resp .StatusCode )
472- } else {
473- s .logComp .Debugf ("Telemetry endpoint response status:%s, request type:%s, status code:%d, url:%s" , resp .Status , reqType , resp .StatusCode , url )
474- }
519+ func (s * senderImpl ) flushSession (ss * senderSession ) error {
520+ // There is nothing to do if there are no payloads
521+ if ss .payloadCount () == 0 {
522+ return nil
475523 }
476524
477- return errs
525+ s .logComp .Debugf ("Flushing Agent Telemetery session with %d payloads" , ss .payloadCount ())
526+
527+ payloads := ss .flush ()
528+ return s .sendPayload (ss .cancelCtx , payloads , payloads .RequestType )
478529}
479530
480531func (s * senderImpl ) sendAgentMetricPayloads (ss * senderSession , metrics []* agentmetric ) {
0 commit comments