ceph/src/common/Journald.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "Journald.h"
   5
   6 #include <endian.h>
   7 #include <fcntl.h>
   8 #include <iterator>
   9 #include <memory>
  10 #include <string>
  11 #include <sys/mman.h>
  12 #include <sys/socket.h>
  13 #include <sys/uio.h>
  14 #include <sys/un.h>
  15 #include <syslog.h>
  16 #include <unistd.h>
  17 #include <fmt/format.h>
  18 #include <fmt/ostream.h>
  19
  20 #include "include/ceph_assert.h"
  21 #include "common/LogEntry.h"
  22 #include "log/Entry.h"
  23 #include "log/SubsystemMap.h"
  24 #include "msg/msg_fmt.h"
  25
  26
  27 namespace ceph::logging {
  28
  29 namespace {
  30 const struct sockaddr_un sockaddr = {
  31   AF_UNIX,
  32   "/run/systemd/journal/socket",
  33 };
  34
  35 ssize_t sendmsg_fd(int transport_fd, int fd)
  36 {
  37   constexpr size_t control_len = CMSG_LEN(sizeof(int));
  38   char control[control_len];
  39   struct msghdr mh = {
  40     (struct sockaddr*)&sockaddr, // msg_name
  41     sizeof(sockaddr),            // msg_namelen
  42     nullptr,                     // msg_iov
  43     0,                           // msg_iovlen
  44     &control,                    // msg_control
  45     control_len,                 // msg_controllen
  46   };
  47   ceph_assert(transport_fd >= 0);
  48
  49   struct cmsghdr *cmsg = CMSG_FIRSTHDR(&mh);
  50   cmsg->cmsg_level = SOL_SOCKET;
  51   cmsg->cmsg_type = SCM_RIGHTS;
  52   cmsg->cmsg_len = CMSG_LEN(sizeof(int));
  53   *reinterpret_cast<int *>(CMSG_DATA(cmsg)) = fd;
  54
  55   return sendmsg(transport_fd, &mh, MSG_NOSIGNAL);
  56 }
  57
  58 char map_prio(short ceph_prio)
  59 {
  60   if (ceph_prio < 0)
  61     return LOG_ERR;
  62   if (ceph_prio == 0)
  63     return LOG_WARNING;
  64   if (ceph_prio < 5)
  65     return LOG_NOTICE;
  66   if (ceph_prio < 10)
  67     return LOG_INFO;
  68   return LOG_DEBUG;
  69 }
  70 }
  71
  72 namespace detail {
  73 class EntryEncoderBase {
  74  public:
  75   EntryEncoderBase():
  76     m_msg_vec {
  77      {}, {}, {}, { (char *)"\n", 1 },
  78     }
  79   {
  80     std::string id = program_invocation_short_name;
  81     for (auto& c : id) {
  82       if (c == '\n')
  83         c = '_';
  84     }
  85     static_segment = "SYSLOG_IDENTIFIER=" + id + "\n";
  86     m_msg_vec[0].iov_base = static_segment.data();
  87     m_msg_vec[0].iov_len = static_segment.size();
  88   }
  89
  90   constexpr struct iovec *iovec() { return this->m_msg_vec; }
  91   constexpr std::size_t iovec_len()
  92   {
  93     return sizeof(m_msg_vec) / sizeof(m_msg_vec[0]);
  94   }
  95
  96  private:
  97   struct iovec m_msg_vec[4];
  98   std::string static_segment;
  99
 100  protected:
 101   fmt::memory_buffer meta_buf;
 102
 103   struct iovec &meta_vec() { return m_msg_vec[1]; }
 104   struct iovec &msg_vec() { return m_msg_vec[2]; }
 105 };
 106
 107 class EntryEncoder : public EntryEncoderBase {
 108  public:
 109   void encode(const Entry& e, const SubsystemMap *s)
 110   {
 111     meta_buf.clear();
 112     fmt::format_to(std::back_inserter(meta_buf),
 113       R"(PRIORITY={:d}
 114 CEPH_SUBSYS={}
 115 TIMESTAMP={}
 116 CEPH_PRIO={}
 117 THREAD={:016x}
 118 MESSAGE
 119 )",
 120       map_prio(e.m_prio),
 121       s->get_name(e.m_subsys),
 122       e.m_stamp.time_since_epoch().count().count,
 123       e.m_prio,
 124       e.m_thread);
 125
 126     uint64_t msg_len = htole64(e.size());
 127     meta_buf.resize(meta_buf.size() + sizeof(msg_len));
 128     *(reinterpret_cast<uint64_t*>(meta_buf.end()) - 1) = htole64(e.size());
 129
 130     meta_vec().iov_base = meta_buf.data();
 131     meta_vec().iov_len = meta_buf.size();
 132
 133     msg_vec().iov_base = (void *)e.strv().data();
 134     msg_vec().iov_len = e.size();
 135   }
 136 };
 137
 138 class LogEntryEncoder : public EntryEncoderBase {
 139  public:
 140   void encode(const LogEntry& le)
 141   {
 142     meta_buf.clear();
 143     fmt::format_to(std::back_inserter(meta_buf),
 144       R"(PRIORITY={:d}
 145 TIMESTAMP={}
 146 CEPH_NAME={}
 147 CEPH_RANK={}
 148 CEPH_SEQ={}
 149 CEPH_CHANNEL={}
 150 MESSAGE
 151 )",
 152       clog_type_to_syslog_level(le.prio),
 153       le.stamp.to_nsec(),
 154       le.name.to_str(),
 155       le.rank,
 156       le.seq,
 157       le.channel);
 158
 159     uint64_t msg_len = htole64(le.msg.size());
 160     meta_buf.resize(meta_buf.size() + sizeof(msg_len));
 161     *(reinterpret_cast<uint64_t*>(meta_buf.end()) - 1) = htole64(le.msg.size());
 162
 163     meta_vec().iov_base = meta_buf.data();
 164     meta_vec().iov_len = meta_buf.size();
 165
 166     msg_vec().iov_base = (void *)le.msg.data();
 167     msg_vec().iov_len = le.msg.size();
 168   }
 169 };
 170
 171 enum class JournaldClient::MemFileMode {
 172   MEMFD_CREATE,
 173   OPEN_TMPFILE,
 174   OPEN_UNLINK,
 175 };
 176
 177 constexpr const char *mem_file_dir = "/dev/shm";
 178
 179 void JournaldClient::detect_mem_file_mode()
 180 {
 181   int memfd = memfd_create("ceph-journald", MFD_ALLOW_SEALING | MFD_CLOEXEC);
 182   if (memfd >= 0) {
 183     mem_file_mode = MemFileMode::MEMFD_CREATE;
 184     close(memfd);
 185     return;
 186   }
 187   memfd = open(mem_file_dir, O_TMPFILE | O_EXCL | O_CLOEXEC, S_IRUSR | S_IWUSR);
 188   if (memfd >= 0) {
 189     mem_file_mode = MemFileMode::OPEN_TMPFILE;
 190     close(memfd);
 191     return;
 192   }
 193   mem_file_mode = MemFileMode::OPEN_UNLINK;
 194 }
 195
 196 int JournaldClient::open_mem_file()
 197 {
 198   switch (mem_file_mode) {
 199   case MemFileMode::MEMFD_CREATE:
 200     return memfd_create("ceph-journald", MFD_ALLOW_SEALING | MFD_CLOEXEC);
 201   case MemFileMode::OPEN_TMPFILE:
 202     return open(mem_file_dir, O_TMPFILE | O_EXCL | O_CLOEXEC, S_IRUSR | S_IWUSR);
 203   case MemFileMode::OPEN_UNLINK:
 204     char mem_file_template[] = "/dev/shm/ceph-journald-XXXXXX";
 205     int fd = mkostemp(mem_file_template, O_CLOEXEC);
 206     unlink(mem_file_template);
 207     return fd;
 208   }
 209   ceph_abort("Unexpected mem_file_mode");
 210 }
 211
 212 JournaldClient::JournaldClient() :
 213   m_msghdr({
 214     (struct sockaddr*)&sockaddr, // msg_name
 215     sizeof(sockaddr),            // msg_namelen
 216   })
 217 {
 218   fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_CLOEXEC, 0);
 219   ceph_assertf(fd > 0, "socket creation failed: %s", strerror(errno));
 220
 221   int sendbuf = 2 * 1024 * 1024;
 222   setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &sendbuf, sizeof(sendbuf));
 223
 224   detect_mem_file_mode();
 225 }
 226
 227 JournaldClient::~JournaldClient()
 228 {
 229   close(fd);
 230 }
 231
 232 int JournaldClient::send()
 233 {
 234   int ret = sendmsg(fd, &m_msghdr, MSG_NOSIGNAL);
 235   if (ret >= 0)
 236     return 0;
 237
 238   /* Fail silently if the journal is not available */
 239   if (errno == ENOENT)
 240     return -1;
 241
 242   if (errno != EMSGSIZE && errno != ENOBUFS) {
 243     std::cerr << "Failed to send log to journald: " << strerror(errno) << std::endl;
 244     return -1;
 245   }
 246   /* Message doesn't fit... Let's dump the data in a memfd and
 247    * just pass a file descriptor of it to the other side.
 248    */
 249   int buffer_fd = open_mem_file();
 250   if (buffer_fd < 0) {
 251     std::cerr << "Failed to open buffer_fd while sending log to journald: " << strerror(errno) << std::endl;
 252     return -1;
 253   }
 254
 255   ret = writev(buffer_fd, m_msghdr.msg_iov, m_msghdr.msg_iovlen);
 256   if (ret < 0) {
 257     std::cerr << "Failed to write to buffer_fd while sending log to journald: " << strerror(errno) << std::endl;
 258     goto err_close_buffer_fd;
 259   }
 260
 261   if (mem_file_mode == MemFileMode::MEMFD_CREATE) {
 262     ret = fcntl(buffer_fd, F_ADD_SEALS, F_SEAL_SHRINK | F_SEAL_GROW | F_SEAL_WRITE | F_SEAL_SEAL);
 263     if (ret) {
 264       std::cerr << "Failed to seal buffer_fd while sending log to journald: " << strerror(errno) << std::endl;
 265       goto err_close_buffer_fd;
 266     }
 267   }
 268
 269   ret = sendmsg_fd(fd, buffer_fd);
 270   if (ret < 0) {
 271     /* Fail silently if the journal is not available */
 272     if (errno == ENOENT)
 273       goto err_close_buffer_fd;
 274
 275     std::cerr << "Failed to send fd while sending log to journald: " << strerror(errno) << std::endl;
 276     goto err_close_buffer_fd;
 277   }
 278   close(buffer_fd);
 279   return 0;
 280
 281 err_close_buffer_fd:
 282   close(buffer_fd);
 283   return -1;
 284 }
 285
 286 } // namespace ceph::logging::detail
 287
 288 JournaldLogger::JournaldLogger(const SubsystemMap *s) :
 289   m_entry_encoder(std::make_unique<detail::EntryEncoder>()),
 290   m_subs(s)
 291 {
 292   client.m_msghdr.msg_iov = m_entry_encoder->iovec();
 293   client.m_msghdr.msg_iovlen = m_entry_encoder->iovec_len();
 294 }
 295
 296 JournaldLogger::~JournaldLogger() = default;
 297
 298 int JournaldLogger::log_entry(const Entry& e)
 299 {
 300   m_entry_encoder->encode(e, m_subs);
 301   return client.send();
 302 }
 303
 304 JournaldClusterLogger::JournaldClusterLogger() :
 305   m_log_entry_encoder(std::make_unique<detail::LogEntryEncoder>())
 306 {
 307   client.m_msghdr.msg_iov = m_log_entry_encoder->iovec();
 308   client.m_msghdr.msg_iovlen = m_log_entry_encoder->iovec_len();
 309 }
 310
 311 JournaldClusterLogger::~JournaldClusterLogger() = default;
 312
 313 int JournaldClusterLogger::log_log_entry(const LogEntry &le)
 314 {
 315   m_log_entry_encoder->encode(le);
 316   return client.send();
 317 }
 318
 319 }