feat(tls): add tls_handshake_timeout config option (#2303)

## Human Summary 

Adds support for `tls_handlshake_timeout` using [newly documented method in tokio-rustls](rustls/tokio-rustls#187). This is roughly follows #178 but based on latest main.

It doesn't handle configuring the connect timeout when using an HTTPS proxy. That requires changes to hyper-http-proxy so I'll follow up with that.

Closes: #178

## AI Summary 
TLS handshakes to the Datadog intake have no built-in timeout independent of the overall request timeout: `hyper_rustls`'s connector fuses the transport connect and the TLS handshake into a single opaque future, so a stalled handshake (e.g. a peer that accepts the TCP connection but never completes the TLS negotiation) is only bounded by `forwarder_timeout`, which is meant to bound the whole request, not just the handshake. This adds a `tls_handshake_timeout` config option by having the HTTP client connector own the TLS layer directly, so it can time out just the handshake step and still distinguish that failure mode from a slow request. This picks up the intent of #1819, an older PR for the same issue, rewritten against the current typed configuration system rather than resurrected via rebase.

```mermaid
sequenceDiagram
    participant Before as Before (hyper_rustls::HttpsConnector)
    participant After as After (owned TLS layer)

    Note over Before: connect + handshake fused into one future
    Before->>Before: TCP connect
    Before->>Before: TLS handshake
    Note over Before: only forwarder_timeout bounds both steps combined

    Note over After: connect and handshake are separate steps
    After->>After: TCP connect (connect_timeout)
    After->>After: TLS handshake (tls_handshake_timeout, new)
    Note over After: a stalled handshake times out on its own,<br/>without racing the whole request
```

## Test plan

- [x] Added `tls_handshake_timeout` to the Datadog config schema overlay (`support: full`) and wired it through the typed config system (`DatadogTranslator`, `SalukiConfiguration`) and the legacy `ForwarderConfiguration` facet-based config, both consumed by the HTTP client builder.
- [x] Added/updated unit tests in `saluki-io`'s `conn.rs` for the new connector split (ALPN protocol selection, including an explicit `http/1.1` ALPN advertisement for `HttpProtocol::Http1` to avoid an ALPN regression from the previous implicit behavior).
- [x] Existing `config_smoke::smoke_test` in `saluki-components` (`ForwarderConfiguration`) exercises the new field's default/deserialization against the config registry.
- [x] Updated classifier unit tests in `datadog-agent-config` that previously used `tls_handshake_timeout` as an example unsupported/incompatible key, substituting other still-unsupported keys since this key is now fully supported.

## Known limitation

Connections made through `proxy_https` bypass this connector and aren't covered by `tls_handshake_timeout` (flagged on the original PR). Left out of scope here; can be addressed separately if needed.


Co-authored-by: jesse.szwedko <jesse.szwedko@datadoghq.com>

Author: jszwedko

docs/agent-data-plane/configuration/configuration.md

@@ -29,7 +29,6 @@ tracking.
 | -------------------------------------------- | ----------------------------------------------- | ------- |
 | `dogstatsd_experimental_http.enabled`        | Enable experimental HTTP/H2C DSD listener       | [#1682] |
 | `dogstatsd_experimental_http.listen_address` | Bind address for experimental HTTP DSD listener | [#1682] |
-| `tls_handshake_timeout`                      | HTTP TLS handshake timeout                      | [#178]  |
 
 <!-- section:unsupported-not-planned -->
 ### Not Planned
@@ -801,6 +800,7 @@ Both commands scrub recognized secret values before writing JSON to standard out
 | `syslog_rfc`                                                   | Use RFC-style syslog header                        |
 | `syslog_uri`                                                   | Syslog destination URI                             |
 | `telemetry.dogstatsd_origin`                                   | Per-origin processed-metrics telemetry             |
+| `tls_handshake_timeout`                                        | HTTP TLS handshake timeout                         |
 | `use_proxy_for_cloud_metadata`                                 | Proxy cloud metadata endpoints                     |
 | `use_v2_api.series`                                            | Send series via V2 protobuf endpoint               |
 | `use_v3_api.series.enabled`                                    | Global V3 series mode                              |

lib/agent-data-plane-config-system/src/translators/datadog_translator.rs

@@ -841,6 +841,10 @@ impl DatadogConfigWitness for DatadogTranslator<'_> {
         self.config.shared.endpoints.tls.min_tls_version = value;
     }
 
+    fn consume_tls_handshake_timeout(&mut self, value: Duration) {
+        self.config.shared.endpoints.tls.handshake_timeout = value;
+    }
+
     fn consume_multi_region_failover_api_key(&mut self, value: String) {
         self.config.domains.multi_region_failover.api_key = non_empty(value);
     }

lib/agent-data-plane-config/src/shared.rs

@@ -120,6 +120,12 @@ pub struct Tls {
 
     /// Path to which TLS session keys are logged, for debugging.
     pub sslkeylogfile: String,
+
+    /// Timeout for completing the TLS handshake after a connection is established.
+    ///
+    /// Defaults to 10 seconds. Bounds only the handshake step, distinct from the overall request timeout. A value
+    /// of zero disables the handshake-specific deadline, leaving the overall request timeout as the only bound.
+    pub handshake_timeout: Duration,
 }
 
 /// Payload compression settings applied before transmission.

lib/datadog-agent/config-testing/build/registry_gen.rs

@@ -172,6 +172,7 @@ static GOLDEN_ORDER: &[(&str, &[&str])] = &[
             "forwarder_storage_max_size_in_bytes",
             "forwarder_storage_path",
             "forwarder_outdated_file_in_days",
+            "tls_handshake_timeout",
         ],
     ),
     (

lib/datadog-agent/config-testing/src/config_registry/forwarder.rs

@@ -313,6 +313,17 @@ crate::declare_annotations! {
         test_json: None,
         pipeline_affinity: PipelineAffinity::CrossCutting,
     };
+    /// `tls_handshake_timeout`-HTTP TLS handshake timeout
+    TLS_HANDSHAKE_TIMEOUT = SalukiAnnotation {
+        schema: &schema::TLS_HANDSHAKE_TIMEOUT,
+        support_level: SupportLevel::Full,
+        additional_yaml_paths: &[],
+        env_var_override: None,
+        used_by: &[structs::FORWARDER_CONFIGURATION],
+        value_type_override: None,
+        test_json: None,
+        pipeline_affinity: PipelineAffinity::CrossCutting,
+    };
     /// `forwarder_apikey_validation_interval`-API key check interval (minutes)
     FORWARDER_APIKEY_VALIDATION_INTERVAL = SalukiAnnotation {
         schema: &schema::FORWARDER_APIKEY_VALIDATION_INTERVAL,

lib/datadog-agent/config-testing/src/config_registry/unsupported.rs

@@ -60,17 +60,6 @@ crate::declare_annotations! {
         test_json: None,
         pipeline_affinity: PipelineAffinity::CrossCutting,
     };
-    /// `tls_handshake_timeout`-HTTP TLS handshake timeout
-    TLS_HANDSHAKE_TIMEOUT = SalukiAnnotation {
-        schema: &schema::TLS_HANDSHAKE_TIMEOUT,
-        support_level: SupportLevel::Incompatible(Severity::Medium),
-        additional_yaml_paths: &[],
-        env_var_override: None,
-        used_by: &[],
-        value_type_override: None,
-        test_json: None,
-        pipeline_affinity: PipelineAffinity::CrossCutting,
-    };
     /// `aggregator_buffer_size`-Channel buffer depth for aggregator queues
     AGGREGATOR_BUFFER_SIZE = SalukiAnnotation {
         schema: &schema::AGGREGATOR_BUFFER_SIZE,

lib/datadog-agent/config/schema/schema_overlay.yaml

@@ -2593,12 +2593,18 @@ inventory:
         config_registry_filename: dogstatsd.rs
 
   tls_handshake_timeout:
-    support: none
-    severity: medium
-    planned: true
+    support: full
     pipelines: [cross_cutting]
     description: "HTTP TLS handshake timeout"
-    documentation: "Existing request timeout covers the gap."
+    documentation: |
+      ADP applies `tls_handshake_timeout` to the TLS handshake step of outbound Datadog intake connections,
+      independent of `forwarder_timeout`, which bounds the full request. ADP owns the TLS layer directly rather than
+      delegating the handshake to an opaque connect-and-handshake future, so it can time out the handshake without
+      also aborting an in-progress TCP connect.
+    test_support:
+      used_by: [ForwarderConfiguration]
+      additional_attributes:
+        config_registry_filename: forwarder.rs
     issue: "#178"
 
   use_dogstatsd:

lib/datadog-agent/config/src/classifier/classifier.rs

@@ -115,7 +115,9 @@ mod tests {
     #[test]
     fn incompatible_non_default() {
         let c = classifier();
-        let result = c.classify("tls_handshake_timeout", &Value::Number(999.into())).unwrap();
+        let result = c
+            .classify("dogstatsd_stats_buffer", &Value::Number(999.into()))
+            .unwrap();
         assert!(matches!(result.support_level, SupportLevel::Incompatible(_)));
         assert!(!result.is_default);
     }
@@ -139,8 +141,10 @@ mod tests {
     #[test]
     fn duration_default_null_is_not_default() {
         let c = classifier();
-        // tls_handshake_timeout has a duration default (10s); a null value can't be normalized.
-        let result = c.classify("tls_handshake_timeout", &Value::Null).unwrap();
+        // dogstatsd_packet_buffer_flush_timeout has a duration default (100ms); a null value can't be normalized.
+        let result = c
+            .classify("dogstatsd_packet_buffer_flush_timeout", &Value::Null)
+            .unwrap();
         assert!(!result.is_default);
     }
 
@@ -149,14 +153,17 @@ mod tests {
         let c = classifier();
         // Neither an empty string nor arbitrary text parses as a duration, so neither matches.
         assert!(
-            !c.classify("tls_handshake_timeout", &Value::String("".into()))
+            !c.classify("dogstatsd_packet_buffer_flush_timeout", &Value::String("".into()))
                 .unwrap()
                 .is_default
         );
         assert!(
-            !c.classify("tls_handshake_timeout", &Value::String("something".into()))
-                .unwrap()
-                .is_default
+            !c.classify(
+                "dogstatsd_packet_buffer_flush_timeout",
+                &Value::String("something".into())
+            )
+            .unwrap()
+            .is_default
         );
     }
 
@@ -165,15 +172,15 @@ mod tests {
         let c = classifier();
         // The default is also matched when supplied as a Go duration string rather than nanoseconds.
         let result = c
-            .classify("tls_handshake_timeout", &Value::String("10s".into()))
+            .classify("dogstatsd_packet_buffer_flush_timeout", &Value::String("100ms".into()))
             .unwrap();
         assert!(result.is_default);
     }
 
     #[test]
     fn incompatible_severity_levels() {
         let c = classifier();
-        let result = c.classify("tls_handshake_timeout", &Value::Number(30.into())).unwrap();
+        let result = c.classify("dogstatsd_stats_buffer", &Value::Number(30.into())).unwrap();
         assert!(matches!(
             result.support_level,
             SupportLevel::Incompatible(Severity::Medium)
@@ -183,20 +190,26 @@ mod tests {
     #[test]
     fn duration_default_as_nanoseconds_is_default() {
         let c = classifier();
-        // The Agent transmits tls_handshake_timeout (schema default "10s") as integer nanoseconds.
-        // The classifier must recognize this as the default and not flag it as an override.
+        // The Agent transmits dogstatsd_packet_buffer_flush_timeout (schema default "100ms") as integer
+        // nanoseconds. The classifier must recognize this as the default and not flag it as an override.
         let result = c
-            .classify("tls_handshake_timeout", &Value::Number(10_000_000_000i64.into()))
+            .classify(
+                "dogstatsd_packet_buffer_flush_timeout",
+                &Value::Number(100_000_000i64.into()),
+            )
             .unwrap();
         assert!(result.is_default);
     }
 
     #[test]
     fn duration_non_default_nanoseconds_is_not_default() {
         let c = classifier();
-        // 5s in nanoseconds is not the 10s default.
+        // 5ms in nanoseconds is not the 100ms default.
         let result = c
-            .classify("tls_handshake_timeout", &Value::Number(5_000_000_000i64.into()))
+            .classify(
+                "dogstatsd_packet_buffer_flush_timeout",
+                &Value::Number(5_000_000i64.into()),
+            )
             .unwrap();
         assert!(!result.is_default);
     }

lib/datadog-agent/config/src/generated/classifier_data.rs

@@ -319,13 +319,6 @@ pub(crate) static CLASSIFIER_ENTRIES: &[ClassifierEntry] = &[
         pipeline_affinity: PipelineAffinity::Pipelines(&[Pipeline::DogStatsD]),
         default: DefaultValue::Json("[]"),
     },
-    ClassifierEntry {
-        yaml_path: "tls_handshake_timeout",
-        aliases: &[],
-        support_level: SupportLevel::Incompatible(Severity::Medium),
-        pipeline_affinity: PipelineAffinity::CrossCutting,
-        default: DefaultValue::DurationNanos(10000000000),
-    },
     ClassifierEntry {
         yaml_path: "use_dogstatsd",
         aliases: &[],

lib/datadog-agent/config/src/generated/datadog_configuration.rs

@@ -412,6 +412,13 @@ pub struct DatadogConfiguration {
     #[serde(default)]
     pub telemetry: Telemetry,
 
+    #[serde(
+        default = "duration_defaults::tls_handshake_timeout",
+
+        deserialize_with = "crate::duration_de::deserialize_go_duration"
+    )]
+    pub tls_handshake_timeout: std::time::Duration,
+
     #[serde(default)]
     pub use_proxy_for_cloud_metadata: bool,
 
@@ -561,6 +568,7 @@ impl Default for DatadogConfiguration {
             syslog_rfc: Default::default(),
             syslog_uri: Default::default(),
             telemetry: Default::default(),
+            tls_handshake_timeout: duration_defaults::tls_handshake_timeout(),
             use_proxy_for_cloud_metadata: Default::default(),
             use_v2_api: Default::default(),
             use_v3_api: Default::default(),
@@ -1869,4 +1877,7 @@ mod duration_defaults {
     pub(super) fn expected_tags_duration() -> std::time::Duration {
         std::time::Duration::from_nanos(0)
     }
+    pub(super) fn tls_handshake_timeout() -> std::time::Duration {
+        std::time::Duration::from_nanos(10000000000)
+    }
 }