feature(enterprise): configurable app name

.github/actions/spelling/expect.txt

@@ -984,6 +984,7 @@ skywalking
 slashin
 slf
 slideover
+slugified
 smallvec
 smartphone
 Smol

src/config/sink.rs

@@ -241,6 +241,9 @@ pub struct SinkContext {
     pub globals: GlobalOptions,
     pub proxy: ProxyConfig,
     pub schema: schema::Options,
+
+    #[cfg(feature = "enterprise")]
+    pub enterprise_enabled: bool,
 }
 
 impl SinkContext {

src/http.rs

@@ -82,9 +82,15 @@ where
         let proxy_connector = build_proxy_connector(tls_settings.into(), proxy_config)?;
         let client = client_builder.build(proxy_connector.clone());
 
+        let app_name = crate::get_app_name(
+            // TODO: correctly set this field.
+            // Note that Vector Enterprise is deprecated so the effort does not seem worthwhile.
+            #[cfg(feature = "enterprise")]
+            false,
+        );
         let version = crate::get_version();
-        let user_agent = HeaderValue::from_str(&format!("Vector/{}", version))
-            .expect("Invalid header value for version!");
+        let user_agent = HeaderValue::from_str(&format!("{}/{}", app_name, version))
+            .expect("Invalid header value for user-agent!");
 
         Ok(HttpClient {
             client,

src/lib.rs

@@ -124,6 +124,35 @@ pub use source_sender::SourceSender;
 pub use vector_common::{shutdown, Error, Result};
 pub use vector_core::{event, metrics, schema, tcp, tls};
 
+/// The name used to identify this Vector application.
+///
+/// This can be set at compile-time through the VECTOR_APP_NAME env variable.
+/// Defaults to "Vector".
+pub fn get_app_name(#[cfg(feature = "enterprise")] enterprise_enabled: bool) -> &'static str {
+    #[cfg(not(feature = "enterprise"))]
+    let default_app_name = "Vector";
+    #[cfg(feature = "enterprise")]
+    let default_app_name = if enterprise_enabled {
+        "Vector Enterprise"
+    } else {
+        "Vector"
+    };
+
+    option_env!("VECTOR_APP_NAME").unwrap_or(default_app_name)
+}
+
+/// Returns a slugified version of the name used to identify this Vector application.
+///
+/// Defaults to "vector".
+pub fn get_slugified_app_name(#[cfg(feature = "enterprise")] enterprise_enabled: bool) -> String {
+    get_app_name(
+        #[cfg(feature = "enterprise")]
+        enterprise_enabled,
+    )
+    .to_lowercase()
+    .replace(' ', "-")
+}
+
 /// The current version of Vector in simplified format.
 /// `<version-number>-nightly`.
 pub fn vector_version() -> impl std::fmt::Display {

src/sinks/datadog/logs/config.rs

@@ -114,7 +114,11 @@ impl DatadogLogsConfig {
         self.get_uri().scheme_str().unwrap_or("http").to_string()
     }
 
-    pub fn build_processor(&self, client: HttpClient) -> crate::Result<VectorSink> {
+    pub fn build_processor(
+        &self,
+        client: HttpClient,
+        dd_evp_origin: String,
+    ) -> crate::Result<VectorSink> {
         let default_api_key: Arc<str> = Arc::from(self.dd_common.default_api_key.inner());
         let request_limits = self.request.tower.unwrap_with(&Default::default());
 
@@ -133,6 +137,7 @@ impl DatadogLogsConfig {
                 client,
                 self.get_uri(),
                 self.request.headers.clone(),
+                dd_evp_origin,
             )?);
 
         let encoding = self.encoding.clone();
@@ -169,7 +174,11 @@ impl SinkConfig for DatadogLogsConfig {
             .dd_common
             .build_healthcheck(client.clone(), self.region.as_ref())?;
 
-        let sink = self.build_processor(client)?;
+        let slugified_app_name = crate::get_slugified_app_name(
+            #[cfg(feature = "enterprise")]
+            cx.enterprise_enabled,
+        );
+        let sink = self.build_processor(client, slugified_app_name)?;
 
         Ok((sink, healthcheck))
     }

src/sinks/datadog/logs/service.rs

@@ -95,20 +95,33 @@ pub struct LogApiService {
     client: HttpClient,
     uri: Uri,
     user_provided_headers: IndexMap<HeaderName, HeaderValue>,
+    default_headers: IndexMap<HeaderName, HeaderValue>,
 }
 
 impl LogApiService {
     pub fn new(
         client: HttpClient,
         uri: Uri,
         headers: IndexMap<String, String>,
+        dd_evp_origin: String,
     ) -> crate::Result<Self> {
-        let headers = validate_headers(&headers)?;
+        let user_provided_headers = validate_headers(&headers)?;
+
+        // Note that these headers cannot be overriden by the user.
+        let default_headers = &[
+            (CONTENT_TYPE.to_string(), "application/json".to_string()),
+            ("DD-EVP-ORIGIN".to_string(), dd_evp_origin),
+            ("DD-EVP-ORIGIN-VERSION".to_string(), crate::get_version()),
+        ]
+        .into_iter()
+        .collect();
+        let default_headers = validate_headers(default_headers)?;
 
         Ok(Self {
             client,
             uri,
-            user_provided_headers: headers,
+            user_provided_headers,
+            default_headers,
         })
     }
 }
@@ -126,11 +139,8 @@ impl Service<LogApiRequest> for LogApiService {
     // Emission of Error internal event is handled upstream by the caller
     fn call(&mut self, mut request: LogApiRequest) -> Self::Future {
         let mut client = self.client.clone();
-        let http_request = Request::post(&self.uri)
-            .header(CONTENT_TYPE, "application/json")
-            .header("DD-EVP-ORIGIN", "vector")
-            .header("DD-EVP-ORIGIN-VERSION", crate::get_version())
-            .header("DD-API-KEY", request.api_key.to_string());
+        let http_request =
+            Request::post(&self.uri).header("DD-API-KEY", request.api_key.to_string());
 
         let http_request = if let Some(ce) = request.compression.content_encoding() {
             http_request.header(CONTENT_ENCODING, ce)
@@ -149,6 +159,9 @@ impl Service<LogApiRequest> for LogApiService {
                 // Replace rather than append to any existing header values
                 headers.insert(name, value.clone());
             }
+            for (name, value) in &self.default_headers {
+                headers.insert(name, value.clone());
+            }
         }
 
         let http_request = http_request

src/sinks/datadog/logs/tests.rs

@@ -402,14 +402,12 @@ async fn enterprise_headers_v1() {
 }
 
 async fn enterprise_headers_inner(api_status: ApiStatus) {
-    let (mut config, cx) = load_sink::<DatadogLogsConfig>(indoc! {r#"
+    let (mut config, mut cx) = load_sink::<DatadogLogsConfig>(indoc! {r#"
             default_api_key = "atoken"
             compression = "none"
-
-            [request]
-            headers.DD-EVP-ORIGIN = "vector-enterprise"
         "#})
     .unwrap();
+    cx.enterprise_enabled = true;
 
     let addr = next_addr();
     // Swap out the endpoint so we can force send it to our local server

src/topology/builder.rs

@@ -568,6 +568,14 @@ impl<'a> Builder<'a> {
                 globals: self.config.global.clone(),
                 proxy: ProxyConfig::merge_with_env(&self.config.global.proxy, sink.proxy()),
                 schema: self.config.schema,
+
+                #[cfg(feature = "enterprise")]
+                enterprise_enabled: self
+                    .config
+                    .enterprise
+                    .as_ref()
+                    .map(|e| e.enabled)
+                    .unwrap_or_default(),
             };
 
             let (sink, healthcheck) = match sink.inner.build(cx).await {