feature(enterprise): configurable app name

.github/actions/spelling/expect.txt

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

src/app.rs

@@ -535,6 +535,18 @@ fn build_enterprise(
     config: &mut Config,
     config_paths: Vec<ConfigPath>,
 ) -> Result<Option<EnterpriseReporter<BoxFuture<'static, ()>>>, ExitCode> {
+    use crate::ENTERPRISE_ENABLED;
+
+    ENTERPRISE_ENABLED
+        .set(
+            config
+                .enterprise
+                .as_ref()
+                .map(|e| e.enabled)
+                .unwrap_or_default(),
+        )
+        .expect("double initialization of enterprise enabled flag");
+
     match EnterpriseMetadata::try_from(&*config) {
         Ok(metadata) => {
             let enterprise = EnterpriseReporter::new();

src/config/sink.rs

@@ -235,12 +235,27 @@ pub trait SinkConfig: DynClone + NamedComponent + core::fmt::Debug + Send + Sync
 
 dyn_clone::clone_trait_object!(SinkConfig);
 
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug)]
 pub struct SinkContext {
     pub healthcheck: SinkHealthcheckOptions,
     pub globals: GlobalOptions,
     pub proxy: ProxyConfig,
     pub schema: schema::Options,
+    pub app_name: String,
+    pub app_name_slug: String,
+}
+
+impl Default for SinkContext {
+    fn default() -> Self {
+        Self {
+            healthcheck: Default::default(),
+            globals: Default::default(),
+            proxy: Default::default(),
+            schema: Default::default(),
+            app_name: crate::get_app_name().to_string(),
+            app_name_slug: crate::get_slugified_app_name(),
+        }
+    }
 }
 
 impl SinkContext {

src/http.rs

@@ -82,9 +82,10 @@ 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();
         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,38 @@ pub use source_sender::SourceSender;
 pub use vector_common::{shutdown, Error, Result};
 pub use vector_core::{event, metrics, schema, tcp, tls};
 
+static APP_NAME_SLUG: std::sync::OnceLock<String> = std::sync::OnceLock::new();
+
+/// Flag denoting whether or not enterprise features are enabled.
+#[cfg(feature = "enterprise")]
+pub static ENTERPRISE_ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
+
+/// 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() -> &'static str {
+    #[cfg(not(feature = "enterprise"))]
+    let app_name = "Vector";
+    #[cfg(feature = "enterprise")]
+    let app_name = if *ENTERPRISE_ENABLED.get().unwrap_or(&false) {
+        "Vector Enterprise"
+    } else {
+        "Vector"
+    };
+
+    option_env!("VECTOR_APP_NAME").unwrap_or(app_name)
+}
+
+/// Returns a slugified version of the name used to identify this Vector application.
+///
+/// Defaults to "vector".
+pub fn get_slugified_app_name() -> String {
+    APP_NAME_SLUG
+        .get_or_init(|| get_app_name().to_lowercase().replace(' ', "-"))
+        .clone()
+}
+
 /// 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,7 @@ impl SinkConfig for DatadogLogsConfig {
             .dd_common
             .build_healthcheck(client.clone(), self.region.as_ref())?;
 
-        let sink = self.build_processor(client)?;
+        let sink = self.build_processor(client, cx.app_name_slug)?;
 
         Ok((sink, healthcheck))
     }

src/sinks/datadog/logs/service.rs

@@ -95,20 +95,31 @@ pub struct LogApiService {
     client: HttpClient,
     uri: Uri,
     user_provided_headers: IndexMap<HeaderName, HeaderValue>,
+    dd_evp_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)?;
+
+        let dd_evp_headers = &[
+            ("DD-EVP-ORIGIN".to_string(), dd_evp_origin),
+            ("DD-EVP-ORIGIN-VERSION".to_string(), crate::get_version()),
+        ]
+        .into_iter()
+        .collect();
+        let dd_evp_headers = validate_headers(dd_evp_headers)?;
 
         Ok(Self {
             client,
             uri,
-            user_provided_headers: headers,
+            user_provided_headers,
+            dd_evp_headers,
         })
     }
 }
@@ -128,8 +139,6 @@ impl Service<LogApiRequest> for LogApiService {
         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 = if let Some(ce) = request.compression.content_encoding() {
@@ -149,6 +158,10 @@ impl Service<LogApiRequest> for LogApiService {
                 // Replace rather than append to any existing header values
                 headers.insert(name, value.clone());
             }
+            // Set DD EVP headers last so that they cannot be overridden.
+            for (name, value) in &self.dd_evp_headers {
+                headers.insert(name, value.clone());
+            }
         }
 
         let http_request = http_request

src/sinks/datadog/logs/tests.rs

@@ -402,14 +402,13 @@ 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.app_name = "Vector Enterprise".to_string();
+    cx.app_name_slug = "vector-enterprise".to_string();
 
     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,8 @@ impl<'a> Builder<'a> {
                 globals: self.config.global.clone(),
                 proxy: ProxyConfig::merge_with_env(&self.config.global.proxy, sink.proxy()),
                 schema: self.config.schema,
+                app_name: crate::get_app_name().to_string(),
+                app_name_slug: crate::get_slugified_app_name(),
             };
 
             let (sink, healthcheck) = match sink.inner.build(cx).await {