Merge pull request #889 from bazsi/log-msg-memory-accounting-fixes

Log msg memory accounting fixes

Author: sodomelle

lib/logmsg/logmsg-serialize.c

@@ -511,6 +511,7 @@ gboolean
 log_msg_deserialize(LogMessage *self, SerializeArchive *sa)
 {
   LogMessageSerializationState state = { 0 };
+  gboolean result = FALSE;
 
   state.sa = sa;
   state.msg = self;
@@ -520,7 +521,15 @@ log_msg_deserialize(LogMessage *self, SerializeArchive *sa)
     }
 
   if (state.version < LGM_V20)
-    return _deserialize_message_version_1x(&state);
-
-  return _deserialize_message_version_2x(&state);
+    result = _deserialize_message_version_1x(&state);
+  else
+    result = _deserialize_message_version_2x(&state);
+
+  self->allocated_bytes = sizeof(LogMessage) +
+     self->alloc_sdata * sizeof(self->sdata[0]) +
+     g_sockaddr_len(self->saddr) + g_sockaddr_len(self->daddr) +
+     (sizeof(self->tags[0]) * self->num_tags) +
+     (sizeof(self->nodes[0]) * self->num_nodes) +
+     (self->payload ? nv_table_get_memory_consumption(self->payload) : 0);
+  return result;
 }

lib/logmsg/logmsg.c

@@ -278,6 +278,13 @@ static StatsCounterItem *count_sdata_updates;
 static StatsCounterItem *count_allocated_bytes;
 static GPrivate priv_macro_value = G_PRIVATE_INIT(__free_macro_value);
 
+static inline void
+log_msg_update_allocation(LogMessage *self, gssize delta)
+{
+  self->allocated_bytes += delta;
+  stats_counter_add(count_allocated_bytes, delta);
+}
+
 static void
 log_msg_update_sdata_slow(LogMessage *self, NVHandle handle, const gchar *name, gssize name_len)
 {
@@ -327,8 +334,7 @@ log_msg_update_sdata_slow(LogMessage *self, NVHandle handle, const gchar *name,
   self->alloc_sdata = alloc_sdata;
   if (self->sdata)
     {
-      self->allocated_bytes += ((self->alloc_sdata - old_alloc_sdata) * sizeof(self->sdata[0]));
-      stats_counter_add(count_allocated_bytes, (self->alloc_sdata - old_alloc_sdata) * sizeof(self->sdata[0]));
+      log_msg_update_allocation(self, ((self->alloc_sdata - old_alloc_sdata) * sizeof(self->sdata[0])));
     }
   /* ok, we have our own SDATA array now which has at least one free slot */
 
@@ -577,8 +583,7 @@ log_msg_set_value_with_type(LogMessage *self, NVHandle handle,
     {
       self->payload = nv_table_clone(self->payload, name_len + value_len + 2);
       log_msg_set_flag(self, LF_STATE_OWN_PAYLOAD);
-      self->allocated_bytes += self->payload->size;
-      stats_counter_add(count_allocated_bytes, self->payload->size);
+      log_msg_update_allocation(self, self->payload->size);
     }
 
   /* we need a loop here as a single realloc may not be enough. Might help
@@ -598,8 +603,7 @@ log_msg_set_value_with_type(LogMessage *self, NVHandle handle,
           break;
         }
       guint32 new_size = self->payload->size;
-      self->allocated_bytes += (new_size - old_size);
-      stats_counter_add(count_allocated_bytes, new_size-old_size);
+      log_msg_update_allocation(self, (new_size - old_size));
       stats_counter_inc(count_payload_reallocs);
     }
 
@@ -650,8 +654,7 @@ log_msg_unset_value(LogMessage *self, NVHandle handle)
           break;
         }
       guint32 new_size = self->payload->size;
-      self->allocated_bytes += (new_size - old_size);
-      stats_counter_add(count_allocated_bytes, new_size-old_size);
+      log_msg_update_allocation(self, (new_size - old_size));
       stats_counter_inc(count_payload_reallocs);
     }
 
@@ -932,6 +935,7 @@ log_msg_set_tag_by_id_onoff(LogMessage *self, LogTagId id, gboolean on)
   if (!log_msg_chk_flag(self, LF_STATE_OWN_TAGS) && self->num_tags)
     {
       self->tags = g_memdup2(self->tags, sizeof(self->tags[0]) * self->num_tags);
+      log_msg_update_allocation(self, sizeof(self->tags[0]) * self->num_tags);
     }
   log_msg_set_flag(self, LF_STATE_OWN_TAGS);
 
@@ -961,6 +965,7 @@ log_msg_set_tag_by_id_onoff(LogMessage *self, LogTagId id, gboolean on)
             self->tags = g_realloc(self->tags, sizeof(self->tags[0]) * self->num_tags);
           else
             self->tags = g_malloc(sizeof(self->tags[0]) * self->num_tags);
+          log_msg_update_allocation(self, (self->num_tags - old_num_tags) * sizeof(self->tags[0]));
           memset(&self->tags[old_num_tags], 0, (self->num_tags - old_num_tags) * sizeof(self->tags[0]));
 
           if (inline_tags)
@@ -1277,9 +1282,14 @@ log_msg_set_saddr(LogMessage *self, GSockAddr *saddr)
 void
 log_msg_set_saddr_ref(LogMessage *self, GSockAddr *saddr)
 {
+  gssize old_len = 0;
   if (log_msg_chk_flag(self, LF_STATE_OWN_SADDR))
-    g_sockaddr_unref(self->saddr);
+    {
+      old_len = g_sockaddr_len(self->saddr);
+      g_sockaddr_unref(self->saddr);
+    }
   self->saddr = saddr;
+  log_msg_update_allocation(self, (gssize) g_sockaddr_len(self->saddr) - old_len);
   self->flags |= LF_STATE_OWN_SADDR;
 }
 
@@ -1292,9 +1302,14 @@ log_msg_set_daddr(LogMessage *self, GSockAddr *daddr)
 void
 log_msg_set_daddr_ref(LogMessage *self, GSockAddr *daddr)
 {
+  gssize old_len = 0;
   if (log_msg_chk_flag(self,  LF_STATE_OWN_DADDR))
-    g_sockaddr_unref(self->daddr);
+    {
+      old_len = g_sockaddr_len(self->daddr);
+      g_sockaddr_unref(self->daddr);
+    }
   self->daddr = daddr;
+  log_msg_update_allocation(self, (gssize) g_sockaddr_len(self->daddr) - old_len);
   self->flags |= LF_STATE_OWN_DADDR;
 }
 
@@ -1392,8 +1407,7 @@ log_msg_alloc(gsize payload_size)
     msg->payload = nv_table_init_borrowed(((gchar *) msg) + payload_ofs, payload_space, LM_V_MAX);
 
   msg->num_nodes = nodes;
-  msg->allocated_bytes = alloc_size + payload_space;
-  stats_counter_add(count_allocated_bytes, msg->allocated_bytes);
+  log_msg_update_allocation(msg, alloc_size);
   return msg;
 }
 
@@ -1444,9 +1458,10 @@ log_msg_alloc_clone(LogMessage *original)
   msg = g_malloc(alloc_size);
 
   memcpy(msg, original, sizeof(*msg));
+  msg->allocated_bytes = 0;
   msg->num_nodes = nodes;
-  msg->allocated_bytes = alloc_size;
-  stats_counter_add(count_allocated_bytes, msg->allocated_bytes);
+  log_msg_update_allocation(msg, alloc_size);
+
   return msg;
 }
 
@@ -2110,20 +2125,6 @@ log_msg_lookup_time_stamp_name(const gchar *name)
   return -1;
 }
 
-gssize
-log_msg_get_size(LogMessage *self)
-{
-  if (!self)
-    return 0;
-
-  return
-    sizeof(LogMessage) + // msg.static fields
-    + self->alloc_sdata * sizeof(self->sdata[0]) +
-    g_sockaddr_len(self->saddr) + g_sockaddr_len(self->daddr) +
-    ((self->num_tags) ? sizeof(self->tags[0]) * self->num_tags : 0) +
-    nv_table_get_memory_consumption(self->payload); // msg.payload (nvtable)
-}
-
 #ifdef __linux__
 
 const gchar *

lib/logmsg/logmsg.h

@@ -638,7 +638,11 @@ void log_msg_registry_foreach(GHFunc func, gpointer user_data);
 
 gint log_msg_lookup_time_stamp_name(const gchar *name);
 
-gssize log_msg_get_size(LogMessage *self);
+static inline gsize
+log_msg_get_size(LogMessage *self)
+{
+  return self->allocated_bytes;
+}
 
 #define evt_tag_msg_reference(msg)             \
     evt_tag_printf("msg", "%p", (msg)),        \
@@ -661,4 +665,6 @@ evt_tag_msg_value_name(const gchar *name, NVHandle value_handle)
   return evt_tag_str(name, value_name);
 }
 
+extern gint logmsg_queue_node_max;
+
 #endif

lib/logmsg/nvtable.h

@@ -463,9 +463,7 @@ nv_table_get_ofs_for_an_entry(NVTable *self, NVEntry *entry)
 static inline gssize
 nv_table_get_memory_consumption(NVTable *self)
 {
-  return sizeof(*self)+
-         self->num_static_entries*sizeof(self->static_entries[0])+
-         self->used;
+  return self->size;
 }
 
 #endif

lib/logmsg/tests/test_log_message.c

@@ -768,103 +768,6 @@ Test(log_message, test_clear_matches_call_resets_all_matches_to_unset)
   log_msg_unref(msg);
 }
 
-#define DEFUN_KEY_VALUE(name, key, value, size) \
-  gchar name ## _key[size]; \
-  gchar name ## _value[size]; \
-  name ## _key[size-1] = name ## _value[size-1] = 0; \
-  memset(name ## _key, key, sizeof(name ##_key)-1); \
-  memset(name ## _value, value, sizeof(name ##_value)-1); \
-
-
-typedef struct
-{
-  gssize nvtable_size_old;
-  gssize nvtable_size_new;
-  gssize msg_size_old;
-  gssize msg_size_new;
-} sizes_t;
-
-static sizes_t
-add_key_value(LogMessage *msg, gchar *key, gchar *value)
-{
-  sizes_t sizes;
-  sizes.msg_size_old = log_msg_get_size(msg);
-  sizes.nvtable_size_old = nv_table_get_memory_consumption(msg->payload);
-  log_msg_set_value_by_name(msg, key, value, -1);
-  sizes.nvtable_size_new = nv_table_get_memory_consumption(msg->payload);
-  sizes.msg_size_new = log_msg_get_size(msg);
-  return sizes;
-}
-
-
-static void
-test_with_sdata(LogMessage *msg, guint32 old_msg_size)
-{
-  sizes_t sizes;
-  gchar key[]          = ".SDATA.**";
-  gchar value[] = "AAAAAAA";
-
-  guint32 single_sdata_kv_size;
-  guint32 sdata_payload_array_size;
-
-  const char iter_length = 17;
-
-  for (char i = 0; i < iter_length; i++)
-    {
-      g_snprintf(key, sizeof(key), ".SDATA.%02d", i);
-      sizes = add_key_value(msg, key, value);
-
-      single_sdata_kv_size = NV_ENTRY_DIRECT_HDR + NV_TABLE_BOUND(strlen(key)+1 + strlen(value)+1);
-
-      /* i+1 is stored, but the sdata array size is calculated when adding the i-th */
-      sdata_payload_array_size = STRICT_ROUND_TO_NEXT_EIGHT(i) * sizeof(msg->sdata[0]);
-      cr_assert_eq(old_msg_size + (i+1) * single_sdata_kv_size + sdata_payload_array_size, sizes.msg_size_new);
-    }
-}
-
-#define SMALL_LENGTH 10
-#define LARGE_LENGTH 256
-
-Test(log_message, test_message_size)
-{
-  LogMessage *msg = log_msg_new_empty();
-
-  sizes_t sizes;
-
-  DEFUN_KEY_VALUE(small, 'C', 'D', SMALL_LENGTH);
-  sizes = add_key_value(msg, small_key, small_value);
-  // (SMALL_LENGTH-1)*'C'+'\0' + (SMALL_LENGTH-1)*'D'+'\0'
-  guint32 entry_size = NV_ENTRY_DIRECT_HDR + NV_TABLE_BOUND(SMALL_LENGTH + SMALL_LENGTH);
-  cr_assert_eq(sizes.nvtable_size_old + entry_size, sizes.nvtable_size_new);
-  cr_assert_eq(sizes.msg_size_old + entry_size, sizes.msg_size_new);  // Size increased because of nvtable
-
-  guint32 msg_size = sizes.msg_size_new;
-  log_msg_set_tag_by_name(msg, "test_tag_storage");
-  cr_assert(log_msg_is_tag_by_name(msg, "test_tag_storage"));
-  cr_assert_eq(msg_size, log_msg_get_size(msg)); // Tag is not increased until tag id 65
-
-  char *tag_name = strdup("00tagname");
-  // (*8 to convert to bits) + no need plus 1 bcause we already added one tag: test_tag_storage
-  for (int i = 0; i < GLIB_SIZEOF_LONG*8; i++)
-    {
-      sprintf(tag_name, "%dtagname", i);
-      log_msg_set_tag_by_name(msg, tag_name);
-    }
-  free(tag_name);
-  cr_assert_eq(msg_size + 2*GLIB_SIZEOF_LONG, log_msg_get_size(msg));
-
-
-  DEFUN_KEY_VALUE(big, 'A', 'B', LARGE_LENGTH);
-  sizes = add_key_value(msg, big_key, big_value); // nvtable is expanded
-  entry_size = NV_ENTRY_DIRECT_HDR + NV_TABLE_BOUND(LARGE_LENGTH + LARGE_LENGTH);
-  cr_assert_eq(sizes.nvtable_size_old + entry_size, sizes.nvtable_size_new); // but only increased by the entry
-  cr_assert_eq(sizes.msg_size_old + entry_size, sizes.msg_size_new);  // nvtable is doubled
-
-  test_with_sdata(msg, sizes.msg_size_new);
-
-  log_msg_unref(msg);
-}
-
 Test(log_message, when_get_indirect_value_with_null_value_len_abort_instead_of_sigsegv, .signal=SIGABRT)
 {
   LogMessageTestParams *params = log_message_test_params_new();

modules/diskq/tests/test_diskq.c

@@ -579,6 +579,11 @@ setup(void)
   configuration->stats_options.level = 1;
 
   cr_assert(cfg_init(configuration));
+
+  /* we bump the number so our tests do not increase it as we send/receive
+   * messages.  This makes the size of a LogMessage stable, thus we can
+   * check stats counters */
+  logmsg_queue_node_max = 10;
 }
 
 static void

news/bugfix-889.md

@@ -0,0 +1,4 @@
+Make message memory usage metrics more accurate: AxoSyslog keeps track of
+memory usage by messages both globally and on a per queue basis. The
+accounting behind those metrics were inaccurate, the value shown being
+smaller than the actual memory use. These inaccuracies were fixed.

tests/light/functional_tests/destination_drivers/network_destination/disk_buffer/helper_functions.py

@@ -154,7 +154,7 @@ def set_expected_metrics_state_when_sending_more_logs_than_buffer_can_handle_wit
             "syslogng_disk_queue_disk_allocated_bytes",
             "syslogng_disk_queue_disk_usage_bytes",
             "syslogng_disk_queue_events",
-            "syslogng_disk_queue_memory_usage_bytes",
+            # "syslogng_disk_queue_memory_usage_bytes",
             "messages_in_disk_buffer",
         ],
     )
@@ -173,7 +173,7 @@ def set_expected_metrics_state_when_sending_more_logs_than_buffer_can_handle_wit
             syslogng_disk_queue_disk_allocated_bytes=1048576,
             syslogng_disk_queue_disk_usage_bytes=1044480,
             syslogng_disk_queue_events=frontCache_size + qdisk_size,
-            syslogng_disk_queue_memory_usage_bytes=764000,
+            # syslogng_disk_queue_memory_usage_bytes=764000,
             messages_in_disk_buffer=qdisk_size,
         ),
         after=BufferState(
@@ -184,7 +184,7 @@ def set_expected_metrics_state_when_sending_more_logs_than_buffer_can_handle_wit
             syslogng_disk_queue_disk_allocated_bytes=1048576,
             syslogng_disk_queue_disk_usage_bytes=0,
             syslogng_disk_queue_events=0,
-            syslogng_disk_queue_memory_usage_bytes=0,
+            # syslogng_disk_queue_memory_usage_bytes=0,
             messages_in_disk_buffer=0,
         ),
     )