Add target,file,line field to Log

duo-api/proto/log.proto

@@ -14,8 +14,15 @@ message Log {
     optional uint64 trace_id = 3;
     // Describes the level of verbosity of a log.
     common.Level level = 4;
+    // The part of the system that the span that this metadata describes
+    // occurred in.
+    string target = 5;
+    // The name of the source code file where the log occurred.
+    optional string file = 6;
+    // The line number in the source code file where the log occurred.
+    optional uint32 line = 7;
     // Timestamp.
-    google.protobuf.Timestamp time = 5;
+    google.protobuf.Timestamp time = 8;
     // Key-value fileds.
-    map<string, common.Value> fields = 6;
+    map<string, common.Value> fields = 9;
 }

duo-subscriber/src/subscriber.rs

@@ -160,6 +160,9 @@ where
             process_id: String::new(),
             span_id,
             trace_id,
+            target: metadata.target().into(),
+            file: metadata.file().map(Into::into),
+            line: metadata.line(),
             level: proto::Level::from(*metadata.level()) as i32,
             time: Some(SystemTime::now().into()),
             fields,

duo/src/arrow.rs

@@ -62,11 +62,14 @@ pub fn convert_log_to_record_batch(logs: Vec<Log>) -> Result<RecordBatch> {
         map.insert("span_id".into(), log.span_id.into());
         map.insert("trace_id".into(), log.trace_id.into());
         map.insert("level".into(), log.level.as_str().into());
+        map.insert("target".into(), log.target.into());
+        map.insert("file".into(), log.file.into());
+        map.insert("line".into(), log.line.into());
         map.insert("time".into(), time.into());
         map.insert("message".into(), log.message.into());
         let mut field_map = Map::new();
         for (key, value) in log.fields {
-            field_map.insert(key.clone(), value.clone());
+            field_map.insert(key, value);
         }
 
         if !field_map.is_empty() {
@@ -76,7 +79,7 @@ pub fn convert_log_to_record_batch(logs: Vec<Log>) -> Result<RecordBatch> {
         data.push(JsonValue::Object(map));
     }
 
-    let inferred_field_schema = infer_json_schema_from_iterator(fields.iter().map(Ok))?;
+    let inferred_field_schema = infer_json_schema_from_iterator(fields.iter().map(Ok)).unwrap();
     let schema = Schema::try_merge(vec![
         (*schema::get_log_schema()).clone(),
         inferred_field_schema,

duo/src/models.rs

@@ -41,6 +41,9 @@ pub struct Log {
     // TODO: change level to i32
     #[serde(with = "deser::level")]
     pub level: Level,
+    pub target: String,
+    pub file: Option<String>,
+    pub line: Option<u32>,
     #[serde(with = "deser::miscrosecond")]
     pub time: OffsetDateTime,
     pub message: String,
@@ -163,6 +166,9 @@ impl From<proto::Log> for Log {
             span_id: log.span_id,
             trace_id: log.trace_id,
             level,
+            target: log.target,
+            file: log.file,
+            line: log.line,
             time: log
                 .time
                 .and_then(|timestamp| {

duo/src/schema.rs

@@ -34,6 +34,9 @@ fn default_log_schema() -> Arc<Schema> {
         Field::new("trace_id", DataType::UInt64, true),
         Field::new("span_id", DataType::UInt64, true),
         Field::new("level", DataType::Utf8, false),
+        Field::new("target", DataType::Utf8, true),
+        Field::new("file", DataType::Utf8, true),
+        Field::new("line", DataType::UInt32, true),
         Field::new("message", DataType::Utf8, true),
     ]))
 }
@@ -51,8 +54,11 @@ pub async fn load() -> Result<()> {
     {
         Ok(data) => {
             let schema = serde_json::from_slice::<Schema>(&data.bytes().await?)?;
+            let latest_schema =
+                Schema::try_merge(vec![(*default_log_schema()).clone(), schema]).unwrap();
+            LOG_SCHEMA_DIRTY.store(true, Ordering::Relaxed);
             LOG_SCHEMA
-                .set(RwLock::new(Arc::new(schema)))
+                .set(RwLock::new(Arc::new(latest_schema)))
                 .expect("LogSchema already initialized");
         }
         Err(_err) => {

duo/src/web/logs.rs

@@ -24,6 +24,8 @@ pub(super) struct QueryParameters {
     pub start: Option<OffsetDateTime>,
     #[serde(default, deserialize_with = "deser::option_miscrosecond")]
     pub end: Option<OffsetDateTime>,
+    keyword: Option<String>,
+    level: Option<String>,
 }
 
 #[tracing::instrument]

duo/src/web/serialize.rs

@@ -163,6 +163,13 @@ impl<'a> Serialize for JaegerLog<'a> {
         let mut fields = HashMap::new();
         fields.insert("message".into(), self.0.message.clone().into());
         fields.insert("level".into(), self.0.level.as_str().into());
+        fields.insert("target".into(), self.0.target.as_str().into());
+        if let Some(file) = &self.0.file {
+            fields.insert(
+                "file".into(),
+                format!("{}:{}", &file, self.0.line.unwrap_or_default()).into(),
+            );
+        }
         fields.extend(self.0.fields.clone());
         map.serialize_entry(
             "fields",