add configurable connect timeout for remote write (#108)

Signed-off-by: Matt Klein <mklein@bitdrift.io>

Author: mattklein123

pulse-metrics/src/admin/server.rs

@@ -15,7 +15,7 @@ use crate::clients::http::{
   PROM_REMOTE_WRITE_HEADERS,
   should_retry,
 };
-use crate::pipeline::config::DEFAULT_REQUEST_TIMEOUT;
+use crate::pipeline::config::{DEFAULT_CONNECT_TIMEOUT, DEFAULT_REQUEST_TIMEOUT};
 use crate::pipeline::outflow::prom::compress_write_request;
 use crate::pipeline::time::{RealTimeProvider, next_flush_interval};
 use anyhow::bail;
@@ -341,6 +341,9 @@ impl MetaStatsEmitter {
             prom_remote_write
               .request_timeout
               .unwrap_duration_or(DEFAULT_REQUEST_TIMEOUT),
+            prom_remote_write
+              .connect_timeout
+              .unwrap_duration_or(DEFAULT_CONNECT_TIMEOUT),
             prom_remote_write.auth.into_option(),
             PROM_REMOTE_WRITE_HEADERS,
             prom_remote_write.request_headers,

pulse-metrics/src/clients/http.rs

@@ -168,20 +168,20 @@ impl HyperHttpRemoteWriteClient {
   pub async fn new(
     endpoint: String,
     timeout: Duration,
+    connect_timeout: Duration,
     auth_config: Option<HttpRemoteWriteAuthConfig>,
     core_request_headers: &[(&str, &str)],
     config_request_headers: Vec<RequestHeader>,
     pool_idle_timeout: Option<Duration>,
   ) -> Result<Self> {
     Ok(Self {
-      // TODO(mattklein123): Make connect timeout configurable.
       inner: Client::builder(TokioExecutor::new())
         .pool_idle_timeout(
           pool_idle_timeout
             .unwrap_or_else(|| 90.seconds())
             .unsigned_abs(),
         )
-        .build(make_tls_connector(250.milliseconds())),
+        .build(make_tls_connector(connect_timeout)),
       endpoint,
       timeout,
       auth: Self::create_auth(auth_config).await?,

pulse-metrics/src/pipeline/config.rs

@@ -95,6 +95,7 @@ impl fmt::Display for PipelineRouteType {
 }
 
 pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::seconds(10);
+pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::milliseconds(250);
 
 #[must_use]
 pub const fn default_max_in_flight() -> u64 {

pulse-metrics/src/pipeline/outflow/http/remote_write.rs

@@ -18,7 +18,11 @@ use crate::clients::http::{
   should_retry,
 };
 use crate::clients::retry::Retry;
-use crate::pipeline::config::{DEFAULT_REQUEST_TIMEOUT, default_max_in_flight};
+use crate::pipeline::config::{
+  DEFAULT_CONNECT_TIMEOUT,
+  DEFAULT_REQUEST_TIMEOUT,
+  default_max_in_flight,
+};
 use crate::pipeline::outflow::http::retry_offload::maybe_queue_for_retry;
 use crate::pipeline::outflow::{OutflowFactoryContext, OutflowStats, PipelineOutflow};
 use crate::pipeline::time::RealTimeProvider;
@@ -142,6 +146,7 @@ pub struct HttpRemoteWriteOutflow {
 impl HttpRemoteWriteOutflow {
   pub(crate) async fn new(
     request_timeout: MessageField<Duration>,
+    connect_timeout: MessageField<Duration>,
     retry_policy: RetryPolicy,
     max_in_flight: Option<u64>,
     batch_router: Arc<dyn BatchRouter>,
@@ -154,10 +159,12 @@ impl HttpRemoteWriteOutflow {
     request_deserializer: RequestDeserializer,
   ) -> anyhow::Result<Arc<Self>> {
     let request_timeout = request_timeout.unwrap_duration_or(DEFAULT_REQUEST_TIMEOUT);
+    let connect_timeout = connect_timeout.unwrap_duration_or(DEFAULT_CONNECT_TIMEOUT);
     let client = Arc::new(
       HyperHttpRemoteWriteClient::new(
         send_to,
         request_timeout,
+        connect_timeout,
         auth_config,
         core_request_headers,
         config_request_headers,

pulse-metrics/src/pipeline/outflow/otlp/mod.rs

@@ -82,6 +82,7 @@ pub async fn make_otlp_outflow(
   let compression = config.compression.enum_value_or_default();
   HttpRemoteWriteOutflow::new(
     config.request_timeout,
+    config.connect_timeout,
     config.retry_policy.unwrap_or_default(),
     config.max_in_flight,
     batch_router,

pulse-metrics/src/pipeline/outflow/prom/mod.rs

@@ -103,6 +103,7 @@ pub async fn make_prom_outflow(
   );
   HttpRemoteWriteOutflow::new(
     config.request_timeout,
+    config.connect_timeout,
     config.retry_policy.unwrap_or_default(),
     config.max_in_flight,
     batch_router,

pulse-promcli/src/main.rs

@@ -104,6 +104,7 @@ async fn main() -> anyhow::Result<()> {
   let client = HyperHttpRemoteWriteClient::new(
     options.endpoint,
     10.seconds(),
+    250.milliseconds(),
     None,
     PROM_REMOTE_WRITE_HEADERS,
     vec![],

pulse-protobuf/proto/pulse/config/outflow/v1/otlp.proto

@@ -27,6 +27,9 @@ message OtlpClientConfig {
   // The request timeout. Defaults to 10s.
   google.protobuf.Duration request_timeout = 2 [(validate.rules).duration.gt = {}];
 
+  // The connection timeout for establishing a new upstream connection. Defaults to 250ms.
+  google.protobuf.Duration connect_timeout = 11 [(validate.rules).duration.gt = {}];
+
   // The maximum number of inflight requests. Defaults to 16.
   optional uint64 max_in_flight = 3;
 

pulse-protobuf/proto/pulse/config/outflow/v1/prom_remote_write.proto

@@ -35,6 +35,9 @@ message PromRemoteWriteClientConfig {
   // The request timeout. Defaults to 10s.
   google.protobuf.Duration request_timeout = 2 [(validate.rules).duration.gt = {}];
 
+  // The connection timeout for establishing a new upstream connection. Defaults to 250ms.
+  google.protobuf.Duration connect_timeout = 14 [(validate.rules).duration.gt = {}];
+
   // The maximum number of inflight requests. Defaults to 16.
   optional uint64 max_in_flight = 3;
 

pulse-protobuf/src/protos/pulse/config/outflow/v1/otlp.rs

@@ -39,6 +39,8 @@ pub struct OtlpClientConfig {
     pub send_to: ::protobuf::Chars,
     // @@protoc_insertion_point(field:pulse.config.outflow.v1.OtlpClientConfig.request_timeout)
     pub request_timeout: ::protobuf::MessageField<::protobuf::well_known_types::duration::Duration>,
+    // @@protoc_insertion_point(field:pulse.config.outflow.v1.OtlpClientConfig.connect_timeout)
+    pub connect_timeout: ::protobuf::MessageField<::protobuf::well_known_types::duration::Duration>,
     // @@protoc_insertion_point(field:pulse.config.outflow.v1.OtlpClientConfig.max_in_flight)
     pub max_in_flight: ::std::option::Option<u64>,
     // @@protoc_insertion_point(field:pulse.config.outflow.v1.OtlpClientConfig.queue_policy)
@@ -72,7 +74,7 @@ impl OtlpClientConfig {
     }
 
     fn generated_message_descriptor_data() -> ::protobuf::reflect::GeneratedMessageDescriptorData {
-        let mut fields = ::std::vec::Vec::with_capacity(10);
+        let mut fields = ::std::vec::Vec::with_capacity(11);
         let mut oneofs = ::std::vec::Vec::with_capacity(0);
         fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>(
             "send_to",
@@ -84,6 +86,11 @@ impl OtlpClientConfig {
             |m: &OtlpClientConfig| { &m.request_timeout },
             |m: &mut OtlpClientConfig| { &mut m.request_timeout },
         ));
+        fields.push(::protobuf::reflect::rt::v2::make_message_field_accessor::<_, ::protobuf::well_known_types::duration::Duration>(
+            "connect_timeout",
+            |m: &OtlpClientConfig| { &m.connect_timeout },
+            |m: &mut OtlpClientConfig| { &mut m.connect_timeout },
+        ));
         fields.push(::protobuf::reflect::rt::v2::make_option_accessor::<_, _>(
             "max_in_flight",
             |m: &OtlpClientConfig| { &m.max_in_flight },
@@ -148,6 +155,9 @@ impl ::protobuf::Message for OtlpClientConfig {
                 18 => {
                     ::protobuf::rt::read_singular_message_into_field(is, &mut self.request_timeout)?;
                 },
+                90 => {
+                    ::protobuf::rt::read_singular_message_into_field(is, &mut self.connect_timeout)?;
+                },
                 24 => {
                     self.max_in_flight = ::std::option::Option::Some(is.read_uint64()?);
                 },
@@ -191,6 +201,10 @@ impl ::protobuf::Message for OtlpClientConfig {
             let len = v.compute_size();
             my_size += 1 + ::protobuf::rt::compute_raw_varint64_size(len) + len;
         }
+        if let Some(v) = self.connect_timeout.as_ref() {
+            let len = v.compute_size();
+            my_size += 1 + ::protobuf::rt::compute_raw_varint64_size(len) + len;
+        }
         if let Some(v) = self.max_in_flight {
             my_size += ::protobuf::rt::uint64_size(3, v);
         }
@@ -231,6 +245,9 @@ impl ::protobuf::Message for OtlpClientConfig {
         if let Some(v) = self.request_timeout.as_ref() {
             ::protobuf::rt::write_message_field_with_cached_size(2, v, os)?;
         }
+        if let Some(v) = self.connect_timeout.as_ref() {
+            ::protobuf::rt::write_message_field_with_cached_size(11, v, os)?;
+        }
         if let Some(v) = self.max_in_flight {
             os.write_uint64(3, v)?;
         }
@@ -274,6 +291,7 @@ impl ::protobuf::Message for OtlpClientConfig {
     fn clear(&mut self) {
         self.send_to.clear();
         self.request_timeout.clear();
+        self.connect_timeout.clear();
         self.max_in_flight = ::std::option::Option::None;
         self.queue_policy.clear();
         self.batch_max_samples = ::std::option::Option::None;
@@ -289,6 +307,7 @@ impl ::protobuf::Message for OtlpClientConfig {
         static instance: OtlpClientConfig = OtlpClientConfig {
             send_to: ::protobuf::Chars::new(),
             request_timeout: ::protobuf::MessageField::none(),
+            connect_timeout: ::protobuf::MessageField::none(),
             max_in_flight: ::std::option::Option::None,
             queue_policy: ::protobuf::MessageField::none(),
             batch_max_samples: ::std::option::Option::None,
@@ -390,23 +409,25 @@ static file_descriptor_proto_data: &'static [u8] = b"\
     \x1a\x1egoogle/protobuf/duration.proto\x1a#pulse/config/common/v1/common\
     .proto\x1a\"pulse/config/common/v1/retry.proto\x1a,pulse/config/outflow/\
     v1/outflow_common.proto\x1a*pulse/config/outflow/v1/queue_policy.proto\
-    \x1a\x17validate/validate.proto\"\xf3\x05\n\x10OtlpClientConfig\x12\x20\
+    \x1a\x17validate/validate.proto\"\xc1\x06\n\x10OtlpClientConfig\x12\x20\
     \n\x07send_to\x18\x01\x20\x01(\tR\x06sendToB\x07\xfaB\x04r\x02\x10\x01\
     \x12L\n\x0frequest_timeout\x18\x02\x20\x01(\x0b2\x19.google.protobuf.Dur\
-    ationR\x0erequestTimeoutB\x08\xfaB\x05\xaa\x01\x02*\0\x12'\n\rmax_in_fli\
-    ght\x18\x03\x20\x01(\x04H\0R\x0bmaxInFlight\x88\x01\x01\x12G\n\x0cqueue_\
-    policy\x18\x04\x20\x01(\x0b2$.pulse.config.outflow.v1.QueuePolicyR\x0bqu\
-    euePolicy\x12/\n\x11batch_max_samples\x18\x05\x20\x01(\x04H\x01R\x0fbatc\
-    hMaxSamples\x88\x01\x01\x12F\n\x04auth\x18\x06\x20\x01(\x0b22.pulse.conf\
-    ig.outflow.v1.HttpRemoteWriteAuthConfigR\x04auth\x12O\n\x0frequest_heade\
-    rs\x18\x07\x20\x03(\x0b2&.pulse.config.outflow.v1.RequestHeaderR\x0erequ\
-    estHeaders\x12F\n\x0cretry_policy\x18\x08\x20\x01(\x0b2#.pulse.config.co\
-    mmon.v1.RetryPolicyR\x0bretryPolicy\x12[\n\x0bcompression\x18\t\x20\x01(\
-    \x0e29.pulse.config.outflow.v1.OtlpClientConfig.OtlpCompressionR\x0bcomp\
-    ression\x12=\n\x1bconvert_names_to_prometheus\x18\n\x20\x01(\x08R\x18con\
-    vertNamesToPrometheus\"'\n\x0fOtlpCompression\x12\n\n\x06SNAPPY\x10\0\
-    \x12\x08\n\x04NONE\x10\x01B\x10\n\x0e_max_in_flightB\x14\n\x12_batch_max\
-    _samplesb\x06proto3\
+    ationR\x0erequestTimeoutB\x08\xfaB\x05\xaa\x01\x02*\0\x12L\n\x0fconnect_\
+    timeout\x18\x0b\x20\x01(\x0b2\x19.google.protobuf.DurationR\x0econnectTi\
+    meoutB\x08\xfaB\x05\xaa\x01\x02*\0\x12'\n\rmax_in_flight\x18\x03\x20\x01\
+    (\x04H\0R\x0bmaxInFlight\x88\x01\x01\x12G\n\x0cqueue_policy\x18\x04\x20\
+    \x01(\x0b2$.pulse.config.outflow.v1.QueuePolicyR\x0bqueuePolicy\x12/\n\
+    \x11batch_max_samples\x18\x05\x20\x01(\x04H\x01R\x0fbatchMaxSamples\x88\
+    \x01\x01\x12F\n\x04auth\x18\x06\x20\x01(\x0b22.pulse.config.outflow.v1.H\
+    ttpRemoteWriteAuthConfigR\x04auth\x12O\n\x0frequest_headers\x18\x07\x20\
+    \x03(\x0b2&.pulse.config.outflow.v1.RequestHeaderR\x0erequestHeaders\x12\
+    F\n\x0cretry_policy\x18\x08\x20\x01(\x0b2#.pulse.config.common.v1.RetryP\
+    olicyR\x0bretryPolicy\x12[\n\x0bcompression\x18\t\x20\x01(\x0e29.pulse.c\
+    onfig.outflow.v1.OtlpClientConfig.OtlpCompressionR\x0bcompression\x12=\n\
+    \x1bconvert_names_to_prometheus\x18\n\x20\x01(\x08R\x18convertNamesToPro\
+    metheus\"'\n\x0fOtlpCompression\x12\n\n\x06SNAPPY\x10\0\x12\x08\n\x04NON\
+    E\x10\x01B\x10\n\x0e_max_in_flightB\x14\n\x12_batch_max_samplesb\x06prot\
+    o3\
 ";
 
 /// `FileDescriptorProto` object which was a source for this generated file