feat(opentelemetry source): add instrumentation scope to logs (#21407)

srstrickland · web-flow · commit 73a03a7fb582 · 2024-10-10T17:02:32.000Z
* feat(opentelemetry source): add instrumentation scope to logs This introduces an optional `scope` object to logs, representing the InstrumentationScope object from the OTel protocol. This object may include: * `name` (if nonempty) * `version` (if nonempty) * `attributes` (if nonempty) * `dropped_attributes_count` (if nonzero) This is particularly important for transmitting the logger name. This PR only addresses instrumentation scope for logs. resolves #21404 * Fix formatting * Fix docs * Fix website docs (round 2)
diff --git a/changelog.d/21404-opentelemetry-logs-instrumentation-scope.feature.md b/changelog.d/21404-opentelemetry-logs-instrumentation-scope.feature.md
@@ -0,0 +1,3 @@
+Adds `scope` information to logs received via the `opentelemetry` source
+
+authors: srstrickland
diff --git a/lib/opentelemetry-proto/src/convert.rs b/lib/opentelemetry-proto/src/convert.rs
@@ -14,7 +14,7 @@ use vrl::{
 };
 
 use super::proto::{
-    common::v1::{any_value::Value as PBValue, KeyValue},
+    common::v1::{any_value::Value as PBValue, InstrumentationScope, KeyValue},
     logs::v1::{LogRecord, ResourceLogs, SeverityNumber},
     resource::v1::Resource,
     trace::v1::{
@@ -27,6 +27,9 @@ const SOURCE_NAME: &str = "opentelemetry";
 
 pub const RESOURCE_KEY: &str = "resources";
 pub const ATTRIBUTES_KEY: &str = "attributes";
+pub const SCOPE_KEY: &str = "scope";
+pub const NAME_KEY: &str = "name";
+pub const VERSION_KEY: &str = "version";
 pub const TRACE_ID_KEY: &str = "trace_id";
 pub const SPAN_ID_KEY: &str = "span_id";
 pub const SEVERITY_TEXT_KEY: &str = "severity_text";
@@ -37,19 +40,20 @@ pub const FLAGS_KEY: &str = "flags";
 
 impl ResourceLogs {
     pub fn into_event_iter(self, log_namespace: LogNamespace) -> impl Iterator<Item = Event> {
-        let resource = self.resource;
         let now = Utc::now();
 
-        self.scope_logs
-            .into_iter()
-            .flat_map(|scope_log| scope_log.log_records)
-            .map(move |log_record| {
+        self.scope_logs.into_iter().flat_map(move |scope_log| {
+            let scope = scope_log.scope;
+            let resource = self.resource.clone();
+            scope_log.log_records.into_iter().map(move |log_record| {
                 ResourceLog {
                     resource: resource.clone(),
+                    scope: scope.clone(),
                     log_record,
                 }
                 .into_event(log_namespace, now)
             })
+        })
     }
 }
 
@@ -92,6 +96,7 @@ impl From<PBValue> for Value {
 
 struct ResourceLog {
     resource: Option<Resource>,
+    scope: Option<InstrumentationScope>,
     log_record: LogRecord,
 }
 
@@ -212,6 +217,49 @@ impl ResourceLog {
             }
         };
 
+        // Insert instrumentation scope (scope name, version, and attributes)
+        if let Some(scope) = self.scope {
+            if !scope.name.is_empty() {
+                log_namespace.insert_source_metadata(
+                    SOURCE_NAME,
+                    &mut log,
+                    Some(LegacyKey::Overwrite(path!(SCOPE_KEY, NAME_KEY))),
+                    path!(SCOPE_KEY, NAME_KEY),
+                    scope.name,
+                );
+            }
+            if !scope.version.is_empty() {
+                log_namespace.insert_source_metadata(
+                    SOURCE_NAME,
+                    &mut log,
+                    Some(LegacyKey::Overwrite(path!(SCOPE_KEY, VERSION_KEY))),
+                    path!(SCOPE_KEY, VERSION_KEY),
+                    scope.version,
+                );
+            }
+            if !scope.attributes.is_empty() {
+                log_namespace.insert_source_metadata(
+                    SOURCE_NAME,
+                    &mut log,
+                    Some(LegacyKey::Overwrite(path!(SCOPE_KEY, ATTRIBUTES_KEY))),
+                    path!(SCOPE_KEY, ATTRIBUTES_KEY),
+                    kv_list_into_value(scope.attributes),
+                );
+            }
+            if scope.dropped_attributes_count > 0 {
+                log_namespace.insert_source_metadata(
+                    SOURCE_NAME,
+                    &mut log,
+                    Some(LegacyKey::Overwrite(path!(
+                        SCOPE_KEY,
+                        DROPPED_ATTRIBUTES_COUNT_KEY
+                    ))),
+                    path!(SCOPE_KEY, DROPPED_ATTRIBUTES_COUNT_KEY),
+                    scope.dropped_attributes_count,
+                );
+            }
+        }
+
         // Optional fields
         if let Some(resource) = self.resource {
             if !resource.attributes.is_empty() {
diff --git a/src/sources/opentelemetry/tests.rs b/src/sources/opentelemetry/tests.rs
@@ -9,7 +9,7 @@ use vector_lib::config::LogNamespace;
 use vector_lib::lookup::path;
 use vector_lib::opentelemetry::proto::{
     collector::logs::v1::{logs_service_client::LogsServiceClient, ExportLogsServiceRequest},
-    common::v1::{any_value, AnyValue, KeyValue},
+    common::v1::{any_value, AnyValue, InstrumentationScope, KeyValue},
     logs::v1::{LogRecord, ResourceLogs, ScopeLogs},
     resource::v1::Resource as OtelResource,
 };
@@ -81,7 +81,17 @@ async fn receive_grpc_logs_vector_namespace() {
                     dropped_attributes_count: 0,
                 }),
                 scope_logs: vec![ScopeLogs {
-                    scope: None,
+                    scope: Some(InstrumentationScope {
+                        name: "some.scope.name".into(),
+                        version: "1.2.3".into(),
+                        attributes: vec![KeyValue {
+                            key: "scope_attr".into(),
+                            value: Some(AnyValue {
+                                value: Some(any_value::Value::StringValue("scope_val".into())),
+                            }),
+                        }],
+                        dropped_attributes_count: 7,
+                    }),
                     log_records: vec![LogRecord {
                         time_unix_nano: 1,
                         observed_time_unix_nano: 2,
@@ -133,6 +143,25 @@ async fn receive_grpc_logs_vector_namespace() {
             meta.get(path!("opentelemetry", "attributes")).unwrap(),
             &value!({attr_key: "attr_val"})
         );
+        assert_eq!(
+            meta.get(path!("opentelemetry", "scope", "name")).unwrap(),
+            &value!("some.scope.name")
+        );
+        assert_eq!(
+            meta.get(path!("opentelemetry", "scope", "version"))
+                .unwrap(),
+            &value!("1.2.3")
+        );
+        assert_eq!(
+            meta.get(path!("opentelemetry", "scope", "attributes"))
+                .unwrap(),
+            &value!({scope_attr: "scope_val"})
+        );
+        assert_eq!(
+            meta.get(path!("opentelemetry", "scope", "dropped_attributes_count"))
+                .unwrap(),
+            &value!(7)
+        );
         assert_eq!(
             meta.get(path!("opentelemetry", "trace_id")).unwrap(),
             &value!("4ac52aadf321c2e531db005df08792f5")
@@ -219,7 +248,17 @@ async fn receive_grpc_logs_legacy_namespace() {
                     dropped_attributes_count: 0,
                 }),
                 scope_logs: vec![ScopeLogs {
-                    scope: None,
+                    scope: Some(InstrumentationScope {
+                        name: "some.scope.name".into(),
+                        version: "1.2.3".into(),
+                        attributes: vec![KeyValue {
+                            key: "scope_attr".into(),
+                            value: Some(AnyValue {
+                                value: Some(any_value::Value::StringValue("scope_val".into())),
+                            }),
+                        }],
+                        dropped_attributes_count: 7,
+                    }),
                     log_records: vec![LogRecord {
                         time_unix_nano: 1,
                         observed_time_unix_nano: 2,
@@ -262,6 +301,18 @@ async fn receive_grpc_logs_legacy_namespace() {
                 "resources",
                 Value::Object(vec_into_btmap(vec![("res_key", "res_val".into())])),
             ),
+            (
+                "scope",
+                Value::Object(vec_into_btmap(vec![
+                    ("name", "some.scope.name".into()),
+                    ("version", "1.2.3".into()),
+                    (
+                        "attributes",
+                        Value::Object(vec_into_btmap(vec![("scope_attr", "scope_val".into())])),
+                    ),
+                    ("dropped_attributes_count", 7.into()),
+                ])),
+            ),
             ("message", "log body".into()),
             ("trace_id", "4ac52aadf321c2e531db005df08792f5".into()),
             ("span_id", "0b9e4bda2a55530d".into()),
diff --git a/website/cue/reference/components/sources/opentelemetry.cue b/website/cue/reference/components/sources/opentelemetry.cue
@@ -112,6 +112,46 @@ components: sources: opentelemetry: {
 						]
 					}
 				}
+				"scope.name": {
+					description: "Instrumentation scope name (often logger name)."
+					required:    false
+					common:      true
+					type: string: {
+						default: null
+						examples: ["some.module.name"]
+					}
+				}
+				"scope.version": {
+					description: "Instrumentation scope version."
+					required:    false
+					common:      false
+					type: string: {
+						default: null
+						examples: ["1.2.3"]
+					}
+				}
+				"scope.attributes": {
+					description: "Set of attributes that belong to the instrumentation scope."
+					required:    false
+					common:      false
+					type: object: {
+						examples: [
+							{
+								"attr1": "value1"
+								"attr2": "value2"
+								"attr3": "value3"
+							},
+						]
+					}
+				}
+				"scope.dropped_attributes_count": {
+					description: "Number of attributes dropped from the instrumentation scope (if not zero)."
+					required:    false
+					common:      false
+					type: uint: {
+						unit: null
+					}
+				}
 				message: {
 					description: "Contains the body of the log record."
 					required:    false

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	+Adds `scope` information to logs received via the `opentelemetry` source
	`2`	`+`
	`3`	`+authors: srstrickland`