journald: use unconnected socket (#1758)

Lets journald subscribers survive a journald restart.

Closes #1745

## Motivation

Currently the journald subscriber immediately connects to the journald
socket.  As such I understand it'd not survive a full restart of
journald.

## Solution

Do not connect the client socket immediately; instead pass the socket
pathname every time we send a message.  This is also what upstream does.

Author: swsnr

tracing-journald/src/lib.rs

@@ -86,6 +86,9 @@ pub struct Layer {
     field_prefix: Option<String>,
 }
 
+#[cfg(unix)]
+const JOURNALD_PATH: &str = "/run/systemd/journal/socket";
+
 impl Layer {
     /// Construct a journald layer
     ///
@@ -95,11 +98,14 @@ impl Layer {
         #[cfg(unix)]
         {
             let socket = UnixDatagram::unbound()?;
-            socket.connect("/run/systemd/journal/socket")?;
-            Ok(Self {
+            let layer = Self {
                 socket,
                 field_prefix: Some("F".into()),
-            })
+            };
+            // Check that we can talk to journald, by sending empty payload which journald discards.
+            // However if the socket didn't exist or if none listened we'd get an error here.
+            layer.send_payload(&[])?;
+            Ok(layer)
         }
         #[cfg(not(unix))]
         Err(io::Error::new(
@@ -125,13 +131,15 @@ impl Layer {
 
     #[cfg(unix)]
     fn send_payload(&self, payload: &[u8]) -> io::Result<usize> {
-        self.socket.send(payload).or_else(|error| {
-            if Some(libc::EMSGSIZE) == error.raw_os_error() {
-                self.send_large_payload(payload)
-            } else {
-                Err(error)
-            }
-        })
+        self.socket
+            .send_to(payload, JOURNALD_PATH)
+            .or_else(|error| {
+                if Some(libc::EMSGSIZE) == error.raw_os_error() {
+                    self.send_large_payload(payload)
+                } else {
+                    Err(error)
+                }
+            })
     }
 
     #[cfg(all(unix, not(target_os = "linux")))]
@@ -154,7 +162,7 @@ impl Layer {
         // Fully seal the memfd to signal journald that its backing data won't resize anymore
         // and so is safe to mmap.
         memfd::seal_fully(mem.as_raw_fd())?;
-        socket::send_one_fd(&self.socket, mem.as_raw_fd())
+        socket::send_one_fd_to(&self.socket, mem.as_raw_fd(), JOURNALD_PATH)
     }
 }
 

tracing-journald/src/socket.rs

@@ -2,8 +2,10 @@
 
 use std::io::{Error, Result};
 use std::mem::{size_of, zeroed};
+use std::os::unix::ffi::OsStrExt;
 use std::os::unix::net::UnixDatagram;
 use std::os::unix::prelude::{AsRawFd, RawFd};
+use std::path::Path;
 use std::ptr;
 
 use libc::*;
@@ -31,18 +33,38 @@ fn cmsg_buffer_size_for_one_fd() {
     assert_cmsg_bufsize()
 }
 
-pub fn send_one_fd(socket: &UnixDatagram, fd: RawFd) -> Result<usize> {
+pub fn send_one_fd_to<P: AsRef<Path>>(socket: &UnixDatagram, fd: RawFd, path: P) -> Result<usize> {
     assert_cmsg_bufsize();
 
-    let mut cmsg_buffer = AlignedBuffer {
-        buffer: ([0u8; CMSG_BUFSIZE]),
+    let mut addr: sockaddr_un = unsafe { zeroed() };
+    let path_bytes = path.as_ref().as_os_str().as_bytes();
+    // path_bytes may have at most sun_path + 1 bytes, to account for the trailing NUL byte.
+    if addr.sun_path.len() <= path_bytes.len() {
+        return Err(Error::from_raw_os_error(ENAMETOOLONG));
+    }
+
+    addr.sun_family = AF_UNIX as _;
+    unsafe {
+        std::ptr::copy_nonoverlapping(
+            path_bytes.as_ptr(),
+            addr.sun_path.as_mut_ptr() as *mut u8,
+            path_bytes.len(),
+        )
     };
+
     let mut msg: msghdr = unsafe { zeroed() };
+    // Set the target address.
+    msg.msg_name = &mut addr as *mut _ as *mut c_void;
+    msg.msg_namelen = size_of::<sockaddr_un>() as socklen_t;
 
     // We send no data body with this message.
     msg.msg_iov = ptr::null_mut();
     msg.msg_iovlen = 0;
 
+    // Create and fill the control message buffer with our file descriptor
+    let mut cmsg_buffer = AlignedBuffer {
+        buffer: ([0u8; CMSG_BUFSIZE]),
+    };
     msg.msg_control = unsafe { cmsg_buffer.buffer.as_mut_ptr() as _ };
     msg.msg_controllen = unsafe { CMSG_SPACE(size_of::<RawFd>() as _) as _ };