mon/LogMonitor: separate out summary by channel

Instead of keeping the last N entries, keep the last N entries for each
channel.  This ensures that lots of audit records don't age out the
cluster records (or vice versa).

The overall approach does not change, and that approach is overall pretty
lame.  We're still rewriting a big summary blob on every log commit.  We
still should refactor this later.  This solves the immediate problem at a
small cost of increasing the log summary structure by ~2x.

Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/common/LogEntry.cc b/src/common/LogEntry.cc
index 72ec66413ae2aab1c936c86ce6aa0f2eaca177b7..2ad8a5411f9607d7d7a4bddc464f7e80f996f8c6 100644 (file)
--- a/src/common/LogEntry.cc
+++ b/src/common/LogEntry.cc
@@ -263,33 +263,86 @@ void LogEntry::generate_test_instances(list<LogEntry*>& o)
 
 // -----
 
+void LogSummary::build_ordered_tail(list<LogEntry> *tail) const
+{
+  tail->clear();
+  // channel -> (begin, end)
+  map<string,pair<list<pair<uint64_t,LogEntry>>::const_iterator,
+                 list<pair<uint64_t,LogEntry>>::const_iterator>> pos;
+  for (auto& i : tail_by_channel) {
+    pos.emplace(i.first, make_pair(i.second.begin(), i.second.end()));
+  }
+  while (true) {
+    uint64_t min_seq = 0;
+    list<pair<uint64_t,LogEntry>>::const_iterator *minp = 0;
+    for (auto& i : pos) {
+      if (i.second.first == i.second.second) {
+       continue;
+      }
+      if (min_seq == 0 || i.second.first->first < min_seq) {
+       min_seq = i.second.first->first;
+       minp = &i.second.first;
+      }
+    }
+    if (min_seq == 0) {
+      break; // done
+    }
+    tail->push_back((*minp)->second);
+    ++(*minp);
+  }
+}
+
 void LogSummary::encode(bufferlist& bl, uint64_t features) const
 {
-  ENCODE_START(2, 2, bl);
+  if (!HAVE_FEATURE(features, SERVER_MIMIC)) {
+    ENCODE_START(2, 2, bl);
+    encode(version, bl);
+    list<LogEntry> tail;
+    build_ordered_tail(&tail);
+    encode(tail, bl, features);
+    ENCODE_FINISH(bl);
+    return;
+  }
+  ENCODE_START(3, 3, bl);
   encode(version, bl);
-  encode(tail, bl, features);
+  encode(seq, bl);
+  encode(tail_by_channel, bl, features);
   ENCODE_FINISH(bl);
 }
 
 void LogSummary::decode(bufferlist::iterator& bl)
 {
-  DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, bl);
+  DECODE_START_LEGACY_COMPAT_LEN(3, 2, 2, bl);
   decode(version, bl);
-  decode(tail, bl);
+  if (struct_v < 3) {
+    list<LogEntry> tail;
+    decode(tail, bl);
+    for (auto& i : tail) {
+      add(i);
+    }
+  } else {
+    decode(seq, bl);
+    decode(tail_by_channel, bl);
+  }
   DECODE_FINISH(bl);
   keys.clear();
-  for (auto& p : tail) {
-    keys.insert(p.key());
+  for (auto& i : tail_by_channel) {
+    for (auto& e : i.second) {
+      keys.insert(e.second.key());
+    }
   }
 }
 
 void LogSummary::dump(Formatter *f) const
 {
   f->dump_unsigned("version", version);
-  f->open_array_section("tail");
-  for (list<LogEntry>::const_iterator p = tail.begin(); p != tail.end(); ++p) {
-    f->open_object_section("entry");
-    p->dump(f);
+  f->open_object_section("tail_by_channel");
+  for (auto& i : tail_by_channel) {
+    f->open_object_section(i.first.c_str());
+    for (auto& j : i.second) {
+      string s = stringify(j.first);
+      f->dump_object(s.c_str(), j.second);
+    }
     f->close_section();
   }
   f->close_section();

diff --git a/src/common/LogEntry.h b/src/common/LogEntry.h
index a25f9c38150d7a8fde1ca84449d94f2bf5681262..1c405d89fecf8738a143186bafb177d30e997329 100644 (file)
--- a/src/common/LogEntry.h
+++ b/src/common/LogEntry.h
@@ -121,19 +121,25 @@ WRITE_CLASS_ENCODER_FEATURES(LogEntry)
 
 struct LogSummary {
   version_t version;
-  list<LogEntry> tail;
+  // channel -> [(seq#, entry), ...]
+  map<string,list<pair<uint64_t,LogEntry>>> tail_by_channel;
+  uint64_t seq = 0;
   ceph::unordered_set<LogEntryKey> keys;
 
   LogSummary() : version(0) {}
 
+  void build_ordered_tail(list<LogEntry> *tail) const;
+
   void add(const LogEntry& e) {
-    tail.push_back(e);
-    keys.insert(tail.back().key());
+    keys.insert(e.key());
+    tail_by_channel[e.channel].push_back(make_pair(++seq, e));
   }
   void prune(size_t max) {
-    while (tail.size() > max) {
-      keys.erase(tail.front().key());
-      tail.pop_front();
+    for (auto& i : tail_by_channel) {
+      while (i.second.size() > max) {
+       keys.erase(i.second.front().second.key());
+       i.second.pop_front();
+      }
     }
   }
   bool contains(const LogEntryKey& k) const {

diff --git a/src/mon/LogMonitor.cc b/src/mon/LogMonitor.cc
index bfcd2a42afa207c459f7976ed45a87c9a547611a..b460c521223c27be8ed0ad8a118bc6dd01a6aac5 100644 (file)
--- a/src/mon/LogMonitor.cc
+++ b/src/mon/LogMonitor.cc
@@ -428,29 +428,57 @@ bool LogMonitor::preprocess_command(MonOpRequestRef op)
 
     // We'll apply this twice, once while counting out lines
     // and once while outputting them.
-    auto match = [level, channel](const LogEntry &entry) {
-      return entry.prio >= level && (entry.channel == channel || channel == "*");
+    auto match = [level](const LogEntry &entry) {
+      return entry.prio >= level;
     };
 
-    auto rp = summary.tail.rbegin();
-    for (; num > 0 && rp != summary.tail.rend(); ++rp) {
-      if (match(*rp)) {
-        num--;
-      }
-    }
-    if (rp == summary.tail.rend()) {
-      --rp;
-    }
     ostringstream ss;
-    for (; rp != summary.tail.rbegin(); --rp) {
-      if (!match(*rp)) {
-        continue;
+    if (channel == "*") {
+      list<LogEntry> full_tail;
+      summary.build_ordered_tail(&full_tail);
+      derr << "full " << full_tail << dendl;
+      auto rp = full_tail.rbegin();
+      for (; num > 0 && rp != full_tail.rend(); ++rp) {
+       if (match(*rp)) {
+         num--;
+       }
       }
-
-      if (f) {
-       f->dump_object("entry", *rp);
-      } else {
-       ss << *rp << "\n";
+      if (rp == full_tail.rend()) {
+       --rp;
+      }
+      for (; rp != full_tail.rbegin(); --rp) {
+       if (!match(*rp)) {
+         continue;
+       }
+       if (f) {
+         f->dump_object("entry", *rp);
+       } else {
+         ss << *rp << "\n";
+       }
+      }
+    } else {
+      derr << "bar" << dendl;
+      auto p = summary.tail_by_channel.find(channel);
+      if (p != summary.tail_by_channel.end()) {
+       auto rp = p->second.rbegin();
+       for (; num > 0 && rp != p->second.rend(); ++rp) {
+         if (match(rp->second)) {
+           num--;
+         }
+       }
+       if (rp == p->second.rend()) {
+         --rp;
+       }
+       for (; rp != p->second.rbegin(); --rp) {
+         if (!match(rp->second)) {
+           continue;
+         }
+         if (f) {
+           f->dump_object("entry", rp->second);
+         } else {
+           ss << rp->second << "\n";
+         }
+       }
       }
     }
     if (f) {