[AGNTLOG-392] Logs Agent HTTP Retry with Exponential Backoff (#43500)

### What does this PR do?
Implements automatic HTTP transport upgrade under an exponential backoff for the Logs Agent when HTTP connectivity is restored after an initial fallback to TCP, along with improved telemetry.

Smart HTTP Retry and Upgrade Mechanism: 
- When the Logs Agent fails initial HTTP connectivity checks, it falls back to TCP protocol
- If configuration permits (no force_use_tcp, socks5_proxy_address, or additional_endpoints set), the agent periodically retries HTTP connectivity
- Once HTTP connectivity is verified, the agent automatically upgrades to HTTP transport
- Commit to HTTP: After successful HTTP connectivity check, the agent commits to HTTP and continuously retries HTTP transport even if restart attempts fail, ensuring eventual upgrade
- Configurable `logs_config.http_connectivity_retry_interval_max` for exponential backoff (`default=1hr`)

Telemetry Metrics
- `datadog.logs_agent.http_connectivity_check` - Tracks HTTP connectivity check results and TCP fallback events
- `datadog.logs_agent.http_connectivity_retry_attempt` - Tracks HTTP connectivity retry attempts
- `datadog.logs_agent.restart_attempt` - Tracks logs agent restart attempts by status and transport type

### Motivation
https://datadoghq.atlassian.net/browse/AGNTLOG-392

### Describe how you validated your changes
unit tests, local testing (see comment below)
### Additional Notes


Co-authored-by: gh123man <[email protected]>

Author: ellataira

comp/core/agenttelemetry/impl/config.go

@@ -227,6 +227,16 @@ var defaultProfiles = `
         - name: logs.encoded_bytes_sent
           aggregate_tags:
             - compression_kind
+        - name: logs.http_connectivity_check
+          aggregate_tags:
+            - status
+        - name: logs.http_connectivity_retry_attempt
+          aggregate_tags:
+            - status
+        - name: logs.restart_attempt
+          aggregate_tags:
+            - status
+            - transport
         - name: logs.sender_latency
         - name: logs.truncated
           aggregate_tags:

comp/logs/agent/agentimpl/agent.go

@@ -134,6 +134,11 @@ type logAgent struct {
 
 	// make restart thread safe
 	restartMutex sync.Mutex
+
+	// HTTP retry state for TCP fallback recovery
+	httpRetryCtx    context.Context
+	httpRetryCancel context.CancelFunc
+	httpRetryMutex  sync.Mutex
 }
 
 func newLogsAgent(deps dependencies) provides {
@@ -212,6 +217,11 @@ func (a *logAgent) start(context.Context) error {
 	}
 
 	a.startPipeline()
+
+	// If we're currently sending over TCP, attempt restart over HTTP
+	if !endpoints.UseHTTP {
+		a.smartHTTPRestart()
+	}
 	return nil
 }
 
@@ -302,6 +312,9 @@ func (a *logAgent) stop(context.Context) error {
 
 	a.log.Info("Stopping logs-agent")
 
+	// Stop HTTP retry loop if running
+	a.stopHTTPRetry()
+
 	status.Clear()
 
 	toStop := []startstop.Stoppable{

comp/logs/agent/agentimpl/agent_core_init.go

@@ -26,6 +26,7 @@ import (
 	"github.com/DataDog/datadog-agent/pkg/logs/launchers/journald"
 	"github.com/DataDog/datadog-agent/pkg/logs/launchers/listener"
 	"github.com/DataDog/datadog-agent/pkg/logs/launchers/windowsevent"
+	"github.com/DataDog/datadog-agent/pkg/logs/metrics"
 	"github.com/DataDog/datadog-agent/pkg/logs/pipeline"
 	"github.com/DataDog/datadog-agent/pkg/logs/schedulers"
 	"github.com/DataDog/datadog-agent/pkg/logs/tailers/file"
@@ -58,12 +59,37 @@ func (a *logAgent) SetupPipeline(processingRules []*config.ProcessingRule, wmeta
 // dependent on configuration and connectivity
 func buildEndpoints(coreConfig model.Reader) (*config.Endpoints, error) {
 	httpConnectivity := config.HTTPConnectivityFailure
-	if endpoints, err := config.BuildHTTPEndpointsWithVectorOverride(coreConfig, intakeTrackType, config.AgentJSONIntakeProtocol, config.DefaultIntakeOrigin); err == nil {
+	if endpoints, err := buildHTTPEndpointsForConnectivityCheck(coreConfig); err == nil {
 		httpConnectivity = http.CheckConnectivity(endpoints.Main, coreConfig)
 	}
+
+	// Publish HTTP connectivity check metric
+	if httpConnectivity == config.HTTPConnectivitySuccess {
+		metrics.TlmHTTPConnectivityCheck.Inc("success")
+	} else {
+		metrics.TlmHTTPConnectivityCheck.Inc("failure")
+	}
+
 	return config.BuildEndpointsWithVectorOverride(coreConfig, httpConnectivity, intakeTrackType, config.AgentJSONIntakeProtocol, config.DefaultIntakeOrigin)
 }
 
+// buildHTTPEndpointsForConnectivityCheck builds HTTP endpoints for connectivity testing only
+func buildHTTPEndpointsForConnectivityCheck(coreConfig model.Reader) (*config.Endpoints, error) {
+	return config.BuildHTTPEndpointsWithVectorOverride(coreConfig, intakeTrackType, config.AgentJSONIntakeProtocol, config.DefaultIntakeOrigin)
+}
+
+// checkHTTPConnectivityStatus performs an HTTP connectivity check and returns the status
+func checkHTTPConnectivityStatus(endpoint config.Endpoint, coreConfig model.Reader) config.HTTPConnectivity {
+	return http.CheckConnectivity(endpoint, coreConfig)
+}
+
+// buildHTTPEndpointsForRestart builds HTTP endpoints for restart without connectivity check
+// This is used when we've already verified HTTP connectivity and are upgrading from TCP
+func buildHTTPEndpointsForRestart(coreConfig model.Reader) (*config.Endpoints, error) {
+	// Force HTTP endpoints since we already confirmed connectivity
+	return config.BuildEndpointsWithVectorOverride(coreConfig, config.HTTPConnectivitySuccess, intakeTrackType, config.AgentJSONIntakeProtocol, config.DefaultIntakeOrigin)
+}
+
 // buildPipelineProvider builds a new pipeline provider with the given configuration
 func buildPipelineProvider(a *logAgent, processingRules []*config.ProcessingRule, diagnosticMessageReceiver *diagnostic.BufferedMessageReceiver, destinationsCtx *client.DestinationsContext) pipeline.Provider {
 	pipelineProvider := pipeline.NewProvider(

comp/logs/agent/agentimpl/agent_restart.go

@@ -13,23 +13,43 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/DataDog/datadog-agent/pkg/logs/metrics"
+	"github.com/DataDog/datadog-agent/comp/logs/agent/config"
+	logsmetrics "github.com/DataDog/datadog-agent/pkg/logs/metrics"
 	"github.com/DataDog/datadog-agent/pkg/logs/status"
+	"github.com/DataDog/datadog-agent/pkg/util/backoff"
 	"github.com/DataDog/datadog-agent/pkg/util/startstop"
 )
 
-// restart conducts a partial restart of the logs-agent pipeline.
-// This is used to switch between transport protocols
-// without disrupting the entire agent.
-func (a *logAgent) restart(context.Context) error {
+const (
+	// Transport types for telemetry
+	transportTCP  = "tcp"
+	transportHTTP = "http"
+
+	// Restart status for telemetry
+	restartStatusSuccess = "success"
+	restartStatusFailure = "failure"
+)
+
+// restart conducts a partial restart of the logs-agent pipeline with the provided endpoints.
+// This is used to switch between transport protocols without disrupting the entire agent.
+func (a *logAgent) restart(_ context.Context, newEndpoints *config.Endpoints) error {
 	a.log.Info("Attempting to restart logs-agent pipeline")
 
 	a.restartMutex.Lock()
 	defer a.restartMutex.Unlock()
 
+	// Store current endpoints for rollback if restart fails
+	previousEndpoints := a.endpoints
+
+	// Determine transport type for metrics
+	targetTransport := transportTCP
+	if newEndpoints.UseHTTP {
+		targetTransport = transportHTTP
+	}
+
 	a.log.Info("Gracefully stopping logs-agent")
 
-	timeout := time.Duration(a.config.GetInt("logs_config.stop_grace_period")) * time.Second
+	timeout := a.config.GetDuration("logs_config.stop_grace_period")
 	_, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
@@ -39,24 +59,71 @@ func (a *logAgent) restart(context.Context) error {
 
 	a.log.Info("Re-starting logs-agent...")
 
-	endpoints, err := buildEndpoints(a.config)
+	a.endpoints = newEndpoints
+
+	err := a.setupAgentForRestart()
+	if err != nil {
+		message := fmt.Sprintf("Could not re-start logs-agent: %v", err)
+		a.log.Error(message)
+		a.log.Error("Attempting rollback to previous transport")
+		logsmetrics.TlmRestartAttempt.Inc(restartStatusFailure, targetTransport)
+		return a.rollbackToPreviousTransport(previousEndpoints)
+	}
+
+	a.restartPipeline()
+	logsmetrics.TlmRestartAttempt.Inc(restartStatusSuccess, targetTransport)
+	return nil
+}
+
+// restartWithHTTPUpgrade upgrades the logs-agent pipeline to HTTP transport.
+// This is called by the smart HTTP restart mechanism after HTTP connectivity has been verified.
+//
+// Since HTTP connectivity was verified before calling this function, we commit to HTTP
+// and will keep retrying HTTP even if the upgrade fails. If restart fails, the base
+// restart() function will rollback to TCP temporarily, but this function returns an
+// error to trigger retry - ensuring we eventually upgrade to HTTP since connectivity exists.
+func (a *logAgent) restartWithHTTPUpgrade(ctx context.Context) error {
+	// Build HTTP endpoints since we already verified HTTP connectivity
+	endpoints, err := buildHTTPEndpointsForRestart(a.config)
 	if err != nil {
-		message := fmt.Sprintf("Invalid endpoints: %v", err)
+		message := fmt.Sprintf("Failed to build HTTP endpoints: %v", err)
 		status.AddGlobalError(invalidEndpoints, message)
+		a.log.Error(message)
+		logsmetrics.TlmRestartAttempt.Inc(restartStatusFailure, transportHTTP)
 		return errors.New(message)
 	}
 
-	a.endpoints = endpoints
+	err = a.restart(ctx, endpoints)
+	if err != nil {
+		// Restart failed (may have rolled back to TCP to keep agent functional)
+		// Since we verified HTTP connectivity, return error to trigger retry
+		a.log.Warnf("HTTP upgrade attempt failed: %v - will retry on next attempt", err)
+		return fmt.Errorf("HTTP upgrade failed: %w", err)
+	}
+
+	a.log.Info("Successfully upgraded to HTTP transport")
+	return nil
+}
 
-	err = a.setupAgentForRestart()
+// rollbackToPreviousTransport attempts to restore the agent to its previous working state
+// after a failed transport switch. This ensures the agent continues functioning
+// rather than being left in a broken state.
+func (a *logAgent) rollbackToPreviousTransport(previousEndpoints *config.Endpoints) error {
+	a.log.Warn("Rolling back to previous transport after failed restart")
+
+	a.endpoints = previousEndpoints
+
+	err := a.setupAgentForRestart()
 	if err != nil {
-		message := fmt.Sprintf("Could not re-start logs-agent: %v", err)
+		// This is a critical failure - we can't recover
+		message := fmt.Sprintf("CRITICAL: Failed to rollback to previous transport: %v", err)
 		a.log.Error(message)
 		return errors.New(message)
 	}
 
 	a.restartPipeline()
-	return nil
+	a.log.Info("Successfully rolled back to previous transport")
+	return errors.New("restart failed, rolled back to previous transport")
 }
 
 // setupAgentForRestart configures and rebuilds only the transient components during a restart.
@@ -76,7 +143,7 @@ func (a *logAgent) setupAgentForRestart() error {
 // Unlike startPipeline, this only starts the transient components (destinations, pipeline, launchers)
 // since persistent components (auditor, schedulers, diagnosticMessageReceiver) remain running.
 func (a *logAgent) restartPipeline() {
-	status.Init(a.started, a.endpoints, a.sources, a.tracker, metrics.LogsExpvars)
+	status.Init(a.started, a.endpoints, a.sources, a.tracker, logsmetrics.LogsExpvars)
 
 	starter := startstop.NewStarter(a.destinationsCtx, a.pipelineProvider, a.launchers)
 	starter.Start()
@@ -114,3 +181,115 @@ func (a *logAgent) partialStop() error {
 
 	return nil
 }
+
+// smartHTTPRestart initiates periodic HTTP connectivity checks with exponential backoff
+// to automatically upgrade from TCP to HTTP when connectivity is restored.
+// This only runs when TCP fallback occurred (not when [force_]use_tcp is configured).
+func (a *logAgent) smartHTTPRestart() {
+	// Check if we're eligible for HTTP retry
+	if config.ShouldUseTCP(a.config) {
+		return
+	}
+
+	a.httpRetryMutex.Lock()
+	// Cancel any existing loop to avoid leaks or duplicate retries
+	if a.httpRetryCancel != nil {
+		a.httpRetryCancel()
+	}
+	a.httpRetryCtx, a.httpRetryCancel = context.WithCancel(context.Background())
+	ctx := a.httpRetryCtx
+	a.httpRetryMutex.Unlock()
+
+	a.log.Info("Starting HTTP connectivity retry with exponential backoff")
+
+	// Start background goroutine for periodic HTTP checks
+	go a.httpRetryLoop(ctx)
+}
+
+// httpRetryLoop runs periodic HTTP connectivity checks with exponential backoff
+// Uses a similar backoff strategy as the TCP connection manager:
+// exponential backoff with randomization [2^(n-1), 2^n) seconds, capped at configured max
+func (a *logAgent) httpRetryLoop(ctx context.Context) {
+	maxRetryInterval := config.HTTPConnectivityRetryIntervalMax(a.config)
+	if maxRetryInterval.Seconds() <= 0 {
+		a.log.Warn("HTTP connectivity retry interval max set to 0 seconds, skipping HTTP connectivity retry")
+		return
+	}
+
+	endpoints, err := buildHTTPEndpointsForConnectivityCheck(a.config)
+	if err != nil {
+		a.log.Errorf("Failed to build HTTP endpoints: %v", err)
+		return
+	}
+
+	policy := backoff.NewExpBackoffPolicy(
+		endpoints.Main.BackoffFactor,
+		endpoints.Main.BackoffBase,
+		maxRetryInterval.Seconds(),
+		endpoints.Main.RecoveryInterval,
+		endpoints.Main.RecoveryReset,
+	)
+
+	attempt := 0
+	for {
+		// Calculate backoff interval similar to connection_manager.go
+		backoffDuration := policy.GetBackoffDuration(attempt)
+
+		a.log.Debugf("Next HTTP connectivity check in %v (attempt %d)", backoffDuration, attempt+1)
+
+		select {
+		case <-time.After(backoffDuration):
+			attempt++
+			a.log.Infof("Checking HTTP connectivity (attempt %d)", attempt)
+
+			if a.checkHTTPConnectivity() {
+				a.log.Info("HTTP connectivity restored - initiating upgrade to HTTP transport")
+
+				// Trigger HTTP upgrade. Since HTTP connectivity is verified,
+				// we commit to HTTP and keep retrying if upgrade fails.
+				if err := a.restartWithHTTPUpgrade(ctx); err != nil {
+					a.log.Errorf("HTTP upgrade failed: %v - will retry", err)
+					// Publish retry failure metric
+					logsmetrics.TlmHTTPConnectivityRetryAttempt.Inc("failure")
+					// Continue retrying - HTTP is available, we want to use it
+					continue
+				}
+
+				// Publish retry success metric
+				logsmetrics.TlmHTTPConnectivityRetryAttempt.Inc("success")
+				a.log.Info("Successfully upgraded to HTTP transport")
+				return
+			}
+
+			a.log.Debug("HTTP connectivity check failed - will retry")
+
+		case <-ctx.Done():
+			a.log.Debug("HTTP retry loop stopped")
+			return
+		}
+	}
+}
+
+// checkHTTPConnectivity tests if HTTP endpoints are reachable
+func (a *logAgent) checkHTTPConnectivity() bool {
+	endpoints, err := buildHTTPEndpointsForConnectivityCheck(a.config)
+	if err != nil {
+		a.log.Debugf("Failed to build HTTP endpoints for connectivity check: %v", err)
+		return false
+	}
+
+	connectivity := checkHTTPConnectivityStatus(endpoints.Main, a.config)
+	return connectivity == config.HTTPConnectivitySuccess
+}
+
+// stopHTTPRetry stops the HTTP retry loop
+func (a *logAgent) stopHTTPRetry() {
+	a.httpRetryMutex.Lock()
+	defer a.httpRetryMutex.Unlock()
+
+	if a.httpRetryCancel != nil {
+		a.httpRetryCancel()
+		a.httpRetryCancel = nil
+		a.httpRetryCtx = nil
+	}
+}

comp/logs/agent/agentimpl/agent_restart_serverless.go

@@ -0,0 +1,18 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2023-present Datadog, Inc.
+
+//go:build serverless
+
+package agentimpl
+
+// smartHTTPRestart is a no-op for serverless builds
+func (a *logAgent) smartHTTPRestart() {
+	// No-op: serverless agents don't need HTTP retry functionality
+}
+
+// stopHTTPRetry is a no-op for serverless builds
+func (a *logAgent) stopHTTPRetry() {
+	// No-op: serverless agents don't need HTTP retry functionality
+}