Allow the LogMessage to be compacted before storing it in disk-buffer

lib/logmsg/logmsg-serialize-fixup.c

@@ -148,8 +148,7 @@ _allocate_handle_for_entry_name(NVHandle old_handle, NVEntry *entry)
 static NVHandle
 _allocate_handle_of_referenced_entry(NVTable *self, NVHandle ref_handle)
 {
-  NVIndexEntry *index_entry;
-  NVEntry *ref_entry = nv_table_get_entry(self, ref_handle, &index_entry);
+  NVEntry *ref_entry = nv_table_get_entry(self, ref_handle, NULL, NULL);
 
   return _allocate_handle_for_entry_name(ref_handle, ref_entry);
 }

lib/logmsg/logmsg-serialize.c

@@ -69,26 +69,36 @@ _serialize_message(LogMessageSerializationState *state)
   serialize_write_uint8(sa, msg->num_sdata);
   serialize_write_uint8(sa, msg->alloc_sdata);
   serialize_write_uint32_array(sa, (guint32 *) msg->sdata, msg->num_sdata);
-  nv_table_serialize(state, msg->payload);
+
+  NVTable *payload;
+
+  if (state->flags & LMSF_COMPACTION)
+    payload = nv_table_compact(msg->payload);
+  else
+    payload = nv_table_ref(msg->payload);
+
+  nv_table_serialize(state, payload);
+  nv_table_unref(payload);
   return TRUE;
 }
 
 gboolean
-log_msg_serialize_with_ts_processed(LogMessage *self, SerializeArchive *sa, const UnixTime *processed)
+log_msg_serialize_with_ts_processed(LogMessage *self, SerializeArchive *sa, const UnixTime *processed, guint32 flags)
 {
   LogMessageSerializationState state = { 0 };
 
   state.version = LGM_V26;
   state.msg = self;
   state.sa = sa;
   state.processed = processed;
+  state.flags = flags;
   return _serialize_message(&state);
 }
 
 gboolean
-log_msg_serialize(LogMessage *self, SerializeArchive *sa)
+log_msg_serialize(LogMessage *self, SerializeArchive *sa, guint32 flags)
 {
-  return log_msg_serialize_with_ts_processed(self, sa, NULL);
+  return log_msg_serialize_with_ts_processed(self, sa, NULL, flags);
 }
 
 static gboolean

lib/logmsg/logmsg-serialize.h

@@ -50,7 +50,7 @@
  *   26      use 32 bit values nvtable
  */
 
-typedef enum _LogMessageVersion
+enum _LogMessageVersion
 {
   LGM_V01 = 1,
   LGM_V10 = 10,
@@ -63,10 +63,16 @@ typedef enum _LogMessageVersion
   LGM_V24 = 24,
   LGM_V25 = 25,
   LGM_V26 = 26
-} LogMessageVersion;
+};
+
+enum _LogMessageSerializationFlags
+{
+  LMSF_COMPACTION = 0x0001,
+};
 
 gboolean log_msg_deserialize(LogMessage *self, SerializeArchive *sa);
-gboolean log_msg_serialize_with_ts_processed(LogMessage *self, SerializeArchive *sa, const UnixTime *processed);
-gboolean log_msg_serialize(LogMessage *self, SerializeArchive *sa);
+gboolean log_msg_serialize_with_ts_processed(LogMessage *self, SerializeArchive *sa, const UnixTime *processed,
+                                             guint32 flags);
+gboolean log_msg_serialize(LogMessage *self, SerializeArchive *sa, guint32 flags);
 
 #endif

lib/logmsg/logmsg.c

@@ -199,6 +199,7 @@ const gchar *builtin_value_names[] =
   "MSGID",
   "SOURCE",
   "LEGACY_MSGHDR",
+  "__RESERVED_LM_V_MAX",
   NULL,
 };
 
@@ -560,6 +561,7 @@ log_msg_set_value(LogMessage *self, NVHandle handle, const gchar *value, gssize
         {
           /* can't grow the payload, it has reached the maximum size */
           msg_info("Cannot store value for this log message, maximum size has been reached",
+                   evt_tag_int("maximum_payload", NV_TABLE_MAX_BYTES),
                    evt_tag_str("name", name),
                    evt_tag_printf("value", "%.32s%s", value, value_len > 32 ? "..." : ""));
           break;
@@ -580,7 +582,24 @@ log_msg_set_value(LogMessage *self, NVHandle handle, const gchar *value, gssize
 void
 log_msg_unset_value(LogMessage *self, NVHandle handle)
 {
-  nv_table_unset_value(self->payload, handle);
+  while (!nv_table_unset_value(self->payload, handle))
+    {
+      /* error allocating string in payload, reallocate */
+      guint32 old_size = self->payload->size;
+      if (!nv_table_realloc(self->payload, &self->payload))
+        {
+          /* can't grow the payload, it has reached the maximum size */
+          const gchar *name = log_msg_get_value_name(handle, NULL);
+          msg_info("Cannot unset value for this log message, maximum size has been reached",
+                   evt_tag_int("maximum_payload", NV_TABLE_MAX_BYTES),
+                   evt_tag_str("name", name));
+          break;
+        }
+      guint32 new_size = self->payload->size;
+      self->allocated_bytes += (new_size - old_size);
+      stats_counter_add(count_allocated_bytes, new_size-old_size);
+      stats_counter_inc(count_payload_reallocs);
+    }
 
   if (_value_invalidates_legacy_header(handle))
     log_msg_unset_value(self, LM_V_LEGACY_MSGHDR);