starnix_core/task/
syslog.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::security;
6use crate::task::dynamic_thread_spawner::SpawnRequestBuilder;
7use crate::task::{
8    CurrentTask, EventHandler, ThreadLockupDetector, WaitCallback, WaitCanceler, WaitQueue, Waiter,
9};
10use crate::vfs::OutputBuffer;
11use diagnostics_data::{Data, Logs, LogsData, Severity};
12use estimate_timeline::{DefaultFetcher, TimeFetcher, TimelineEstimator};
13use fidl_fuchsia_diagnostics as fdiagnostics;
14use fuchsia_component::client::connect_to_protocol_sync;
15use fuchsia_inspect::Inspector;
16use futures::FutureExt;
17use serde::Deserialize;
18use starnix_sync::{LockDepMutex, Locked, SyslogStateLock, SyslogSubscriptionLock, Unlocked};
19use starnix_uapi::auth::CAP_SYSLOG;
20use starnix_uapi::errors::{EAGAIN, Errno, errno, error};
21use starnix_uapi::syslog::SyslogAction;
22use starnix_uapi::vfs::FdEvents;
23use std::cmp;
24use std::collections::VecDeque;
25use std::io::{self, Write};
26use std::sync::atomic::Ordering;
27use std::sync::{Arc, OnceLock, mpsc};
28use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout, TryFromBytes};
29
30const BUFFER_SIZE: i32 = 1_049_000;
31
32const NANOS_PER_SECOND: i64 = 1_000_000_000;
33const MICROS_PER_NANOSECOND: i64 = 1_000;
34
35#[derive(Default)]
36pub struct Syslog {
37    syscall_subscription: OnceLock<LockDepMutex<LogSubscription, SyslogSubscriptionLock>>,
38    state: Arc<LockDepMutex<TimelineEstimator<DefaultFetcher>, SyslogStateLock>>,
39}
40
41#[derive(Debug)]
42pub enum SyslogAccess {
43    DevKmsgRead,
44    ProcKmsg(SyslogAction),
45    Syscall(SyslogAction),
46}
47
48impl Syslog {
49    pub fn init(&self, system_task: &CurrentTask) -> Result<(), anyhow::Error> {
50        let state = self.state.clone();
51        system_task.kernel.inspect_node.record_lazy_child("syslog", move || {
52            let state = state.clone();
53            async move {
54                let inspector = Inspector::default();
55                let state_guard = state.lock();
56                inspector.root().record_uint("max_timeline_size", state_guard.max_timeline_size());
57                inspector
58                    .root()
59                    .record_uint("timeline_overflows", state_guard.timeline_overflows());
60                Ok(inspector)
61            }
62            .boxed()
63        });
64
65        let subscription = LogSubscription::snapshot_then_subscribe(system_task)?;
66        self.syscall_subscription.set(subscription.into()).expect("syslog inititialized once");
67        Ok(())
68    }
69
70    pub fn access(
71        &self,
72        current_task: &CurrentTask,
73        access: SyslogAccess,
74    ) -> Result<GrantedSyslog<'_>, Errno> {
75        Self::validate_access(current_task, access)?;
76        let syscall_subscription = self.subscription()?;
77        Ok(GrantedSyslog { syscall_subscription })
78    }
79
80    /// Validates that syslog access is unrestricted, or that the `current_task` has the relevant
81    /// capability, and applies the SELinux policy.
82    pub fn validate_access(current_task: &CurrentTask, access: SyslogAccess) -> Result<(), Errno> {
83        let (action, check_capabilities) = match access {
84            SyslogAccess::ProcKmsg(SyslogAction::Open) => (SyslogAction::Open, true),
85            SyslogAccess::DevKmsgRead => (SyslogAction::ReadAll, true),
86            SyslogAccess::Syscall(a) => (a, true),
87            // If we got here we already validated Open on /proc/kmsg.
88            SyslogAccess::ProcKmsg(a) => (a, false),
89        };
90
91        // According to syslog(2) man, ReadAll (3) and SizeBuffer (10) are allowed unprivileged
92        // access only if restrict_dmsg is 0.
93        let action_is_privileged = !matches!(
94            access,
95            SyslogAccess::Syscall(SyslogAction::ReadAll | SyslogAction::SizeBuffer)
96                | SyslogAccess::DevKmsgRead,
97        );
98        let restrict_dmesg = current_task.kernel().restrict_dmesg.load(Ordering::Relaxed);
99        if check_capabilities && (action_is_privileged || restrict_dmesg) {
100            security::check_task_capable(current_task, CAP_SYSLOG)?;
101        }
102
103        security::check_syslog_access(current_task, action)?;
104        Ok(())
105    }
106
107    pub fn snapshot_then_subscribe(current_task: &CurrentTask) -> Result<LogSubscription, Errno> {
108        LogSubscription::snapshot_then_subscribe(current_task)
109    }
110
111    pub fn subscribe(current_task: &CurrentTask) -> Result<LogSubscription, Errno> {
112        LogSubscription::subscribe(current_task)
113    }
114
115    fn subscription(
116        &self,
117    ) -> Result<&LockDepMutex<LogSubscription, SyslogSubscriptionLock>, Errno> {
118        self.syscall_subscription.get().ok_or_else(|| errno!(ENOENT))
119    }
120}
121
122pub struct GrantedSyslog<'a> {
123    syscall_subscription: &'a LockDepMutex<LogSubscription, SyslogSubscriptionLock>,
124}
125
126impl GrantedSyslog<'_> {
127    pub fn read(&self, out: &mut dyn OutputBuffer) -> Result<i32, Errno> {
128        let mut subscription = self.syscall_subscription.lock();
129        if let Some(log) = subscription.try_next()? {
130            let size_to_write = cmp::min(log.len(), out.available() as usize);
131            out.write(&log[..size_to_write])?;
132            return Ok(size_to_write as i32);
133        }
134        Ok(0)
135    }
136
137    pub fn wait(&self, waiter: &Waiter, events: FdEvents, handler: EventHandler) -> WaitCanceler {
138        self.syscall_subscription.lock().wait(waiter, events, handler)
139    }
140
141    pub fn blocking_read(
142        &self,
143        locked: &mut Locked<Unlocked>,
144        current_task: &CurrentTask,
145        out: &mut dyn OutputBuffer,
146    ) -> Result<i32, Errno> {
147        let mut write_log = |log: Vec<u8>| {
148            let size_to_write = cmp::min(log.len(), out.available() as usize);
149            out.write(&log[..size_to_write])?;
150            Ok(size_to_write as i32)
151        };
152        match self.syscall_subscription.lock().try_next() {
153            Err(errno) if errno == EAGAIN => {}
154            Err(errno) => return Err(errno),
155            Ok(Some(log)) => return write_log(log),
156            Ok(None) => return Ok(0),
157        }
158        let waiter = Waiter::new();
159        loop {
160            let mut subscription = self.syscall_subscription.lock();
161            subscription.wait(&waiter, FdEvents::POLLIN | FdEvents::POLLHUP, WaitCallback::none());
162            match subscription.try_next() {
163                Err(errno) if errno == EAGAIN => {}
164                Err(errno) => return Err(errno),
165                Ok(Some(log)) => return write_log(log),
166                Ok(None) => return Ok(0),
167            }
168            // Drop the subscription lock to avoid holding a lock while waiting.
169            std::mem::drop(subscription);
170            waiter.wait(locked, current_task)?;
171        }
172    }
173
174    pub fn read_all(
175        &self,
176        current_task: &CurrentTask,
177        out: &mut dyn OutputBuffer,
178    ) -> Result<i32, Errno> {
179        let mut subscription = LogSubscription::snapshot(current_task)?;
180        let mut buffer = ResultBuffer::new(out.available());
181        while let Some(log_result) = subscription.next() {
182            buffer.push(log_result?);
183        }
184        let result: Vec<u8> = buffer.into();
185        out.write(result.as_slice())?;
186        Ok(result.len() as i32)
187    }
188
189    pub fn size_unread(&self) -> Result<i32, Errno> {
190        let mut subscription = self.syscall_subscription.lock();
191        Ok(subscription.available()?.try_into().unwrap_or(std::i32::MAX))
192    }
193
194    pub fn size_buffer(&self) -> Result<i32, Errno> {
195        // For now always return a constant for this.
196        Ok(BUFFER_SIZE)
197    }
198}
199
200#[derive(Debug)]
201pub struct LogSubscription {
202    pending: Option<Vec<u8>>,
203    receiver: mpsc::Receiver<Result<Vec<u8>, Errno>>,
204    waiters: Arc<WaitQueue>,
205}
206
207#[derive(Debug, Deserialize)]
208#[serde(untagged)]
209enum OneOrMany<T> {
210    Many(Vec<T>),
211    One(T),
212}
213
214impl LogSubscription {
215    pub fn wait(&self, waiter: &Waiter, events: FdEvents, handler: EventHandler) -> WaitCanceler {
216        self.waiters.wait_async_fd_events(waiter, events, handler)
217    }
218
219    pub fn available(&mut self) -> Result<usize, Errno> {
220        if let Some(log) = &self.pending {
221            return Ok(log.len());
222        }
223        match self.try_next() {
224            Err(err) if err == EAGAIN => Ok(0),
225            Err(err) => Err(err),
226            Ok(Some(log)) => {
227                let size = log.len();
228                self.pending.replace(log);
229                return Ok(size);
230            }
231            Ok(None) => Ok(0),
232        }
233    }
234
235    fn snapshot(current_task: &CurrentTask) -> Result<LogIterator, Errno> {
236        LogIterator::new(&current_task.kernel.syslog, fdiagnostics::StreamMode::Snapshot)
237    }
238
239    fn subscribe(current_task: &CurrentTask) -> Result<Self, Errno> {
240        Self::new_listening(current_task, fdiagnostics::StreamMode::Subscribe)
241    }
242
243    fn snapshot_then_subscribe(current_task: &CurrentTask) -> Result<Self, Errno> {
244        Self::new_listening(current_task, fdiagnostics::StreamMode::SnapshotThenSubscribe)
245    }
246
247    fn new_listening(
248        current_task: &CurrentTask,
249        mode: fdiagnostics::StreamMode,
250    ) -> Result<Self, Errno> {
251        let iterator = LogIterator::new(&current_task.kernel.syslog, mode)?;
252        let (snd, receiver) = mpsc::sync_channel(1);
253        let waiters = Arc::new(WaitQueue::default());
254        let waiters_clone = waiters.clone();
255        let closure = move |_: &mut Locked<Unlocked>, _: &CurrentTask| {
256            scopeguard::defer! {
257                waiters_clone.notify_fd_events(FdEvents::POLLHUP);
258            };
259            for log in iterator {
260                let send_result = {
261                    let _waiting_guard = ThreadLockupDetector::pause_tracking();
262                    snd.send(log)
263                };
264                if send_result.is_err() {
265                    break;
266                };
267                waiters_clone.notify_fd_events(FdEvents::POLLIN);
268            }
269        };
270        let req = SpawnRequestBuilder::new()
271            .with_debug_name("syslog-listener")
272            .with_sync_closure(closure)
273            .build();
274        current_task.kernel().kthreads.spawner().spawn_from_request(req);
275
276        Ok(Self { receiver, waiters, pending: Default::default() })
277    }
278
279    fn try_next(&mut self) -> Result<Option<Vec<u8>>, Errno> {
280        if let Some(value) = self.pending.take() {
281            return Ok(Some(value));
282        }
283        match self.receiver.try_recv() {
284            // We got the next log.
285            Ok(Ok(log)) => Ok(Some(log)),
286            // An error happened attempting to get the next log.
287            Ok(Err(err)) => Err(err),
288            // The channel was closed and there's no more messages in the queue.
289            Err(mpsc::TryRecvError::Disconnected) => Ok(None),
290            // No messages available but the channel hasn't closed.
291            Err(mpsc::TryRecvError::Empty) => error!(EAGAIN),
292        }
293    }
294}
295
296struct LogIterator {
297    iterator: fdiagnostics::BatchIteratorSynchronousProxy,
298    pending_formatted_contents: VecDeque<fdiagnostics::FormattedContent>,
299    pending_datas: VecDeque<Data<Logs>>,
300    state: Arc<LockDepMutex<TimelineEstimator<DefaultFetcher>, SyslogStateLock>>,
301    tags: std::collections::HashMap<u64, diagnostics_message::MonikerWithUrl>,
302}
303
304impl LogIterator {
305    fn new(syslog: &Syslog, mode: fdiagnostics::StreamMode) -> Result<Self, Errno> {
306        let accessor = connect_to_protocol_sync::<fdiagnostics::ArchiveAccessorMarker>()
307            .map_err(|_| errno!(ENOENT, format!("Failed to connecto to ArchiveAccessor")))?;
308        let is_subscribe = matches!(mode, fdiagnostics::StreamMode::Subscribe);
309        let stream_parameters = fdiagnostics::StreamParameters {
310            stream_mode: Some(mode),
311            data_type: Some(fdiagnostics::DataType::Logs),
312            format: Some(fdiagnostics::Format::Fxt),
313            client_selector_configuration: Some(
314                fdiagnostics::ClientSelectorConfiguration::SelectAll(true),
315            ),
316            ..fdiagnostics::StreamParameters::default()
317        };
318        let (client_end, server_end) =
319            fidl::endpoints::create_endpoints::<fdiagnostics::BatchIteratorMarker>();
320        accessor.stream_diagnostics(&stream_parameters, server_end).map_err(|err| {
321            errno!(EIO, format!("ArchiveAccessor/StreamDiagnostics failed: {err}"))
322        })?;
323        let iterator = fdiagnostics::BatchIteratorSynchronousProxy::new(client_end.into_channel());
324        if is_subscribe {
325            let () = iterator.wait_for_ready(zx::MonotonicInstant::INFINITE).map_err(|err| {
326                errno!(EIO, format!("Failed to wait for BatchIterator being ready: {err}"))
327            })?;
328        }
329        Ok(Self {
330            iterator,
331            pending_formatted_contents: VecDeque::new(),
332            pending_datas: VecDeque::new(),
333            state: syslog.state.clone(),
334            tags: std::collections::HashMap::new(),
335        })
336    }
337
338    // TODO(b/315520045): Investigate if we should make this
339    // not allocate anything.
340    fn get_next(&mut self) -> Result<Option<Vec<u8>>, Errno> {
341        'main_loop: loop {
342            while let Some(data) = self.pending_datas.pop_front() {
343                if let Some(log) = format_log(data, &self.state).map_err(|_| errno!(EIO))? {
344                    return Ok(Some(log));
345                }
346            }
347            while let Some(formatted_content) = self.pending_formatted_contents.pop_front() {
348                let output: OneOrMany<Data<Logs>> = match formatted_content {
349                    fdiagnostics::FormattedContent::Fxt(data) => {
350                        let buf = data
351                            .read_to_vec(
352                                0,
353                                data.get_content_size().map_err(|a| {
354                                    errno!(EIO, format!("Error {a} getting VMO size"))
355                                })?,
356                            )
357                            .map_err(|err| {
358                                errno!(EIO, format!("failed to read logs vmo: {err}"))
359                            })?;
360                        let mut current_slice = buf.as_ref();
361                        let mut ret: Option<OneOrMany<LogsData>> = None;
362                        loop {
363                            let (record, remaining) =
364                                diagnostics_log_encoding::parse::parse_record(current_slice)
365                                    .map_err(|a| errno!(EIO, format!("Error {a} parsing FXT")))?;
366
367                            let record_len = current_slice.len() - remaining.len();
368                            let record_bytes = &current_slice[..record_len];
369
370                            let header = diagnostics_log_encoding::Header::read_from_bytes(
371                                &current_slice[..8],
372                            )
373                            .map_err(|_| errno!(EIO, "Invalid FXT header"))?;
374                            let tag = header.tag();
375                            let is_manifest =
376                                (tag & diagnostics_log_encoding::LOG_CONTROL_BIT) != 0;
377                            let actual_tag = tag & !diagnostics_log_encoding::LOG_CONTROL_BIT;
378
379                            if is_manifest {
380                                let mut moniker = None;
381                                let mut url = None;
382                                for arg in &record.arguments {
383                                    use diagnostics_log_encoding::Value;
384                                    if arg.name() == "moniker" {
385                                        if let Value::Text(t) = arg.value() {
386                                            moniker = Some(diagnostics_data::ExtendedMoniker::parse_str(&t).unwrap_or_else(|_| diagnostics_data::ExtendedMoniker::ComponentInstance(moniker::Moniker::parse_str("unknown").unwrap())));
387                                        }
388                                    } else if arg.name() == "url" {
389                                        if let Value::Text(t) = arg.value() {
390                                            url = Some(flyweights::FlyStr::new(t));
391                                        }
392                                    }
393                                }
394                                if let (Some(moniker), Some(url)) = (moniker, url) {
395                                    self.tags.insert(
396                                        actual_tag as u64,
397                                        diagnostics_message::MonikerWithUrl { moniker, url },
398                                    );
399                                }
400                            } else {
401                                let source = self
402                                    .tags
403                                    .get(&(actual_tag as u64))
404                                    .cloned()
405                                    .unwrap_or_else(|| diagnostics_message::MonikerWithUrl {
406                                        moniker:
407                                            diagnostics_data::ExtendedMoniker::ComponentInstance(
408                                                moniker::Moniker::parse_str("unknown").unwrap(),
409                                            ),
410                                        url: flyweights::FlyStr::new("unknown"),
411                                    });
412
413                                let data =
414                                    diagnostics_message::from_structured(source, record_bytes)
415                                        .map_err(|a| {
416                                            errno!(EIO, format!("Error {a} parsing FXT"))
417                                        })?;
418
419                                ret = Some(match ret.take() {
420                                    Some(OneOrMany::One(one)) => OneOrMany::Many(vec![one, data]),
421                                    Some(OneOrMany::Many(mut many)) => {
422                                        many.push(data);
423                                        OneOrMany::Many(many)
424                                    }
425                                    None => OneOrMany::One(data),
426                                });
427                            }
428
429                            if remaining.is_empty() {
430                                break;
431                            }
432                            current_slice = remaining;
433                        }
434                        ret.unwrap_or_else(|| OneOrMany::Many(vec![]))
435                    }
436                    format => {
437                        unreachable!("we only request and expect one format. Got: {format:?}")
438                    }
439                };
440                match output {
441                    OneOrMany::One(data) => {
442                        if let Some(log) = format_log(data, &self.state).map_err(|_| errno!(EIO))? {
443                            return Ok(Some(log));
444                        }
445                    }
446                    OneOrMany::Many(datas) => {
447                        if datas.len() > 0 {
448                            self.pending_datas.extend(datas);
449                            continue 'main_loop;
450                        }
451                    }
452                }
453            }
454            let next_batch = {
455                let _waiting_guard = ThreadLockupDetector::pause_tracking();
456                self.iterator
457                    .get_next(zx::MonotonicInstant::INFINITE)
458                    .map_err(|_| errno!(ENOENT))?
459                    .map_err(|_| errno!(ENOENT))?
460            };
461            if next_batch.is_empty() {
462                return Ok(None);
463            }
464            self.pending_formatted_contents = VecDeque::from(next_batch);
465        }
466    }
467}
468
469impl Iterator for LogIterator {
470    type Item = Result<Vec<u8>, Errno>;
471
472    fn next(&mut self) -> Option<Result<Vec<u8>, Errno>> {
473        self.get_next().transpose()
474    }
475}
476
477impl Iterator for LogSubscription {
478    type Item = Result<Vec<u8>, Errno>;
479
480    fn next(&mut self) -> Option<Self::Item> {
481        self.try_next().transpose()
482    }
483}
484
485struct ResultBuffer {
486    max_size: usize,
487    buffer: VecDeque<Vec<u8>>,
488    current_size: usize,
489}
490
491impl ResultBuffer {
492    fn new(max_size: usize) -> Self {
493        Self { max_size, buffer: VecDeque::default(), current_size: 0 }
494    }
495
496    fn push(&mut self, data: Vec<u8>) {
497        while !self.buffer.is_empty() && self.current_size + data.len() > self.max_size {
498            let old = self.buffer.pop_front().unwrap();
499            self.current_size -= old.len();
500        }
501        self.current_size += data.len();
502        self.buffer.push_back(data);
503    }
504}
505
506impl Into<Vec<u8>> for ResultBuffer {
507    fn into(self) -> Vec<u8> {
508        let mut result = Vec::with_capacity(self.current_size);
509        for mut item in self.buffer {
510            result.append(&mut item);
511        }
512        // If we still exceed the size (for example, a single message of size N in a buffer of
513        // size M when N>M), we trim the output.
514        result.truncate(self.max_size);
515        result
516    }
517}
518
519#[derive(Debug, Eq, PartialEq, Copy, Clone, KnownLayout, TryFromBytes, Immutable, IntoBytes)]
520#[repr(u8)]
521pub enum KmsgLevel {
522    Emergency = 0,
523    Alert = 1,
524    Critical = 2,
525    Error = 3,
526    Warning = 4,
527    Notice = 5,
528    Info = 6,
529    Debug = 7,
530}
531
532impl KmsgLevel {
533    fn from_raw(value: u8) -> Option<KmsgLevel> {
534        zerocopy::try_transmute!(value).ok()
535    }
536}
537
538/// Given a string starting with <[0-9]*>, returns the level interpreted from the lower 3 bits.
539/// The next 8 is the facility, which we ignore atm.
540/// If the string doesn't start with a valid level, we return None.
541/// The slice returned is the rest of the message after the closing '>'.
542///
543/// Reference: https://www.kernel.org/doc/Documentation/ABI/testing/dev-kmsg
544pub(crate) fn extract_level(msg: &[u8]) -> Option<(KmsgLevel, &[u8])> {
545    let mut bytes_iter = msg.iter();
546    let Some(c) = bytes_iter.next() else {
547        return None;
548    };
549    if *c != b'<' {
550        return None;
551    }
552    let Some(end) = bytes_iter.enumerate().find(|(_, c)| **c == b'>').map(|(i, _)| i + 1) else {
553        return None;
554    };
555    std::str::from_utf8(&msg[1..end])
556        .ok()
557        .and_then(|s| s.parse::<u64>().ok())
558        .map(|level| (level & 0x07) as u8)
559        .and_then(KmsgLevel::from_raw)
560        .map(|level| (level, &msg[end + 1..]))
561}
562
563fn format_log<T: TimeFetcher>(
564    data: Data<Logs>,
565    state: &Arc<LockDepMutex<TimelineEstimator<T>, SyslogStateLock>>,
566) -> Result<Option<Vec<u8>>, io::Error> {
567    let mut formatted_tags = match data.tags() {
568        None => vec![],
569        Some(tags) => {
570            let mut formatted = vec![];
571            for (i, tag) in tags.iter().enumerate() {
572                // TODO(b/299533466): remove this.
573                if tag.contains("fxlogcat") {
574                    return Ok(None);
575                }
576                if i != 0 {
577                    write!(&mut formatted, ",")?;
578                }
579                write!(&mut formatted, "{tag}")?;
580            }
581            write!(&mut formatted, ": ")?;
582            formatted
583        }
584    };
585
586    let mut result = Vec::<u8>::new();
587    let (level, msg_after_level) = match data.msg().and_then(|msg| extract_level(msg.as_bytes())) {
588        Some((level, remaining_msg)) => (level as u8, Some(remaining_msg)),
589        None => match data.severity() {
590            Severity::Trace | Severity::Debug => (KmsgLevel::Debug as u8, None),
591            Severity::Info => (KmsgLevel::Info as u8, None),
592            Severity::Warn => (KmsgLevel::Warning as u8, None),
593            Severity::Error => (KmsgLevel::Error as u8, None),
594            Severity::Fatal => (KmsgLevel::Critical as u8, None),
595        },
596    };
597
598    // TODO(https://fxbug.dev/433724019): this isn't correct strictly speaking, but will be in most
599    // cases. We unapply the *current* offset and in the case where suspension happened between
600    // when the log message was generated and when Starnix is forwarding the log message, this will
601    // be different from the *actual* offset prior to suspension.
602    let time = state.lock().boot_time_to_monotonic_time(data.metadata.timestamp);
603    let time_nanos = time.into_nanos();
604    let time_secs = time_nanos / NANOS_PER_SECOND;
605    // Microsecond-level precision fractional time.
606    let time_fract = (time_nanos % NANOS_PER_SECOND) / MICROS_PER_NANOSECOND;
607    let component_name = data.component_name();
608    write!(&mut result, "<{level}>[{time_secs:05}.{time_fract:06}] {component_name}",)?;
609
610    match data.metadata.pid {
611        Some(pid) => write!(&mut result, "[{pid}]: ")?,
612        None => write!(&mut result, ": ")?,
613    }
614
615    result.append(&mut formatted_tags);
616
617    if let Some(msg) = msg_after_level {
618        write!(&mut result, "{}", String::from_utf8_lossy(msg))?;
619    } else if let Some(msg) = data.msg() {
620        write!(&mut result, "{msg}")?;
621    }
622
623    for kvp in data.payload_keys_strings() {
624        write!(&mut result, " {kvp}")?;
625    }
626    write!(&mut result, "\n")?;
627    Ok(Some(result))
628}
629
630#[cfg(test)]
631mod tests {
632    use super::*;
633
634    #[test]
635    fn test_result_buffer() {
636        let mut buffer = ResultBuffer::new(100);
637        buffer.push(vec![0; 200]);
638        let result: Vec<u8> = buffer.into();
639        assert_eq!(result.len(), 100);
640
641        let mut buffer = ResultBuffer::new(100);
642        buffer.push(Vec::from_iter(0..20));
643        buffer.push(Vec::from_iter(20..50));
644        let result: Vec<u8> = buffer.into();
645        assert_eq!(result.len(), 50);
646        for i in 0..50u8 {
647            assert_eq!(result[i as usize], i);
648        }
649
650        let mut buffer = ResultBuffer::new(100);
651        buffer.push(Vec::from_iter(0..20));
652        buffer.push(Vec::from_iter(20..150));
653        let result: Vec<u8> = buffer.into();
654        assert_eq!(result.len(), 100);
655        for i in 0..100u8 {
656            assert_eq!(result[i as usize], i + 20u8);
657        }
658
659        let mut buffer = ResultBuffer::new(100);
660        buffer.push(Vec::from_iter(0..20));
661        buffer.push(Vec::from_iter(20..150));
662        buffer.push(Vec::from_iter(150..210));
663        let result: Vec<u8> = buffer.into();
664        assert_eq!(result.len(), 60);
665        for i in 0..60u8 {
666            assert_eq!(result[i as usize], i + 150u8);
667        }
668    }
669
670    #[test]
671    fn test_extract_level() {
672        for level in 0..8 {
673            let msg = format!("<{level}> some message");
674            let result = extract_level(msg.as_bytes()).map(|(x, i)| (x as u8, i));
675            assert_eq!(Some((level, " some message".as_bytes())), result);
676        }
677    }
678
679    #[test]
680    fn parse_message_accepts_levels_greater_than_7() {
681        assert_eq!(
682            Some((KmsgLevel::Warning, " message".as_bytes())),
683            extract_level("<100> message".as_bytes())
684        );
685    }
686
687    #[test]
688    fn parse_message_defaults_when_non_numbers() {
689        assert_eq!(None, extract_level("<a> some message".as_bytes()));
690    }
691
692    #[test]
693    fn parse_message_defaults_when_invalid_level_syntax() {
694        assert_eq!(None, extract_level("<1 some message".as_bytes()));
695    }
696
697    #[test]
698    fn parse_message_defaults_when_no_level() {
699        assert_eq!(None, extract_level("some message".as_bytes()));
700    }
701}
starnix_core/task/syslog.rs

starnix_core/task/
syslog.rs
