Merge branch 'master' of github.com:vectordotdev/vector into opentele…

…metry-add-log
vectordotdev · Jul 14, 2022 · ac77281 · ac77281
2 parents ee359d4 + 9a9340a
commit ac77281
Show file tree

Hide file tree

Showing 143 changed files with 2,220 additions and 4,309 deletions.
diff --git a/benches/codecs/encoder.rs b/benches/codecs/encoder.rs
@@ -48,17 +48,6 @@ fn encoder(c: &mut Criterion) {
         "key3" => "value3"
     }));
 
-    group.throughput(Throughput::Bytes(input.size_of() as u64));
-    group.bench_with_input("vector::sinks::util::encode_log", &(), |b, ()| {
-        b.iter_batched(
-            || vector::sinks::util::Encoding::Json.into(),
-            |encoding| {
-                vector::sinks::util::encode_log(input.clone(), &encoding).unwrap();
-            },
-            BatchSize::SmallInput,
-        )
-    });
-
     group.throughput(Throughput::Bytes(input.size_of() as u64));
     group.bench_with_input("JsonLogVecSerializer::encode", &(), |b, ()| {
         b.iter_batched(

diff --git a/benches/files.rs b/benches/files.rs
@@ -1,6 +1,7 @@
 use std::{convert::TryInto, path::PathBuf};
 
 use bytes::Bytes;
+use codecs::{encoding::FramingConfig, TextSerializerConfig};
 use criterion::{criterion_group, BatchSize, Criterion, SamplingMode, Throughput};
 use futures::{stream, SinkExt, StreamExt};
 use tempfile::tempdir;
@@ -54,10 +55,7 @@ fn benchmark_files_no_partitions(c: &mut Criterion) {
                     sinks::file::FileSinkConfig {
                         path: output.try_into().unwrap(),
                         idle_timeout_secs: None,
-                        encoding: sinks::util::encoding::EncodingConfig::from(
-                            sinks::file::Encoding::Text,
-                        )
-                        .into(),
+                        encoding: (None::<FramingConfig>, TextSerializerConfig::new()).into(),
                         compression: sinks::file::Compression::None,
                         acknowledgements: Default::default(),
                     },

diff --git a/benches/http.rs b/benches/http.rs
@@ -1,5 +1,6 @@
 use std::net::SocketAddr;
 
+use codecs::{encoding::FramingConfig, TextSerializerConfig};
 use criterion::{criterion_group, BatchSize, BenchmarkId, Criterion, SamplingMode, Throughput};
 use futures::TryFutureExt;
 use hyper::{
@@ -9,7 +10,7 @@ use hyper::{
 use tokio::runtime::Runtime;
 use vector::{
     config, sinks,
-    sinks::util::{encoding::EncodingConfigWithFramingAdapter, BatchConfig, Compression},
+    sinks::util::{BatchConfig, Compression},
     sources,
     test_util::{next_addr, random_lines, runtime, send_lines, start_topology, wait_for_tcp},
     Error,
@@ -53,9 +54,8 @@ fn benchmark_http(c: &mut Criterion) {
                                 auth: Default::default(),
                                 headers: Default::default(),
                                 batch,
-                                encoding: EncodingConfigWithFramingAdapter::legacy(
-                                    sinks::http::Encoding::Text.into(),
-                                ),
+                                encoding: (None::<FramingConfig>, TextSerializerConfig::new())
+                                    .into(),
                                 request: Default::default(),
                                 tls: Default::default(),
                                 acknowledgements: Default::default(),

diff --git a/config/examples/docs_example.toml b/config/examples/docs_example.toml
@@ -32,12 +32,13 @@ index        = "vector-%Y-%m-%d"             # daily indices
 
 # Send structured data to a cost-effective long-term storage
 [sinks.s3_archives]
-inputs       = ["apache_parser"]             # don't sample for S3
-type         = "aws_s3"
-region       = "us-east-1"
-bucket       = "my-log-archives"
-key_prefix   = "date=%Y-%m-%d"               # daily partitions, hive friendly format
-compression  = "gzip"                        # compress final objects
-encoding     = "ndjson"                      # new line delimited JSON
+inputs         = ["apache_parser"]             # don't sample for S3
+type           = "aws_s3"
+region         = "us-east-1"
+bucket         = "my-log-archives"
+key_prefix     = "date=%Y-%m-%d"               # daily partitions, hive friendly format
+compression    = "gzip"                        # compress final objects
+framing.method = "newline_delimited"           # new line delimited...
+encoding.codec = "json"                        # ...JSON
 [sinks.s3_archives.batch]
-max_bytes   = 10000000                      # 10mb uncompressed
+max_bytes      = 10000000                      # 10mb uncompressed
diff --git a/config/examples/environment_variables.toml b/config/examples/environment_variables.toml
@@ -28,6 +28,6 @@ data_dir = "/var/lib/vector"
 # Print the data to STDOUT for inspection
 # Docs: https://vector.dev/docs/reference/sinks/console
 [sinks.out]
-  inputs       = ["add_host"]
-  type         = "console"
-  encoding     = "json"
+  inputs         = ["add_host"]
+  type           = "console"
+  encoding.codec = "json"
diff --git a/config/examples/es_s3_hybrid.toml b/config/examples/es_s3_hybrid.toml
@@ -26,11 +26,12 @@ data_dir = "/var/lib/vector"
 
 # Send structured data to S3, a durable long-term storage
 [sinks.s3_archives]
-  inputs       = ["apache_logs"] # don't sample
-  type         = "aws_s3"
-  region       = "us-east-1"
-  bucket       = "my_log_archives"
-  encoding     = "ndjson"
-  compression  = "gzip"
+  inputs         = ["apache_logs"] # don't sample
+  type           = "aws_s3"
+  region         = "us-east-1"
+  bucket         = "my_log_archives"
+  framing.method = "newline_delimited"
+  encoding.codec = "json"
+  compression    = "gzip"
   [sinks.s3_archives.batch]
-    max_size   = 10000000 # 10mb uncompressed
+    max_size     = 10000000 # 10mb uncompressed
diff --git a/config/examples/file_to_cloudwatch_metrics.toml b/config/examples/file_to_cloudwatch_metrics.toml
@@ -34,12 +34,12 @@ tags = {method = "{{method}}", status = "{{status}}"}
 [sinks.console_metrics]
 inputs = ["log_to_metric"]
 type = "console"
-encoding = "json"
+encoding.codec = "json"
 
 [sinks.console_logs]
 inputs = ["remap"]
 type = "console"
-encoding = "json"
+encoding.codec = "json"
 
 [sinks.cloudwatch]
 inputs = ["log_to_metric"]

diff --git a/config/examples/file_to_prometheus.toml b/config/examples/file_to_prometheus.toml
@@ -51,12 +51,12 @@ name = "bytes_out_histogram"
 [sinks.console_metrics]
 inputs = ["log_to_metric"]
 type = "console"
-encoding = "json"
+encoding.codec = "json"
 
 [sinks.console_logs]
 inputs = ["remap"]
 type = "console"
-encoding = "text"
+encoding.codec = "text"
 
 [sinks.prometheus]
 inputs = ["log_to_metric"]

diff --git a/config/examples/namespacing/sinks/s3_archives.toml b/config/examples/namespacing/sinks/s3_archives.toml
@@ -1,10 +1,11 @@
 # Send structured data to a cost-effective long-term storage
-inputs       = ["apache_parser"]             # don't sample for S3
-type         = "aws_s3"
-region       = "us-east-1"
-bucket       = "my-log-archives"
-key_prefix   = "date=%Y-%m-%d"               # daily partitions, hive friendly format
-compression  = "gzip"                        # compress final objects
-encoding     = "ndjson"                      # new line delimited JSON
+inputs         = ["apache_parser"]           # don't sample for S3
+type           = "aws_s3"
+region         = "us-east-1"
+bucket         = "my-log-archives"
+key_prefix     = "date=%Y-%m-%d"             # daily partitions, hive friendly format
+compression    = "gzip"                      # compress final objects
+framing.method = "newline_delimited"         # new line delimited...
+encoding.codec = "json"                      # ...JSON
 [batch]
-  max_bytes   = 10000000                      # 10mb uncompressed
+  max_bytes    = 10000000                    # 10mb uncompressed
diff --git a/config/examples/prometheus_to_console.toml b/config/examples/prometheus_to_console.toml
@@ -14,4 +14,4 @@ scrape_interval_secs = 2
 [sinks.console]
 inputs = ["prometheus"]
 type = "console"
-encoding = "json"
+encoding.codec = "json"
diff --git a/config/examples/stdio.toml b/config/examples/stdio.toml
@@ -11,4 +11,4 @@
 [sinks.out]
     inputs = ["in"]
     type = "console"
-    encoding = "text"
+    encoding.codec = "text"
diff --git a/config/examples/wrapped_json.toml b/config/examples/wrapped_json.toml
@@ -34,6 +34,6 @@ data_dir = "/var/lib/vector"
 # Print the data to STDOUT for inspection
 # Docs: https://vector.dev/docs/reference/sinks/console
 [sinks.out]
-  inputs       = ["parse_json"]
-  type         = "console"
-  encoding     = "json"
+  inputs         = ["parse_json"]
+  type           = "console"
+  encoding.codec = "json"
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
@@ -167,7 +167,7 @@ type = "stdin"
 [sinks.bar]
 type = "console"
 inputs = ["foo"]
-encoding = "json"
+encoding.codec = "json"
 ```
 
 After the component construction phase, we'll be left with the tasks for each

diff --git a/lib/codecs/src/encoding/format/json.rs b/lib/codecs/src/encoding/format/json.rs
@@ -68,22 +68,100 @@ impl Encoder<Event> for JsonSerializer {
 #[cfg(test)]
 mod tests {
     use bytes::BytesMut;
+    use chrono::{TimeZone, Utc};
     use vector_common::btreemap;
-    use vector_core::event::{LogEvent, Value};
+    use vector_core::event::{LogEvent, Metric, MetricKind, MetricValue, StatisticKind, Value};
 
     use super::*;
 
     #[test]
-    fn serialize_json() {
+    fn serialize_json_log() {
         let event = Event::Log(LogEvent::from(btreemap! {
-            "foo" => Value::from("bar")
+            "x" => Value::from("23"),
+            "z" => Value::from(25),
+            "a" => Value::from("0"),
         }));
         let mut serializer = JsonSerializer::new();
         let mut bytes = BytesMut::new();
 
         serializer.encode(event, &mut bytes).unwrap();
 
-        assert_eq!(bytes.freeze(), r#"{"foo":"bar"}"#);
+        assert_eq!(bytes.freeze(), r#"{"a":"0","x":"23","z":25}"#);
+    }
+
+    #[test]
+    fn serialize_json_metric_counter() {
+        let event = Event::Metric(
+            Metric::new(
+                "foos",
+                MetricKind::Incremental,
+                MetricValue::Counter { value: 100.0 },
+            )
+            .with_namespace(Some("vector"))
+            .with_tags(Some(
+                vec![
+                    ("key2".to_owned(), "value2".to_owned()),
+                    ("key1".to_owned(), "value1".to_owned()),
+                    ("Key3".to_owned(), "Value3".to_owned()),
+                ]
+                .into_iter()
+                .collect(),
+            ))
+            .with_timestamp(Some(Utc.ymd(2018, 11, 14).and_hms_nano(8, 9, 10, 11))),
+        );
+
+        let mut serializer = JsonSerializer::new();
+        let mut bytes = BytesMut::new();
+
+        serializer.encode(event, &mut bytes).unwrap();
+
+        assert_eq!(
+            bytes.freeze(),
+            r#"{"name":"foos","namespace":"vector","tags":{"Key3":"Value3","key1":"value1","key2":"value2"},"timestamp":"2018-11-14T08:09:10.000000011Z","kind":"incremental","counter":{"value":100.0}}"#
+        );
+    }
+
+    #[test]
+    fn serialize_json_metric_set() {
+        let event = Event::Metric(Metric::new(
+            "users",
+            MetricKind::Incremental,
+            MetricValue::Set {
+                values: vec!["bob".into()].into_iter().collect(),
+            },
+        ));
+
+        let mut serializer = JsonSerializer::new();
+        let mut bytes = BytesMut::new();
+
+        serializer.encode(event, &mut bytes).unwrap();
+
+        assert_eq!(
+            bytes.freeze(),
+            r#"{"name":"users","kind":"incremental","set":{"values":["bob"]}}"#
+        );
+    }
+
+    #[test]
+    fn serialize_json_metric_histogram_without_timestamp() {
+        let event = Event::Metric(Metric::new(
+            "glork",
+            MetricKind::Incremental,
+            MetricValue::Distribution {
+                samples: vector_core::samples![10.0 => 1],
+                statistic: StatisticKind::Histogram,
+            },
+        ));
+
+        let mut serializer = JsonSerializer::new();
+        let mut bytes = BytesMut::new();
+
+        serializer.encode(event, &mut bytes).unwrap();
+
+        assert_eq!(
+            bytes.freeze(),
+            r#"{"name":"glork","kind":"incremental","distribution":{"samples":[{"value":10.0,"rate":1}],"statistic":"histogram"}}"#
+        );
     }
 
     #[test]

diff --git a/lib/k8s-e2e-tests/tests/vector-agent.rs b/lib/k8s-e2e-tests/tests/vector-agent.rs
@@ -43,7 +43,7 @@ const CUSTOM_RESOURCE_VECTOR_CONFIG: &str = indoc! {r#"
         [sinks.stdout]
             type = "console"
             inputs = ["kubernetes_logs"]
-            encoding = "json"
+            encoding.codec = "json"
 "#};
 
 /// This test validates that vector picks up logs at the simplest case

diff --git a/rfcs/2020-03-06-1999-api-extensions-for-lua-transform.md b/rfcs/2020-03-06-1999-api-extensions-for-lua-transform.md
@@ -475,7 +475,7 @@ Here `event` is an encoded event to be produced by the transform, and `lane` is
 >    [sinks.example_console]
 >      type = "console"
 >      inputs = ["example_transform.example_lane"] # would output the event from `example_lane`
->      encoding = "text"
+>      encoding.codec = "text"
 >    ```
 >
 > Other components connected to the same transform, but with different lanes names or without lane names at all would not receive any event.

diff --git a/rfcs/2020-04-15-2341-wasm-plugins.md b/rfcs/2020-04-15-2341-wasm-plugins.md
@@ -333,7 +333,7 @@ module = "target/wasm32-wasi/release/echo.wasm"
 healthcheck = true
 inputs = ["demo"]
 type = "console"
-encoding = "json"
+encoding.codec = "json"
 buffer.type = "memory"
 buffer.max_events = 500
 buffer.when_full = "block"

diff --git a/rfcs/2021-07-20-8288-csv-enrichment.md b/rfcs/2021-07-20-8288-csv-enrichment.md
@@ -37,7 +37,7 @@ To represent the CSV file we have a new top level configuration option.
 ```toml
 [enrichment_tables.csv_file]
   type = "file"
-  encoding = "csv"
+  encoding.codec = "csv"
   path = "\path_to_csv"
   delimiter = ","
 ```

diff --git a/skaffold/manifests/config.yaml b/skaffold/manifests/config.yaml
@@ -8,4 +8,4 @@ data:
         type = "console"
         inputs = ["kubernetes_logs", "internal_metrics"]
         target = "stdout"
-        encoding = "json"
+        encoding.codec = "json"
diff --git a/soaks/disabled-tests/fluent_remap_aws_firehose/vector.toml b/soaks/disabled-tests/fluent_remap_aws_firehose/vector.toml
@@ -41,7 +41,7 @@ stream_name = "soak_fluent_remap_firehose"
 endpoint = "http://localhost:8080"
 healthcheck.enabled = true
 compression = "none"
-encoding = "json"
+encoding.codec = "json"
 region = "us-east-2"
 auth.access_key_id = "totallyanaccesskeyid"
 auth.secret_access_key = "alsoasecretaccesskey"
diff --git a/soaks/disabled-tests/http_datadog_filter_blackhole/vector.toml b/soaks/disabled-tests/http_datadog_filter_blackhole/vector.toml
@@ -12,7 +12,8 @@ type = "internal_metrics"
 [sources.logs]
 type = "http"
 address = "0.0.0.0:8282"
-encoding = "ndjson"
+framing.method = "newline_delimited"
+decoding.codec = "json"
 
 ##
 ## Transforms

diff --git a/soaks/tests/http_pipelines_blackhole/vector.toml b/soaks/tests/http_pipelines_blackhole/vector.toml
@@ -10,7 +10,7 @@ type = "internal_metrics"
 [sources.logs]
 type = "http"
 address = "0.0.0.0:8282"
-encoding = "text"
+decoding.codec = "bytes"
 
 ##
 ## Transforms

diff --git a/soaks/tests/http_pipelines_blackhole_acks/vector.toml b/soaks/tests/http_pipelines_blackhole_acks/vector.toml
@@ -10,7 +10,7 @@ type = "internal_metrics"
 [sources.logs]
 type = "http"
 address = "0.0.0.0:8282"
-encoding = "text"
+decoding.codec = "bytes"
 acknowledgements = true
 
 ##

diff --git a/soaks/tests/http_pipelines_no_grok_blackhole/vector.toml b/soaks/tests/http_pipelines_no_grok_blackhole/vector.toml
@@ -10,7 +10,7 @@ type = "internal_metrics"
 [sources.logs]
 type = "http"
 address = "0.0.0.0:8282"
-encoding = "text"
+decoding.codec = "bytes"
 
 ##
 ## Transforms