From a853234a59bfe72799642cc824a7440dbf8b2f4c Mon Sep 17 00:00:00 2001
From: Kruno Tomola Fabro <krunotf@gmail.com>
Date: Wed, 26 Aug 2020 21:34:01 +0200
Subject: [PATCH 1/6] Rename

Signed-off-by: Kruno Tomola Fabro <krunotf@gmail.com>
---
 .meta/sinks/clickhouse.toml.erb            |  4 ++--
 .meta/sinks/datadog_logs.toml.erb          |  4 ++--
 .meta/sinks/datadog_metrics.toml.erb       |  2 +-
 .meta/sinks/elasticsearch.toml.erb         |  6 +++---
 .meta/sinks/humio_logs.toml.erb            |  4 ++--
 .meta/sinks/logdna.toml.erb                |  4 ++--
 .meta/sinks/papertrail.toml.erb            |  4 ++--
 .meta/sinks/pulsar.toml.erb                |  4 ++--
 .meta/sinks/sematext_logs.toml.erb         |  6 +++---
 .meta/sinks/splunk_hec.toml.erb            |  4 ++--
 .meta/sources/prometheus.toml.erb          |  4 ++--
 .meta/transforms/aws_ec2_metadata.toml.erb |  4 ++--
 src/sinks/clickhouse.rs                    |  8 +++++---
 src/sinks/datadog/logs.rs                  | 10 ++++++----
 src/sinks/datadog/metrics.rs               | 13 +++++++------
 src/sinks/elasticsearch.rs                 | 10 ++++++----
 src/sinks/humio_logs.rs                    |  8 +++++---
 src/sinks/logdna.rs                        | 10 ++++++----
 src/sinks/papertrail.rs                    | 12 +++++++-----
 src/sinks/pulsar.rs                        |  6 ++++--
 src/sinks/sematext_logs.rs                 |  9 +++++----
 src/sinks/splunk_hec.rs                    | 11 +++++++----
 src/sources/prometheus/mod.rs              |  8 +++++---
 src/transforms/aws_ec2_metadata.rs         |  6 ++++--
 24 files changed, 92 insertions(+), 69 deletions(-)

diff --git a/.meta/sinks/clickhouse.toml.erb b/.meta/sinks/clickhouse.toml.erb
index 1ccfa1d2932c0..e9d3b6f5b52e4 100644
--- a/.meta/sinks/clickhouse.toml.erb
+++ b/.meta/sinks/clickhouse.toml.erb
@@ -92,12 +92,12 @@ description = "The token to use for bearer authentication"
   }
 ) %>
 
-[sinks.clickhouse.options.host]
+[sinks.clickhouse.options.endpoint]
 type = "string"
 common = true
 examples = ["http://localhost:8123"]
 required = true
-description = "The host url of the [Clickhouse][urls.clickhouse] server."
+description = "The endpoint of the [Clickhouse][urls.clickhouse] server."
 
 [sinks.clickhouse.options.table]
 type = "string"
diff --git a/.meta/sinks/datadog_logs.toml.erb b/.meta/sinks/datadog_logs.toml.erb
index a15da3282ce41..e5c46d80962f6 100644
--- a/.meta/sinks/datadog_logs.toml.erb
+++ b/.meta/sinks/datadog_logs.toml.erb
@@ -37,12 +37,12 @@ examples = ["${DATADOG_API_KEY_ENV_VAR}", "ef8d5de700e7989468166c40fc8a0ccd"]
 required = true
 description = "Datadog [API key](https://docs.datadoghq.com/api/?lang=bash#authentication)"
 
-[sinks.datadog_logs.options.endpoint]
+[sinks.datadog_logs.options.address]
 type = "string"
 required = false
 default = "intake.logs.datadoghq.com:10516"
 examples = ["127.0.0.1:8080", "example.com:12345"]
-description = "The endpoint to stream logs to."
+description = "The address to stream logs to."
 
 <%= render(
   "_partials/fields/_tls_connector_options.toml",
diff --git a/.meta/sinks/datadog_metrics.toml.erb b/.meta/sinks/datadog_metrics.toml.erb
index b7e47b3c25e6c..84ddb17c84f98 100644
--- a/.meta/sinks/datadog_metrics.toml.erb
+++ b/.meta/sinks/datadog_metrics.toml.erb
@@ -41,7 +41,7 @@ examples = ["${DATADOG_API_KEY}", "ef8d5de700e7989468166c40fc8a0ccd"]
 required = true
 description = "Datadog [API key](https://docs.datadoghq.com/api/?lang=bash#authentication)"
 
-[sinks.datadog_metrics.options.host]
+[sinks.datadog_metrics.options.endpoint]
 type = "string"
 examples = ["https://api.datadoghq.com", "https://api.datadoghq.eu"]
 default = "https://api.datadoghq.com"
diff --git a/.meta/sinks/elasticsearch.toml.erb b/.meta/sinks/elasticsearch.toml.erb
index acc14cc563f0d..5ef0a8b3bb46b 100644
--- a/.meta/sinks/elasticsearch.toml.erb
+++ b/.meta/sinks/elasticsearch.toml.erb
@@ -124,7 +124,7 @@ relevant_when = {strategy = "aws"}
 required = false
 description = """\
 The [AWS region][urls.aws_regions] of the target service. \
-This defaults to the region named in the host parameter, \
+This defaults to the region named in the endpoint parameter, \
 or the value of the `$AWS_REGION` or `$AWS_DEFAULT_REGION` environment \
 variables if that cannot be determined, or "us-east-1".\
 """
@@ -160,12 +160,12 @@ description = """\
 A custom header to be added to each outgoing Elasticsearch request.\
 """
 
-[sinks.elasticsearch.options.host]
+[sinks.elasticsearch.options.endpoint]
 type = "string"
 common = true
 examples = ["http://10.24.32.122:9000"]
 description = """\
-The host of your Elasticsearch cluster. This should be the full URL as shown \
+The endpoint of your Elasticsearch cluster. This should be the full URL as shown \
 in the example.\
 """
 
diff --git a/.meta/sinks/humio_logs.toml.erb b/.meta/sinks/humio_logs.toml.erb
index 1d38a480d0cd6..c5e0350c94c08 100644
--- a/.meta/sinks/humio_logs.toml.erb
+++ b/.meta/sinks/humio_logs.toml.erb
@@ -78,11 +78,11 @@ to use to ingest the data.
 If unset, Humio will default it to none.
 """
 
-[sinks.humio_logs.options.host]
+[sinks.humio_logs.options.endpoint]
 type = "string"
 default = "https://cloud.humio.com"
 examples = ["http://myhumiohost.com"]
-description = "The optional host to send Humio logs to."
+description = "The optional endpoint to send Humio logs to."
 
 <%= render("_partials/fields/_compression_options.toml",
   namespace: "sinks.humio_logs.options",
diff --git a/.meta/sinks/logdna.toml.erb b/.meta/sinks/logdna.toml.erb
index a78a889f9458e..9db9597dde52a 100644
--- a/.meta/sinks/logdna.toml.erb
+++ b/.meta/sinks/logdna.toml.erb
@@ -51,11 +51,11 @@ required = true
 examples = ["${LOGDNA_API_KEY}", "ef8d5de700e7989468166c40fc8a0ccd"]
 description = "The Ingestion API key."
 
-[sinks.logdna.options.host]
+[sinks.logdna.options.endpoint]
 type = "string"
 required = false
 examples = ["http://127.0.0.1", "http://example.com"]
-description = "An optional host that will override the default one."
+description = "An optional endpoint that will override the default one."
 
 [sinks.logdna.options.hostname]
 type = "string"
diff --git a/.meta/sinks/papertrail.toml.erb b/.meta/sinks/papertrail.toml.erb
index de82004ccc0f6..46408dca22d99 100644
--- a/.meta/sinks/papertrail.toml.erb
+++ b/.meta/sinks/papertrail.toml.erb
@@ -31,9 +31,9 @@ write_to_description = "[Papertrail][urls.papertrail] via [Syslog][urls.papertra
   encodings: ["json", "text"],
 ) %>
 
-[sinks.papertrail.options.endpoint]
+[sinks.papertrail.options.address]
 type = "string"
 common = true
 required = true
 examples = ["logs.papertrailapp.com:12345"]
-description = "The endpoint to stream logs to."
+description = "The address to stream logs to."
diff --git a/.meta/sinks/pulsar.toml.erb b/.meta/sinks/pulsar.toml.erb
index 2cde6c77a0386..154dc9cf9ed49 100644
--- a/.meta/sinks/pulsar.toml.erb
+++ b/.meta/sinks/pulsar.toml.erb
@@ -30,12 +30,12 @@ write_to_description = "[Apache Pulsar][urls.pulsar] via the [Pulsar protocol][u
   default: "text"
 ) %>
 
-[sinks.pulsar.options.address]
+[sinks.pulsar.options.endpoint]
 type = "string"
 common = true
 examples = ["pulsar://127.0.0.1:6650"]
 required = true
-description = "A host and port pair that the pulsar client should connect to."
+description = "Endpoint to which the pulsar client should connect to."
 
 [sinks.pulsar.options.topic]
 type = "string"
diff --git a/.meta/sinks/sematext_logs.toml.erb b/.meta/sinks/sematext_logs.toml.erb
index 13e8563ae0df4..19819276b1f2c 100644
--- a/.meta/sinks/sematext_logs.toml.erb
+++ b/.meta/sinks/sematext_logs.toml.erb
@@ -48,13 +48,13 @@ write_to_description = "[Sematext][urls.sematext] via the [Elasticsearch API][ur
 type = "string"
 required = false
 examples = ["na", "eu"]
-description = "The region destination to send logs to. This option is required if `host` is not set."
+description = "The region destination to send logs to. This option is required if `endpoint` is not set."
 
-[sinks.sematext_logs.options.host]
+[sinks.sematext_logs.options.endpoint]
 type = "string"
 required = false
 examples = ["http://127.0.0.1", "http://example.com"]
-description = "The host that will be used to send logs to. This option is required if `region` is not set."
+description = "The endpoint that will be used to send logs to. This option is required if `region` is not set."
 
 [sinks.sematext_logs.options.token]
 type = "string"
diff --git a/.meta/sinks/splunk_hec.toml.erb b/.meta/sinks/splunk_hec.toml.erb
index ead84550dd0a1..7d8e6b9b48a53 100644
--- a/.meta/sinks/splunk_hec.toml.erb
+++ b/.meta/sinks/splunk_hec.toml.erb
@@ -48,12 +48,12 @@ write_to_description = "a [Splunk's HTTP Event Collector][urls.splunk_hec]"
   default: "text"
 ) %>
 
-[sinks.splunk_hec.options.host]
+[sinks.splunk_hec.options.endpoint]
 type = "string"
 common = true
 examples = ["http://my-splunk-host.com"]
 required = true
-description = "Your Splunk HEC host."
+description = "Your Splunk HEC endpoint."
 
 [sinks.splunk_hec.options.host_key]
 type = "string"
diff --git a/.meta/sources/prometheus.toml.erb b/.meta/sources/prometheus.toml.erb
index 59115dd71635d..f0aa238dfbba1 100644
--- a/.meta/sources/prometheus.toml.erb
+++ b/.meta/sources/prometheus.toml.erb
@@ -18,12 +18,12 @@ through_description = "the [Prometheus text exposition format][urls.prometheus_t
 
 <%= render("_partials/fields/_component_options.toml", type: "source", name: "prometheus") %>
 
-[sources.prometheus.options.hosts]
+[sources.prometheus.options.endpoints]
 type = "[string]"
 common = true
 required = true
 examples = [["http://localhost:9090"]]
-description = "Host addresses to scrape metrics from."
+description = "Endpoints to scrape metrics from."
 
 [sources.prometheus.options.scrape_interval_secs]
 type = "uint"
diff --git a/.meta/transforms/aws_ec2_metadata.toml.erb b/.meta/transforms/aws_ec2_metadata.toml.erb
index 250453da5ad5c..921ef5c227a54 100644
--- a/.meta/transforms/aws_ec2_metadata.toml.erb
+++ b/.meta/transforms/aws_ec2_metadata.toml.erb
@@ -40,11 +40,11 @@ common = true
 default = 10
 description = "The interval in seconds at which the EC2 Metadata api will be called."
 
-[transforms.aws_ec2_metadata.options.host]
+[transforms.aws_ec2_metadata.options.endpoint]
 type = "string"
 common = true
 default = "http://169.254.169.254"
-description = "Override the default EC2 Metadata host."
+description = "Override the default EC2 Metadata endpoint."
 
 [transforms.aws_ec2_metadata.fields.log.fields.ami-id]
 type = "string"
diff --git a/src/sinks/clickhouse.rs b/src/sinks/clickhouse.rs
index 7bf170a3eb568..6f464c9f100b0 100644
--- a/src/sinks/clickhouse.rs
+++ b/src/sinks/clickhouse.rs
@@ -21,7 +21,9 @@ use snafu::ResultExt;
 #[derive(Deserialize, Serialize, Debug, Clone, Default)]
 #[serde(deny_unknown_fields)]
 pub struct ClickhouseConfig {
-    pub host: String,
+    // Deprecated name
+    #[serde(alias = "host")]
+    pub endpoint: String,
     pub table: String,
     pub database: Option<String>,
     #[serde(default = "Compression::default_gzip")]
@@ -114,7 +116,7 @@ impl HttpSink for ClickhouseConfig {
             "default"
         };
 
-        let uri = encode_uri(&self.host, database, &self.table).expect("Unable to encode uri");
+        let uri = encode_uri(&self.endpoint, database, &self.table).expect("Unable to encode uri");
 
         let mut builder = Request::post(&uri).header("Content-Type", "application/x-ndjson");
 
@@ -134,7 +136,7 @@ impl HttpSink for ClickhouseConfig {
 
 async fn healthcheck(mut client: HttpClient, config: ClickhouseConfig) -> crate::Result<()> {
     // TODO: check if table exists?
-    let uri = format!("{}/?query=SELECT%201", config.host);
+    let uri = format!("{}/?query=SELECT%201", config.endpoint);
     let mut request = Request::get(uri).body(Body::empty()).unwrap();
 
     if let Some(auth) = &config.auth {
diff --git a/src/sinks/datadog/logs.rs b/src/sinks/datadog/logs.rs
index b4dd17d6dc147..8134842fdb8bc 100644
--- a/src/sinks/datadog/logs.rs
+++ b/src/sinks/datadog/logs.rs
@@ -16,7 +16,9 @@ use serde::{Deserialize, Serialize};
 #[derive(Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct DatadogLogsConfig {
-    endpoint: Option<UriSerde>,
+    // Deprecated name
+    #[serde(alias = "endpoint")]
+    address: Option<UriSerde>,
     api_key: String,
     encoding: EncodingConfig<Encoding>,
     tls: Option<TlsConfig>,
@@ -29,13 +31,13 @@ inventory::submit! {
 #[typetag::serde(name = "datadog_logs")]
 impl SinkConfig for DatadogLogsConfig {
     fn build(&self, cx: SinkContext) -> crate::Result<(super::RouterSink, super::Healthcheck)> {
-        let (host, port, tls) = if let Some(uri) = &self.endpoint {
+        let (host, port, tls) = if let Some(uri) = &self.address {
             let host = uri
                 .host()
-                .ok_or_else(|| "A host is required for endpoints".to_string())?;
+                .ok_or_else(|| "A host is required for address".to_string())?;
             let port = uri
                 .port_u16()
-                .ok_or_else(|| "A port is required for endpoints".to_string())?;
+                .ok_or_else(|| "A port is required for address".to_string())?;
 
             (host.to_string(), port, self.tls.clone())
         } else {
diff --git a/src/sinks/datadog/metrics.rs b/src/sinks/datadog/metrics.rs
index 73a2d62f5d6ab..381571abfc88f 100644
--- a/src/sinks/datadog/metrics.rs
+++ b/src/sinks/datadog/metrics.rs
@@ -35,8 +35,9 @@ struct DatadogState {
 #[serde(deny_unknown_fields)]
 pub struct DatadogConfig {
     pub namespace: String,
-    #[serde(default = "default_host")]
-    pub host: String,
+    // Deprecated name
+    #[serde(alias = "host", default = "default_endpoint")]
+    pub endpoint: String,
     pub api_key: String,
     #[serde(default)]
     pub batch: BatchConfig,
@@ -63,7 +64,7 @@ struct DatadogRequest {
     series: Vec<DatadogMetric>,
 }
 
-pub fn default_host() -> String {
+pub fn default_endpoint() -> String {
     String::from("https://api.datadoghq.com")
 }
 
@@ -114,7 +115,7 @@ impl SinkConfig for DatadogConfig {
             .parse_config(self.batch)?;
         let request = self.request.unwrap_with(&REQUEST_DEFAULTS);
 
-        let uri = build_uri(&self.host)?;
+        let uri = build_uri(&self.endpoint);
         let timestamp = Utc::now().timestamp();
 
         let sink = DatadogSink {
@@ -179,7 +180,7 @@ fn build_uri(host: &str) -> crate::Result<Uri> {
 }
 
 async fn healthcheck(config: DatadogConfig, mut client: HttpClient) -> crate::Result<()> {
-    let uri = format!("{}/api/v1/validate", config.host)
+    let uri = format!("{}/api/v1/validate", config.endpoint)
         .parse::<Uri>()
         .context(super::UriParseError)?;
 
@@ -417,7 +418,7 @@ mod tests {
         let timestamp = Utc::now().timestamp();
         let sink = DatadogSink {
             config: sink,
-            uri: build_uri(&default_host()).unwrap(),
+            uri: build_uri(&default_endpoint()),
             last_sent_timestamp: AtomicI64::new(timestamp),
         };
 
diff --git a/src/sinks/elasticsearch.rs b/src/sinks/elasticsearch.rs
index 72912b456d7b1..ea9014020f8fe 100644
--- a/src/sinks/elasticsearch.rs
+++ b/src/sinks/elasticsearch.rs
@@ -35,7 +35,9 @@ use std::convert::TryFrom;
 #[derive(Deserialize, Serialize, Debug, Clone, Default)]
 #[serde(deny_unknown_fields)]
 pub struct ElasticSearchConfig {
-    pub host: String,
+    // Deprecated name
+    #[serde(alias = "host")]
+    pub endpoint: String,
     pub index: Option<String>,
     pub doc_type: Option<String>,
     pub id_key: Option<String>,
@@ -358,14 +360,14 @@ impl ElasticSearchCommon {
             _ => None,
         };
 
-        let base_url = config.host.clone();
+        let base_url = config.endpoint.clone();
         let region = match &config.aws {
             Some(region) => Region::try_from(region)?,
-            None => region_from_endpoint(&config.host)?,
+            None => region_from_endpoint(&config.endpoint)?,
         };
 
         // Test the configured host, but ignore the result
-        let uri = format!("{}/_test", &config.host);
+        let uri = format!("{}/_test", &config.endpoint);
         let uri = uri
             .parse::<Uri>()
             .with_context(|| InvalidHost { host: &base_url })?;
diff --git a/src/sinks/humio_logs.rs b/src/sinks/humio_logs.rs
index 823447f4b0b50..00559184d41a8 100644
--- a/src/sinks/humio_logs.rs
+++ b/src/sinks/humio_logs.rs
@@ -13,7 +13,9 @@ const HOST: &str = "https://cloud.humio.com";
 #[derive(Clone, Debug, Deserialize, Serialize, Default)]
 pub struct HumioLogsConfig {
     token: String,
-    host: Option<String>,
+    // Deprecated name
+    #[serde(alias = "host")]
+    endpoint: Option<String>,
     source: Option<Template>,
     #[serde(
         skip_serializing_if = "crate::serde::skip_serializing_if_default",
@@ -72,11 +74,11 @@ impl SinkConfig for HumioLogsConfig {
 
 impl HumioLogsConfig {
     fn build_hec_config(&self) -> HecSinkConfig {
-        let host = self.host.clone().unwrap_or_else(|| HOST.to_string());
+        let endpoint = self.endpoint.clone().unwrap_or_else(|| HOST.to_string());
 
         HecSinkConfig {
             token: self.token.clone(),
-            host,
+            endpoint,
             source: self.source.clone(),
             sourcetype: self.event_type.clone(),
             encoding: self.encoding.clone().transmute(),
diff --git a/src/sinks/logdna.rs b/src/sinks/logdna.rs
index bb612c102db4e..96598c0bc48a7 100644
--- a/src/sinks/logdna.rs
+++ b/src/sinks/logdna.rs
@@ -23,7 +23,9 @@ const PATH: &str = "/logs/ingest";
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct LogdnaConfig {
     api_key: String,
-    host: Option<UriSerde>,
+    // Deprecated name
+    #[serde(alias = "host")]
+    endpoint: Option<UriSerde>,
 
     hostname: String,
     mac: Option<String>,
@@ -187,7 +189,7 @@ impl HttpSink for LogdnaConfig {
 
 impl LogdnaConfig {
     fn build_uri(&self, query: &str) -> Uri {
-        let host: Uri = self.host.clone().unwrap_or_else(|| HOST.clone()).into();
+        let host: Uri = self.endpoint.clone().unwrap_or_else(|| HOST.clone()).into();
 
         let uri = format!("{}{}?{}", host, PATH, query);
 
@@ -275,8 +277,8 @@ mod tests {
         let addr = test_util::next_addr();
         // Swap out the host so we can force send it
         // to our local server
-        let host = format!("http://{}", addr).parse::<http::Uri>().unwrap();
-        config.host = Some(host.into());
+        let endpoint = format!("http://{}", addr).parse::<http::Uri>().unwrap();
+        config.endpoint = Some(endpoint.into());
 
         let (sink, _) = config.build(cx).unwrap();
 
diff --git a/src/sinks/papertrail.rs b/src/sinks/papertrail.rs
index 56f0e667f2f35..2b273d57a11e9 100644
--- a/src/sinks/papertrail.rs
+++ b/src/sinks/papertrail.rs
@@ -16,7 +16,9 @@ use syslog::{Facility, Formatter3164, LogFormat, Severity};
 #[derive(Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct PapertrailConfig {
-    endpoint: UriSerde,
+    // Deprecated name
+    #[serde(alias = "endpoint")]
+    address: UriSerde,
     encoding: EncodingConfig<Encoding>,
 }
 
@@ -28,14 +30,14 @@ inventory::submit! {
 impl SinkConfig for PapertrailConfig {
     fn build(&self, cx: SinkContext) -> crate::Result<(super::RouterSink, super::Healthcheck)> {
         let host = self
-            .endpoint
+            .address
             .host()
             .map(str::to_string)
-            .ok_or_else(|| "A host is required for endpoints".to_string())?;
+            .ok_or_else(|| "A host is required for address".to_string())?;
         let port = self
-            .endpoint
+            .address
             .port_u16()
-            .ok_or_else(|| "A port is required for endpoints".to_string())?;
+            .ok_or_else(|| "A port is required for address".to_string())?;
 
         let sink = TcpSink::new(
             host,
diff --git a/src/sinks/pulsar.rs b/src/sinks/pulsar.rs
index 2f2e50033efe3..68e31a58f0888 100644
--- a/src/sinks/pulsar.rs
+++ b/src/sinks/pulsar.rs
@@ -26,7 +26,9 @@ enum BuildError {
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct PulsarSinkConfig {
-    address: String,
+    // Deprecated name
+    #[serde(alias = "address")]
+    endpoint: String,
     topic: String,
     encoding: EncodingConfigWithDefault<Encoding>,
     auth: Option<AuthConfig>,
@@ -100,7 +102,7 @@ impl SinkConfig for PulsarSinkConfig {
 
 impl PulsarSinkConfig {
     async fn create_pulsar_producer(&self) -> Result<Producer<TokioExecutor>, PulsarError> {
-        let mut builder = Pulsar::builder(&self.address, TokioExecutor);
+        let mut builder = Pulsar::builder(&self.endpoint, TokioExecutor);
         if let Some(auth) = &self.auth {
             builder = builder.with_auth(Authentication {
                 name: auth.name.clone(),
diff --git a/src/sinks/sematext_logs.rs b/src/sinks/sematext_logs.rs
index 86bb54c66ccb2..9b7efc6455f84 100644
--- a/src/sinks/sematext_logs.rs
+++ b/src/sinks/sematext_logs.rs
@@ -12,8 +12,9 @@ use serde::{Deserialize, Serialize};
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct SematextLogsConfig {
     region: Option<Region>,
-    // TODO: replace this with `UriEncode` once that is on master.
-    host: Option<String>,
+    // Deprecated name
+    #[serde(alias = "host")]
+    endpoint: Option<String>,
     token: String,
 
     #[serde(
@@ -43,7 +44,7 @@ pub enum Region {
 #[typetag::serde(name = "sematext")]
 impl SinkConfig for SematextLogsConfig {
     fn build(&self, cx: SinkContext) -> crate::Result<(super::RouterSink, super::Healthcheck)> {
-        let host = match (&self.host, &self.region) {
+        let host = match (&self.endpoint, &self.region) {
             (Some(host), None) => host.clone(),
             (None, Some(Region::Na)) => "https://logsene-receiver.sematext.com".to_owned(),
             (None, Some(Region::Eu)) => "https://logsene-receiver.eu.sematext.com".to_owned(),
@@ -122,7 +123,7 @@ mod tests {
         let addr = test_util::next_addr();
         // Swap out the host so we can force send it
         // to our local server
-        config.host = Some(format!("http://{}", addr));
+        config.endpoint = Some(format!("http://{}", addr));
         config.region = None;
 
         let (sink, _) = config.build(cx).unwrap();
diff --git a/src/sinks/splunk_hec.rs b/src/sinks/splunk_hec.rs
index a006e4ea384be..af43ea72e2e8f 100644
--- a/src/sinks/splunk_hec.rs
+++ b/src/sinks/splunk_hec.rs
@@ -34,7 +34,9 @@ pub enum BuildError {
 #[serde(deny_unknown_fields)]
 pub struct HecSinkConfig {
     pub token: String,
-    pub host: String,
+    // Deprecated name
+    #[serde(alias = "host")]
+    pub endpoint: String,
     #[serde(default = "default_host_key")]
     pub host_key: Atom,
     #[serde(default)]
@@ -206,7 +208,8 @@ impl HttpSink for HecSinkConfig {
     }
 
     async fn build_request(&self, events: Self::Output) -> crate::Result<Request<Vec<u8>>> {
-        let uri = build_uri(&self.host, "/services/collector/event").expect("Unable to parse URI");
+        let uri =
+            build_uri(&self.endpoint, "/services/collector/event").expect("Unable to parse URI");
 
         let mut builder = Request::post(uri)
             .header("Content-Type", "application/json")
@@ -229,8 +232,8 @@ enum HealthcheckError {
 }
 
 pub async fn healthcheck(config: HecSinkConfig, mut client: HttpClient) -> crate::Result<()> {
-    let uri =
-        build_uri(&config.host, "/services/collector/health/1.0").context(super::UriParseError)?;
+    let uri = build_uri(&config.endpoint, "/services/collector/health/1.0")
+        .context(super::UriParseError)?;
 
     let request = Request::get(uri)
         .header("Authorization", format!("Splunk {}", config.token))
diff --git a/src/sources/prometheus/mod.rs b/src/sources/prometheus/mod.rs
index b77f8b6d0a46b..c07ad6688984d 100644
--- a/src/sources/prometheus/mod.rs
+++ b/src/sources/prometheus/mod.rs
@@ -22,7 +22,9 @@ pub mod parser;
 
 #[derive(Deserialize, Serialize, Clone, Debug)]
 struct PrometheusConfig {
-    hosts: Vec<String>,
+    // Deprecated name
+    #[serde(alias = "hosts")]
+    endpoints: Vec<String>,
     #[serde(default = "default_scrape_interval_secs")]
     scrape_interval_secs: u64,
 }
@@ -41,7 +43,7 @@ impl crate::config::SourceConfig for PrometheusConfig {
         out: Pipeline,
     ) -> crate::Result<super::Source> {
         let mut urls = Vec::new();
-        for host in self.hosts.iter() {
+        for host in self.endpoints.iter() {
             let base_uri = host.parse::<http::Uri>().context(super::UriParseError)?;
             urls.push(format!("{}metrics", base_uri));
         }
@@ -207,7 +209,7 @@ mod test {
         config.add_source(
             "in",
             PrometheusConfig {
-                hosts: vec![format!("http://{}", in_addr)],
+                endpoints: vec![format!("http://{}", in_addr)],
                 scrape_interval_secs: 1,
             },
         );
diff --git a/src/transforms/aws_ec2_metadata.rs b/src/transforms/aws_ec2_metadata.rs
index f1b47f03fac7d..51dd52a4fabe7 100644
--- a/src/transforms/aws_ec2_metadata.rs
+++ b/src/transforms/aws_ec2_metadata.rs
@@ -79,7 +79,9 @@ lazy_static::lazy_static! {
 
 #[derive(Default, Debug, Serialize, Deserialize)]
 pub struct Ec2Metadata {
-    host: Option<String>,
+    // Deprecated name
+    #[serde(alias = "host")]
+    endpoint: Option<String>,
     namespace: Option<String>,
     refresh_interval_secs: Option<u64>,
     fields: Option<Vec<String>>,
@@ -132,7 +134,7 @@ impl TransformConfig for Ec2Metadata {
         let keys = Keys::new(&namespace);
 
         let host = self
-            .host
+            .endpoint
             .clone()
             .map(|s| Uri::from_maybe_shared(s).unwrap())
             .unwrap_or_else(|| HOST.clone());

From b08fa7896a10bf2ecb5d48c59031fb828bd8d93e Mon Sep 17 00:00:00 2001
From: Kruno Tomola Fabro <krunotf@gmail.com>
Date: Wed, 26 Aug 2020 22:08:47 +0200
Subject: [PATCH 2/6] Update usages of renamed

Signed-off-by: Kruno Tomola Fabro <krunotf@gmail.com>
---
 src/sinks/datadog/metrics.rs | 4 ++--
 src/sinks/elasticsearch.rs   | 2 +-
 src/sinks/sematext_logs.rs   | 4 ++--
 src/sinks/splunk_hec.rs      | 2 +-
 src/sources/splunk_hec.rs    | 2 +-
 5 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/sinks/datadog/metrics.rs b/src/sinks/datadog/metrics.rs
index 381571abfc88f..1432396cccbdf 100644
--- a/src/sinks/datadog/metrics.rs
+++ b/src/sinks/datadog/metrics.rs
@@ -115,7 +115,7 @@ impl SinkConfig for DatadogConfig {
             .parse_config(self.batch)?;
         let request = self.request.unwrap_with(&REQUEST_DEFAULTS);
 
-        let uri = build_uri(&self.endpoint);
+        let uri = build_uri(&self.endpoint)?;
         let timestamp = Utc::now().timestamp();
 
         let sink = DatadogSink {
@@ -418,7 +418,7 @@ mod tests {
         let timestamp = Utc::now().timestamp();
         let sink = DatadogSink {
             config: sink,
-            uri: build_uri(&default_endpoint()),
+            uri: build_uri(&default_endpoint()).unwrap(),
             last_sent_timestamp: AtomicI64::new(timestamp),
         };
 
diff --git a/src/sinks/elasticsearch.rs b/src/sinks/elasticsearch.rs
index ea9014020f8fe..7b78f0e1f74cf 100644
--- a/src/sinks/elasticsearch.rs
+++ b/src/sinks/elasticsearch.rs
@@ -373,7 +373,7 @@ impl ElasticSearchCommon {
             .with_context(|| InvalidHost { host: &base_url })?;
         if uri.host().is_none() {
             return Err(ParseError::HostMustIncludeHostname {
-                host: config.host.clone(),
+                host: config.endpoint.clone(),
             }
             .into());
         }
diff --git a/src/sinks/sematext_logs.rs b/src/sinks/sematext_logs.rs
index 9b7efc6455f84..4aa55a1d298ac 100644
--- a/src/sinks/sematext_logs.rs
+++ b/src/sinks/sematext_logs.rs
@@ -44,7 +44,7 @@ pub enum Region {
 #[typetag::serde(name = "sematext")]
 impl SinkConfig for SematextLogsConfig {
     fn build(&self, cx: SinkContext) -> crate::Result<(super::RouterSink, super::Healthcheck)> {
-        let host = match (&self.endpoint, &self.region) {
+        let endpoint = match (&self.endpoint, &self.region) {
             (Some(host), None) => host.clone(),
             (None, Some(Region::Na)) => "https://logsene-receiver.sematext.com".to_owned(),
             (None, Some(Region::Eu)) => "https://logsene-receiver.eu.sematext.com".to_owned(),
@@ -57,7 +57,7 @@ impl SinkConfig for SematextLogsConfig {
         };
 
         let (sink, healthcheck) = ElasticSearchConfig {
-            host,
+            endpoint,
             compression: Compression::None,
             doc_type: Some("logs".to_string()),
             index: Some(self.token.clone()),
diff --git a/src/sinks/splunk_hec.rs b/src/sinks/splunk_hec.rs
index af43ea72e2e8f..2422d295a069b 100644
--- a/src/sinks/splunk_hec.rs
+++ b/src/sinks/splunk_hec.rs
@@ -86,7 +86,7 @@ inventory::submit! {
 #[typetag::serde(name = "splunk_hec")]
 impl SinkConfig for HecSinkConfig {
     fn build(&self, cx: SinkContext) -> crate::Result<(super::RouterSink, super::Healthcheck)> {
-        validate_host(&self.host)?;
+        validate_host(&self.endpoint)?;
 
         let batch = BatchSettings::default()
             .bytes(bytesize::mib(1u64))
diff --git a/src/sources/splunk_hec.rs b/src/sources/splunk_hec.rs
index ebbce50875d1d..670fef43d98cb 100644
--- a/src/sources/splunk_hec.rs
+++ b/src/sources/splunk_hec.rs
@@ -803,7 +803,7 @@ mod tests {
         compression: Compression,
     ) -> (RouterSink, Healthcheck) {
         HecSinkConfig {
-            host: format!("http://{}", address),
+            endpoint: format!("http://{}", address),
             token: TOKEN.to_owned(),
             encoding: encoding.into(),
             compression,

From 6fb39cd7a00b6b65bd6ba519289d8c921fa6ef15 Mon Sep 17 00:00:00 2001
From: Kruno Tomola Fabro <krunotf@gmail.com>
Date: Thu, 27 Aug 2020 22:56:07 +0200
Subject: [PATCH 3/6] Revert endpoint

Signed-off-by: Kruno Tomola Fabro <krunotf@gmail.com>
---
 src/sinks/datadog/logs.rs | 10 ++++------
 src/sinks/papertrail.rs   | 12 +++++-------
 2 files changed, 9 insertions(+), 13 deletions(-)

diff --git a/src/sinks/datadog/logs.rs b/src/sinks/datadog/logs.rs
index 8134842fdb8bc..76e25225dea75 100644
--- a/src/sinks/datadog/logs.rs
+++ b/src/sinks/datadog/logs.rs
@@ -16,9 +16,7 @@ use serde::{Deserialize, Serialize};
 #[derive(Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct DatadogLogsConfig {
-    // Deprecated name
-    #[serde(alias = "endpoint")]
-    address: Option<UriSerde>,
+    endpoint: Option<UriSerde>,
     api_key: String,
     encoding: EncodingConfig<Encoding>,
     tls: Option<TlsConfig>,
@@ -31,13 +29,13 @@ inventory::submit! {
 #[typetag::serde(name = "datadog_logs")]
 impl SinkConfig for DatadogLogsConfig {
     fn build(&self, cx: SinkContext) -> crate::Result<(super::RouterSink, super::Healthcheck)> {
-        let (host, port, tls) = if let Some(uri) = &self.address {
+        let (host, port, tls) = if let Some(uri) = &self.endpoint {
             let host = uri
                 .host()
-                .ok_or_else(|| "A host is required for address".to_string())?;
+                .ok_or_else(|| "A host is required for endpoint".to_string())?;
             let port = uri
                 .port_u16()
-                .ok_or_else(|| "A port is required for address".to_string())?;
+                .ok_or_else(|| "A port is required for endpoint".to_string())?;
 
             (host.to_string(), port, self.tls.clone())
         } else {
diff --git a/src/sinks/papertrail.rs b/src/sinks/papertrail.rs
index 2b273d57a11e9..d489ef01f1f61 100644
--- a/src/sinks/papertrail.rs
+++ b/src/sinks/papertrail.rs
@@ -16,9 +16,7 @@ use syslog::{Facility, Formatter3164, LogFormat, Severity};
 #[derive(Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct PapertrailConfig {
-    // Deprecated name
-    #[serde(alias = "endpoint")]
-    address: UriSerde,
+    endpoint: UriSerde,
     encoding: EncodingConfig<Encoding>,
 }
 
@@ -30,14 +28,14 @@ inventory::submit! {
 impl SinkConfig for PapertrailConfig {
     fn build(&self, cx: SinkContext) -> crate::Result<(super::RouterSink, super::Healthcheck)> {
         let host = self
-            .address
+            .endpoint
             .host()
             .map(str::to_string)
-            .ok_or_else(|| "A host is required for address".to_string())?;
+            .ok_or_else(|| "A host is required for endpoint".to_string())?;
         let port = self
-            .address
+            .endpoint
             .port_u16()
-            .ok_or_else(|| "A port is required for address".to_string())?;
+            .ok_or_else(|| "A port is required for endpoint".to_string())?;
 
         let sink = TcpSink::new(
             host,

From 375ce0302cd4010dcd66e85337c1d3b88363ac31 Mon Sep 17 00:00:00 2001
From: Kruno Tomola Fabro <krunotf@gmail.com>
Date: Thu, 27 Aug 2020 23:56:47 +0200
Subject: [PATCH 4/6] Update tests

Signed-off-by: Kruno Tomola Fabro <krunotf@gmail.com>
---
 src/sinks/aws_kinesis_firehose.rs  |  2 +-
 src/sinks/clickhouse.rs            |  6 +++---
 src/sinks/elasticsearch.rs         | 12 ++++++------
 src/sinks/pulsar.rs                |  4 ++--
 src/sinks/splunk_hec.rs            |  6 +++---
 src/transforms/aws_ec2_metadata.rs |  8 ++++----
 6 files changed, 19 insertions(+), 19 deletions(-)

diff --git a/src/sinks/aws_kinesis_firehose.rs b/src/sinks/aws_kinesis_firehose.rs
index 47691fdc74b12..3f315d28a360e 100644
--- a/src/sinks/aws_kinesis_firehose.rs
+++ b/src/sinks/aws_kinesis_firehose.rs
@@ -335,7 +335,7 @@ mod integration_tests {
 
             let config = ElasticSearchConfig {
                 auth: Some(ElasticSearchAuth::Aws { assume_role: None }),
-                host: "http://localhost:4571".into(),
+                endpoint: "http://localhost:4571".into(),
                 index: Some(stream.clone()),
                 ..Default::default()
             };
diff --git a/src/sinks/clickhouse.rs b/src/sinks/clickhouse.rs
index 6f464c9f100b0..89c98ca27b267 100644
--- a/src/sinks/clickhouse.rs
+++ b/src/sinks/clickhouse.rs
@@ -256,7 +256,7 @@ mod integration_tests {
             let host = String::from("http://localhost:8123");
 
             let config = ClickhouseConfig {
-                host: host.clone(),
+                endpoint: host.clone(),
                 table: table.clone(),
                 compression: Compression::None,
                 batch: BatchConfig {
@@ -303,7 +303,7 @@ mod integration_tests {
             encoding.timestamp_format = Some(TimestampFormat::Unix);
 
             let config = ClickhouseConfig {
-                host: host.clone(),
+                endpoint: host.clone(),
                 table: table.clone(),
                 compression: Compression::None,
                 encoding,
@@ -428,7 +428,7 @@ timestamp_format = "unix""#,
             let host = String::from("http://localhost:8123");
 
             let config = ClickhouseConfig {
-                host: host.clone(),
+                endpoint: host.clone(),
                 table: table.clone(),
                 compression: Compression::None,
                 batch: BatchConfig {
diff --git a/src/sinks/elasticsearch.rs b/src/sinks/elasticsearch.rs
index 7b78f0e1f74cf..377eb7bd6d445 100644
--- a/src/sinks/elasticsearch.rs
+++ b/src/sinks/elasticsearch.rs
@@ -587,7 +587,7 @@ mod integration_tests {
         let pipeline = String::from("test-pipeline");
 
         let config = ElasticSearchConfig {
-            host: "http://localhost:9200".into(),
+            endpoint: "http://localhost:9200".into(),
             index: Some(index.clone()),
             pipeline: Some(pipeline.clone()),
             ..config()
@@ -603,7 +603,7 @@ mod integration_tests {
 
         let index = gen_index();
         let config = ElasticSearchConfig {
-            host: "http://localhost:9200".into(),
+            endpoint: "http://localhost:9200".into(),
             index: Some(index.clone()),
             doc_type: Some("log_lines".into()),
             id_key: Some("my_id".into()),
@@ -660,7 +660,7 @@ mod integration_tests {
     fn insert_events_over_http() {
         run_insert_tests(
             ElasticSearchConfig {
-                host: "http://localhost:9200".into(),
+                endpoint: "http://localhost:9200".into(),
                 doc_type: Some("log_lines".into()),
                 compression: Compression::None,
                 ..config()
@@ -673,7 +673,7 @@ mod integration_tests {
     fn insert_events_over_https() {
         run_insert_tests(
             ElasticSearchConfig {
-                host: "https://localhost:9201".into(),
+                endpoint: "https://localhost:9201".into(),
                 doc_type: Some("log_lines".into()),
                 compression: Compression::None,
                 tls: Some(TlsOptions {
@@ -691,7 +691,7 @@ mod integration_tests {
         run_insert_tests(
             ElasticSearchConfig {
                 auth: Some(ElasticSearchAuth::Aws { assume_role: None }),
-                host: "http://localhost:4571".into(),
+                endpoint: "http://localhost:4571".into(),
                 ..config()
             },
             false,
@@ -702,7 +702,7 @@ mod integration_tests {
     fn insert_events_with_failure() {
         run_insert_tests(
             ElasticSearchConfig {
-                host: "http://localhost:9200".into(),
+                endpoint: "http://localhost:9200".into(),
                 doc_type: Some("log_lines".into()),
                 compression: Compression::None,
                 ..config()
diff --git a/src/sinks/pulsar.rs b/src/sinks/pulsar.rs
index 68e31a58f0888..77c8b2d64b1a2 100644
--- a/src/sinks/pulsar.rs
+++ b/src/sinks/pulsar.rs
@@ -243,13 +243,13 @@ mod integration_tests {
 
             let topic = format!("test-{}", random_string(10));
             let cnf = PulsarSinkConfig {
-                address: "pulsar://127.0.0.1:6650".to_owned(),
+                endpoint: "pulsar://127.0.0.1:6650".to_owned(),
                 topic: topic.clone(),
                 encoding: Encoding::Text.into(),
                 auth: None,
             };
 
-            let pulsar = Pulsar::<TokioExecutor>::builder(&cnf.address, TokioExecutor)
+            let pulsar = Pulsar::<TokioExecutor>::builder(&cnf.endpoint, TokioExecutor)
                 .build()
                 .await
                 .unwrap();
diff --git a/src/sinks/splunk_hec.rs b/src/sinks/splunk_hec.rs
index 2422d295a069b..e7937c676b5a5 100644
--- a/src/sinks/splunk_hec.rs
+++ b/src/sinks/splunk_hec.rs
@@ -647,7 +647,7 @@ mod integration_tests {
             // Server not listening at address
             {
                 let config = HecSinkConfig {
-                    host: "http://localhost:1111".to_string(),
+                    endpoint: "http://localhost:1111".to_string(),
                     ..config(Encoding::Text, vec![]).await
                 };
                 let healthcheck = config_to_healthcheck(config);
@@ -670,7 +670,7 @@ mod integration_tests {
             // Unhealthy server
             {
                 let config = HecSinkConfig {
-                    host: "http://localhost:5503".to_string(),
+                    endpoint: "http://localhost:5503".to_string(),
                     ..config(Encoding::Text, vec![]).await
                 };
 
@@ -724,7 +724,7 @@ mod integration_tests {
         indexed_fields: Vec<Atom>,
     ) -> super::HecSinkConfig {
         super::HecSinkConfig {
-            host: "http://localhost:8088/".into(),
+            endpoint: "http://localhost:8088/".into(),
             token: get_token().await,
             host_key: "host".into(),
             compression: Compression::None,
diff --git a/src/transforms/aws_ec2_metadata.rs b/src/transforms/aws_ec2_metadata.rs
index 51dd52a4fabe7..d53cbb09d4f05 100644
--- a/src/transforms/aws_ec2_metadata.rs
+++ b/src/transforms/aws_ec2_metadata.rs
@@ -511,7 +511,7 @@ mod integration_tests {
         let mut rt = runtime();
 
         let config = Ec2Metadata {
-            host: Some(HOST.clone()),
+            endpoint: Some(HOST.clone()),
             ..Default::default()
         };
         let mut transform = rt.block_on_std(async move {
@@ -563,7 +563,7 @@ mod integration_tests {
         let mut rt = runtime();
 
         let config = Ec2Metadata {
-            host: Some(HOST.clone()),
+            endpoint: Some(HOST.clone()),
             fields: Some(vec!["public-ipv4".into(), "region".into()]),
             ..Default::default()
         };
@@ -598,7 +598,7 @@ mod integration_tests {
         let mut rt = runtime();
 
         let config = Ec2Metadata {
-            host: Some(HOST.clone()),
+            endpoint: Some(HOST.clone()),
             namespace: Some("ec2.metadata".into()),
             ..Default::default()
         };
@@ -628,7 +628,7 @@ mod integration_tests {
 
         // Set an empty namespace to ensure we don't prepend one.
         let config = Ec2Metadata {
-            host: Some(HOST.clone()),
+            endpoint: Some(HOST.clone()),
             namespace: Some("".into()),
             ..Default::default()
         };

From 50e05fb6bac7357ac1a537898f43d0369f097384 Mon Sep 17 00:00:00 2001
From: ktf <krunotf@gmail.com>
Date: Fri, 28 Aug 2020 10:33:37 +0200
Subject: [PATCH 5/6] Revert docs

Signed-off-by: ktf <krunotf@gmail.com>
---
 .meta/sinks/datadog_logs.toml.erb | 4 ++--
 .meta/sinks/papertrail.toml.erb   | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/.meta/sinks/datadog_logs.toml.erb b/.meta/sinks/datadog_logs.toml.erb
index e5c46d80962f6..a15da3282ce41 100644
--- a/.meta/sinks/datadog_logs.toml.erb
+++ b/.meta/sinks/datadog_logs.toml.erb
@@ -37,12 +37,12 @@ examples = ["${DATADOG_API_KEY_ENV_VAR}", "ef8d5de700e7989468166c40fc8a0ccd"]
 required = true
 description = "Datadog [API key](https://docs.datadoghq.com/api/?lang=bash#authentication)"
 
-[sinks.datadog_logs.options.address]
+[sinks.datadog_logs.options.endpoint]
 type = "string"
 required = false
 default = "intake.logs.datadoghq.com:10516"
 examples = ["127.0.0.1:8080", "example.com:12345"]
-description = "The address to stream logs to."
+description = "The endpoint to stream logs to."
 
 <%= render(
   "_partials/fields/_tls_connector_options.toml",
diff --git a/.meta/sinks/papertrail.toml.erb b/.meta/sinks/papertrail.toml.erb
index 46408dca22d99..de82004ccc0f6 100644
--- a/.meta/sinks/papertrail.toml.erb
+++ b/.meta/sinks/papertrail.toml.erb
@@ -31,9 +31,9 @@ write_to_description = "[Papertrail][urls.papertrail] via [Syslog][urls.papertra
   encodings: ["json", "text"],
 ) %>
 
-[sinks.papertrail.options.address]
+[sinks.papertrail.options.endpoint]
 type = "string"
 common = true
 required = true
 examples = ["logs.papertrailapp.com:12345"]
-description = "The address to stream logs to."
+description = "The endpoint to stream logs to."

From 7baddf47241f34be09c496e9fb031c2e088cd19d Mon Sep 17 00:00:00 2001
From: ktf <krunotf@gmail.com>
Date: Fri, 28 Aug 2020 13:17:08 +0200
Subject: [PATCH 6/6] More test updates

Signed-off-by: ktf <krunotf@gmail.com>
---
 src/sinks/pulsar.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/sinks/pulsar.rs b/src/sinks/pulsar.rs
index 76c066321bdca..5ce6fb23eb0e7 100644
--- a/src/sinks/pulsar.rs
+++ b/src/sinks/pulsar.rs
@@ -247,7 +247,7 @@ mod integration_tests {
             auth: None,
         };
 
-        let pulsar = Pulsar::<TokioExecutor>::builder(&cnf.address, TokioExecutor)
+        let pulsar = Pulsar::<TokioExecutor>::builder(&cnf.endpoint, TokioExecutor)
             .build()
             .await
             .unwrap();