refactor logic of last matched line

core/reader/JsonLogFileReader.cpp

@@ -19,8 +19,10 @@
 using namespace std;
 
 namespace logtail {
-int32_t
-JsonLogFileReader::LastMatchedLine(char* buffer, int32_t size, int32_t& rollbackLineFeedCount, bool allowRollback) {
+int32_t JsonLogFileReader::RemoveLastIncompleteLog(char* buffer,
+                                                   int32_t size,
+                                                   int32_t& rollbackLineFeedCount,
+                                                   bool allowRollback) {
     int32_t readBytes = 0;
     int32_t endIdx = 0;
     int32_t beginIdx = 0;

core/reader/JsonLogFileReader.h

@@ -30,8 +30,10 @@ class JsonLogFileReader : public LogFileReader {
         : LogFileReader(hostLogPathDir, hostLogPathFile, devInode, readerConfig, multilineConfig) {}
 
 protected:
-    int32_t
-    LastMatchedLine(char* buffer, int32_t size, int32_t& rollbackLineFeedCount, bool allowRollback = true) override;
+    int32_t RemoveLastIncompleteLog(char* buffer,
+                                    int32_t size,
+                                    int32_t& rollbackLineFeedCount,
+                                    bool allowRollback = true) override;
 
 private:
     bool FindJsonMatch(
@@ -40,7 +42,7 @@ class JsonLogFileReader : public LogFileReader {
 #ifdef APSARA_UNIT_TEST_MAIN
     friend class JsonLogFileReaderUnittest;
     friend class JsonParseLogLineUnittest;
-    friend class LastMatchedLineUnittest;
+    friend class RemoveLastIncompleteLogUnittest;
 #endif
 };
 

core/reader/LogFileReader.cpp

@@ -1698,15 +1698,15 @@ void LogFileReader::ReadUTF8(LogBuffer& logBuffer, int64_t end, bool& moreData,
         }
         if (allowRollback || mReaderConfig.second->RequiringJsonReader()) {
             int32_t rollbackLineFeedCount;
-            nbytes = LastMatchedLine(stringBuffer, alignedBytes, rollbackLineFeedCount, allowRollback);
+            nbytes = RemoveLastIncompleteLog(stringBuffer, alignedBytes, rollbackLineFeedCount, allowRollback);
         }
 
         if (nbytes == 0) {
             if (moreData) { // excessively long line without '\n' or multiline begin or valid wchar
                 nbytes = alignedBytes ? alignedBytes : BUFFER_SIZE;
                 if (mReaderConfig.second->RequiringJsonReader()) {
                     int32_t rollbackLineFeedCount;
-                    nbytes = LastMatchedLine(stringBuffer, nbytes, rollbackLineFeedCount, false);
+                    nbytes = RemoveLastIncompleteLog(stringBuffer, nbytes, rollbackLineFeedCount, false);
                 }
                 LOG_WARNING(sLogger,
                             ("Log is too long and forced to be split at offset: ",
@@ -1856,15 +1856,15 @@ void LogFileReader::ReadGBK(LogBuffer& logBuffer, int64_t end, bool& moreData, b
     int32_t rollbackLineFeedCount = 0;
     int32_t bakResultCharCount = resultCharCount;
     if (allowRollback || mReaderConfig.second->RequiringJsonReader()) {
-        resultCharCount = LastMatchedLine(stringBuffer, resultCharCount, rollbackLineFeedCount, allowRollback);
+        resultCharCount = RemoveLastIncompleteLog(stringBuffer, resultCharCount, rollbackLineFeedCount, allowRollback);
     }
     if (resultCharCount == 0) {
         if (moreData) {
             resultCharCount = bakResultCharCount;
             rollbackLineFeedCount = 0;
             if (mReaderConfig.second->RequiringJsonReader()) {
                 int32_t rollbackLineFeedCount;
-                LastMatchedLine(stringBuffer, resultCharCount, rollbackLineFeedCount, false);
+                RemoveLastIncompleteLog(stringBuffer, resultCharCount, rollbackLineFeedCount, false);
             }
             // Cannot get the split position here, so just mark a flag and send alarm later
             logTooLongSplitFlag = true;
@@ -2027,74 +2027,70 @@ LogFileReader::FileCompareResult LogFileReader::CompareToFile(const string& file
         1. xxx\nend\n -> xxx\nend
         1. xxx\nend\nxxx\n -> xxx\nend
 */
-int32_t LogFileReader::LastMatchedLine(char* buffer, int32_t size, int32_t& rollbackLineFeedCount, bool allowRollback) {
+/*
+    return: the number of bytes left
+*/
+int32_t
+LogFileReader::RemoveLastIncompleteLog(char* buffer, int32_t size, int32_t& rollbackLineFeedCount, bool allowRollback) {
     if (!allowRollback) {
         return size;
     }
-    int endPs = size - 1; // buffer[size] = 0 , buffer[size-1] = '\n'
+    int endPs = size - 1;
     rollbackLineFeedCount = 0;
-    // Single line rollback
-    if (!mMultilineConfig.first->IsMultiline()) {
-        while (endPs >= 0) {
-            if (buffer[endPs] == '\n') {
-                if (endPs != size - 1) { // if last line dose not end with '\n', rollback
-                    ++rollbackLineFeedCount;
-                }
-                return endPs + 1;
-            }
-            endPs--;
-        }
-        return 0;
-    }
     // Multiline rollback
-    int begPs = size - 2;
-    std::string exception;
-    while (begPs >= 0) {
-        if (buffer[begPs] == '\n' || begPs == 0) {
-            int lineBegin = begPs == 0 ? 0 : begPs + 1;
-            if (mMultilineConfig.first->GetContinuePatternReg()
-                && BoostRegexMatch(buffer + lineBegin,
-                                   endPs - lineBegin,
-                                   *mMultilineConfig.first->GetContinuePatternReg(),
-                                   exception)) {
-                ++rollbackLineFeedCount;
-                endPs = begPs;
-            } else if (mMultilineConfig.first->GetEndPatternReg()
-                       && BoostRegexMatch(buffer + lineBegin,
-                                          endPs - lineBegin,
-                                          *mMultilineConfig.first->GetEndPatternReg(),
-                                          exception)) {
-                // Ensure the end line is complete
-                if (buffer[endPs] == '\n') {
-                    return endPs + 1;
-                } else {
-                    ++rollbackLineFeedCount;
-                    endPs = begPs;
+    if (mMultilineConfig.first->IsMultiline()) {
+        std::string exception;
+        while (endPs >= 0) {
+            StringView content = GetNextLine(StringView(buffer, size), endPs);
+            if (mMultilineConfig.first->GetEndPatternReg()) {
+                // start + end, continue + end, end
+                if (BoostRegexMatch(
+                        content.data(), content.size(), *mMultilineConfig.first->GetEndPatternReg(), exception)) {
+                    // Ensure the end line is complete
+                    if (buffer[endPs] == '\n') {
+                        return endPs + 1;
+                    }
                 }
             } else if (mMultilineConfig.first->GetStartPatternReg()
-                       && BoostRegexMatch(buffer + lineBegin,
-                                          endPs - lineBegin,
-                                          *mMultilineConfig.first->GetStartPatternReg(),
-                                          exception)) {
+                       && BoostRegexMatch(
+                           content.data(), content.size(), *mMultilineConfig.first->GetStartPatternReg(), exception)) {
+                // start + continue, start
                 ++rollbackLineFeedCount;
                 // Keep all the buffer if rollback all
-                return lineBegin;
-            } else if (mMultilineConfig.first->GetContinuePatternReg()) {
-                // We can confirm the logs before are complete if continue is configured but no regex pattern can match.
-                if (buffer[endPs] == '\n') {
-                    return endPs + 1;
-                } else {
-                    // Keep all the buffer if rollback all
-                    return lineBegin;
-                }
-            } else {
-                ++rollbackLineFeedCount;
-                endPs = begPs;
+                return content.data() - buffer;
             }
+            ++rollbackLineFeedCount;
+            endPs = content.data() - buffer - 1;
         }
-        begPs--;
     }
-    return 0;
+    // Single line rollback or all unmatch rollback
+    rollbackLineFeedCount = 0;
+    StringView content = GetNextLine(StringView(buffer, size), size);
+    size_t rollbackSize = content.data() - buffer;
+    if (rollbackSize < size) {
+        ++rollbackLineFeedCount;
+    }
+    return rollbackSize;
+}
+
+/*
+    params:
+        buffer: all read logs
+        end: the end position of current line
+    return:
+        next line (backward), with \n
+*/
+StringView LogFileReader::GetNextLine(StringView buffer, size_t end) {
+    if (end == 0) {
+        return buffer;
+    }
+
+    for (size_t begin = end; begin > 0; --begin) {
+        if (buffer[begin - 1] == '\n') {
+            return StringView(buffer.data() + begin, end - begin);
+        }
+    }
+    return StringView(buffer.data(), end);
 }
 
 size_t LogFileReader::AlignLastCharacter(char* buffer, size_t size) {
@@ -2108,8 +2104,8 @@ size_t LogFileReader::AlignLastCharacter(char* buffer, size_t size) {
         // 1. The number of byte for one character can be 1, 2, 4.
         // 2. 1 byte character: the top bit is 0.
         // 3. 2 bytes character: the 1st byte is between 0x81 and 0xFE; the 2nd byte is between 0x40 and 0xFE.
-        // 4. 4 bytes character: the 1st and 3rd byte is between 0x81 and 0xFE; the 2nd and 4th byte are between 0x30
-        // and 0x39. (not supported to align)
+        // 4. 4 bytes character: the 1st and 3rd byte is between 0x81 and 0xFE; the 2nd and 4th byte are between
+        // 0x30 and 0x39. (not supported to align)
 
         // 1 byte character, 2nd byte of 2 bytes, 2nd or 4th byte of 4 bytes
         if ((buffer[endPs] & 0x80) == 0 || size == 1) {

core/reader/LogFileReader.h

@@ -128,7 +128,7 @@ class LogFileReader {
     FileCompareResult CompareToFile(const std::string& filePath);
 
     virtual int32_t
-    LastMatchedLine(char* buffer, int32_t size, int32_t& rollbackLineFeedCount, bool allowRollback = true);
+    RemoveLastIncompleteLog(char* buffer, int32_t size, int32_t& rollbackLineFeedCount, bool allowRollback = true);
 
     size_t AlignLastCharacter(char* buffer, size_t size);
 
@@ -502,6 +502,8 @@ class LogFileReader {
     // @param fromCpt: if the read size is recoveried from checkpoint, set it to true.
     size_t getNextReadSize(int64_t fileEnd, bool& fromCpt);
 
+    StringView GetNextLine(StringView buffer, size_t begin);
+
     // Update current checkpoint's read offset and length after success read.
     void setExactlyOnceCheckpointAfterRead(size_t readSize);
 
@@ -587,8 +589,7 @@ class LogFileReader {
     friend class LogSplitUnittest;
     friend class LogSplitDiscardUnmatchUnittest;
     friend class LogSplitNoDiscardUnmatchUnittest;
-    friend class LastMatchedLineDiscardUnmatchUnittest;
-    friend class LastMatchedLineNoDiscardUnmatchUnittest;
+    friend class RemoveLastIncompleteLogMultilineUnittest;
     friend class LogFileReaderCheckpointUnittest;
 
 protected:

core/unittest/reader/CMakeLists.txt

@@ -24,7 +24,7 @@ target_link_libraries(file_reader_options_unittest unittest_base)
 add_executable(json_log_file_reader_unittest JsonLogFileReaderUnittest.cpp)
 target_link_libraries(json_log_file_reader_unittest unittest_base)
 
-add_executable(last_matched_line_unittest LastMatchedLineUnittest.cpp)
+add_executable(last_matched_line_unittest RemoveLastIncompleteLogUnittest.cpp)
 target_link_libraries(last_matched_line_unittest unittest_base)
 
 add_executable(log_file_reader_unittest LogFileReaderUnittest.cpp)