[fix](be) Fix file descriptor leak when resuming log files (#68926)

### What problem does this PR solve?

Issue Number: N/A

Related PR: #3649 (the existing-log resume path)

Problem Summary:

When a BE restarts with `log_split_method=size` and existing
INFO/WARNING logs, Doris's glog patch opens each resumed file with
`open()` and then opens it again with `fopen()`. The first descriptor
has no owner on the success path. Rotation and shutdown close only the
`FILE*`; once the file quota unlinks the old log, the leaked descriptor
keeps its disk blocks allocated until the BE exits.

Use `fdopen(fd, "a")` in both the glog 0.4.0 and 0.6.0 patches so the
stream owns the existing append-only, close-on-exec descriptor.
Initialize the saved file length only after `fdopen()` succeeds, and
close the descriptor on failure before using the existing new-file path.
Skip resume when the file list is empty, including when a quota of one
removes the last existing file.

Add a standalone regression test against the actual static glog library
and run it after the Linux third-party build. It covers INFO and WARNING
logs, append behavior, rotation using the saved file length, repeated
quota deletion, shutdown, empty directories, quota one, and injected
`fdopen()` failure.

### Release note

Fix leaked file descriptors that could retain disk space for deleted BE
logs after restarting with size-based log splitting.

### Check List (For Author)

- Test
    - [ ] Regression test
    - [ ] Unit Test
- [x] Manual test (standalone third-party library regression; details
below)
    - [ ] No need to test or manual test.
- Behavior changed:
    - [ ] No.
- [x] Yes. Resumed log descriptors are released during rotation and
shutdown; an empty resume list and failed `fdopen()` use the normal
new-file path.
- Does this need documentation?
    - [x] No.
    - [ ] Yes.

Validation:

- Applied both updated patches to clean upstream source archives with
`patch --dry-run --fuzz=0`, then rebuilt and installed the static
libraries in isolated directories (0.4.0 using autotools, 0.6.0 using
CMake; optional gflags and libunwind disabled).
- Ran `TP_INSTALL_DIR=<rebuilt-glog-install> CXX=/usr/bin/c++ bash
thirdparty/test/glog-logfile-resume-test.sh` for each version: all four
scenarios passed on each version. Two active log descriptors, zero
descriptors for deleted logs through quota rotations, and zero log
descriptors after shutdown.
- The same test against the existing unmodified Doris glog failed with
`initial write: held=4 expected=2`, confirming that it detects the
original leak.
- `git diff --check`, Bash syntax, ShellCheck 0.8.0, shfmt 3.5.1,
embedded C++ formatting with clang-format 16, and workflow YAML
validation passed.
- A full BE build, BE unit tests, and cluster regression were not run;
validation directly rebuilt and exercised the patched dependency.

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label

.github/workflows/build-thirdparty.yml

@@ -201,6 +201,9 @@ jobs:
           cd "${GITHUB_WORKSPACE}"
           thirdparty/test/libunwind-page-size-test.sh
 
+          # Check descriptor ownership when resuming and rotating existing logs.
+          thirdparty/test/glog-logfile-resume-test.sh
+
   build_macos:
     name: Build Third Party Libraries (macOS)
     needs: changes

regression-test/pipeline/nonConcurrent/conf/fe.conf

@@ -30,7 +30,7 @@ LOG_DIR = ${DORIS_HOME}/log
 JAVA_OPTS="-Djavax.security.auth.useSubjectCredsOnly=false -Xss4m -Xmx4096m -XX:+HeapDumpOnOutOfMemoryError -XX:+UnlockExperimentalVMOptions -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:+PrintGCDateStamps -XX:+PrintGCDetails -XX:+PrintClassHistogramAfterFullGC -Xloggc:$DORIS_HOME/log/fe.gc.log.$CUR_DATE -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=50M -Dlog4j2.formatMsgNoLookups=true -Dcom.mysql.cj.disableAbandonedConnectionCleanup=true"
 
 # For jdk 17, this JAVA_OPTS will be used as default JVM options
-JAVA_OPTS_FOR_JDK_17="-Djavax.security.auth.useSubjectCredsOnly=false -Xmx8192m -Xms8192m -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$DORIS_HOME/log/ -Xlog:gc*,classhisto*=trace:$DORIS_HOME/log/fe.gc.log.$CUR_DATE:time,uptime:filecount=10,filesize=50M -Dcom.mysql.cj.disableAbandonedConnectionCleanup=true --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED"
+JAVA_OPTS_FOR_JDK_17="-Djavax.security.auth.useSubjectCredsOnly=false -Xmx8192m -Xms8192m -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$DORIS_HOME/log/ -Xlog:gc*,classhisto*=trace:$DORIS_HOME/log/fe.gc.log.$CUR_DATE:time,uptime:filecount=10,filesize=50M -Dcom.mysql.cj.disableAbandonedConnectionCleanup=true -Darrow.adbc.driver.jni.library.path=${DORIS_HOME}/lib --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/jdk.internal.ref=ALL-UNNAMED"
 
 sys_log_level = INFO
 sys_log_mode = NORMAL

thirdparty/patches/glog-0.4.0.patch

@@ -194,7 +194,7 @@ diff -uprN a/src/logging.cc b/src/logging.cc
  void LogFileObject::Write(bool force_flush,
                            time_t timestamp,
                            const char* message,
-@@ -972,14 +1055,55 @@ void LogFileObject::Write(bool force_flu
+@@ -972,14 +1055,58 @@ void LogFileObject::Write(bool force_flu
      return;
    }
  
@@ -232,27 +232,30 @@ diff -uprN a/src/logging.cc b/src/logging.cc
  
 +  if ((file_ == NULL) && (!inited_) && (FLAGS_log_split_method == "size")) {
 +    CheckFileNumQuota();
-+    const char* filename = file_list_.back().name.c_str();
-+    int fd = open(filename, O_WRONLY | O_CREAT | O_APPEND, FLAGS_logfile_mode);
-+    if (fd != -1) {
++    if (!file_list_.empty()) {
++      const char* filename = file_list_.back().name.c_str();
++      int fd = open(filename, O_WRONLY | O_CREAT | O_APPEND, FLAGS_logfile_mode);
++      if (fd != -1) {
 +#ifdef HAVE_FCNTL
-+      // Mark the file close-on-exec. We don't really care if this fails
-+      fcntl(fd, F_SETFD, FD_CLOEXEC);
++        // Mark the file close-on-exec. We don't really care if this fails
++        fcntl(fd, F_SETFD, FD_CLOEXEC);
 +#endif
-+      file_ = fopen(filename, "a+"); // Read and append a FILE*.
-+      if (file_ == NULL) {      // Man, we're screwed!, try to create new log file
-+        close(fd);
++        file_ = fdopen(fd, "a");
++        if (file_ == NULL) {
++          close(fd);
++        } else {
++          fseek(file_, 0, SEEK_END);
++          file_length_ = bytes_since_flush_ = ftell(file_);
++          inited_ = true;
++        }
 +      }
-+      fseek(file_, 0, SEEK_END);
-+      file_length_ = bytes_since_flush_ = ftell(file_);
-+      inited_ = true;
 +    }
 +  }
 +
    // If there's no destination file, make one before outputting
    if (file_ == NULL) {
      // Try to rollover the log file every 32 log messages.  The only time
-@@ -988,7 +1112,16 @@ void LogFileObject::Write(bool force_flu
+@@ -988,7 +1115,16 @@ void LogFileObject::Write(bool force_flu
      if (++rollover_attempt_ != kRolloverAttemptFrequency) return;
      rollover_attempt_ = 0;
  
@@ -270,7 +273,7 @@ diff -uprN a/src/logging.cc b/src/logging.cc
      localtime_r(&timestamp, &tm_time);
  
      // The logfile's filename will have the date/time & pid in it
-@@ -996,13 +1129,19 @@ void LogFileObject::Write(bool force_flu
+@@ -996,13 +1132,19 @@ void LogFileObject::Write(bool force_flu
      time_pid_stream.fill('0');
      time_pid_stream << 1900+tm_time.tm_year
                      << setw(2) << 1+tm_time.tm_mon
@@ -297,7 +300,7 @@ diff -uprN a/src/logging.cc b/src/logging.cc
      const string& time_pid_string = time_pid_stream.str();
  
      if (base_filename_selected_) {
-@@ -1035,10 +1174,7 @@ void LogFileObject::Write(bool force_flu
+@@ -1035,10 +1177,7 @@ void LogFileObject::Write(bool force_flu
        // attempt to hold on to the same mutex, and get into a
        // deadlock. Simply use a name like invalid-user.
        if (uidname.empty()) uidname = "invalid-user";
@@ -309,7 +312,7 @@ diff -uprN a/src/logging.cc b/src/logging.cc
        // We're going to (potentially) try to put logs in several different dirs
        const vector<string> & log_dirs = GetLoggingDirectories();
  
-@@ -1062,28 +1198,6 @@ void LogFileObject::Write(bool force_flu
+@@ -1062,28 +1201,6 @@ void LogFileObject::Write(bool force_flu
          return;
        }
      }

thirdparty/patches/glog-0.6.0.patch

@@ -214,7 +214,7 @@ index 4028ccc..e8d9e32 100644
  void LogFileObject::Write(bool force_flush,
                            time_t timestamp,
                            const char* message,
-@@ -1149,13 +1238,55 @@ void LogFileObject::Write(bool force_flush,
+@@ -1149,13 +1238,58 @@ void LogFileObject::Write(bool force_flush,
      return;
    }
  
@@ -251,27 +251,30 @@ index 4028ccc..e8d9e32 100644
  
 +  if ((file_ == NULL) && (!inited_) && (FLAGS_log_split_method == "size")) {
 +    CheckFileNumQuota();
-+    const char* filename = file_list_.back().name.c_str();
-+    int fd = open(filename, O_WRONLY | O_CREAT | O_APPEND, FLAGS_logfile_mode);
-+    if (fd != -1) {
++    if (!file_list_.empty()) {
++      const char* filename = file_list_.back().name.c_str();
++      int fd = open(filename, O_WRONLY | O_CREAT | O_APPEND, FLAGS_logfile_mode);
++      if (fd != -1) {
 +#ifdef HAVE_FCNTL
-+      // Mark the file close-on-exec. We don't really care if this fails
-+      fcntl(fd, F_SETFD, FD_CLOEXEC);
++        // Mark the file close-on-exec. We don't really care if this fails
++        fcntl(fd, F_SETFD, FD_CLOEXEC);
 +#endif
-+      file_ = fopen(filename, "a+"); // Read and append a FILE*.
-+      if (file_ == NULL) {      // Man, we're screwed!, try to create new log file
-+        close(fd);
++        file_ = fdopen(fd, "a");
++        if (file_ == NULL) {
++          close(fd);
++        } else {
++          fseek(file_, 0, SEEK_END);
++          file_length_ = bytes_since_flush_ = ftell(file_);
++          inited_ = true;
++        }
 +      }
-+      fseek(file_, 0, SEEK_END);
-+      file_length_ = bytes_since_flush_ = ftell(file_);
-+      inited_ = true;
 +    }
 +  }
 +
    // If there's no destination file, make one before outputting
    if (file_ == NULL) {
      // Try to rollover the log file every 32 log messages.  The only time
-@@ -1164,7 +1295,19 @@ void LogFileObject::Write(bool force_flush,
+@@ -1164,7 +1298,19 @@ void LogFileObject::Write(bool force_flush,
      if (++rollover_attempt_ != kRolloverAttemptFrequency) return;
      rollover_attempt_ = 0;
  
@@ -292,7 +295,7 @@ index 4028ccc..e8d9e32 100644
      if (FLAGS_log_utc_time) {
        gmtime_r(&timestamp, &tm_time);
      } else {
-@@ -1176,13 +1319,19 @@ void LogFileObject::Write(bool force_flush,
+@@ -1176,13 +1322,19 @@ void LogFileObject::Write(bool force_flush,
      time_pid_stream.fill('0');
      time_pid_stream << 1900+tm_time.tm_year
                      << setw(2) << 1+tm_time.tm_mon
@@ -319,7 +322,7 @@ index 4028ccc..e8d9e32 100644
      const string& time_pid_string = time_pid_stream.str();
  
      if (base_filename_selected_) {
-@@ -1216,9 +1365,7 @@ void LogFileObject::Write(bool force_flush,
+@@ -1216,9 +1368,7 @@ void LogFileObject::Write(bool force_flush,
        // deadlock. Simply use a name like invalid-user.
        if (uidname.empty()) uidname = "invalid-user";
  
@@ -330,7 +333,7 @@ index 4028ccc..e8d9e32 100644
        // We're going to (potentially) try to put logs in several different dirs
        const vector<string> & log_dirs = GetLoggingDirectories();
  
-@@ -1243,36 +1390,6 @@ void LogFileObject::Write(bool force_flush,
+@@ -1243,36 +1393,6 @@ void LogFileObject::Write(bool force_flush,
        }
      }
  

thirdparty/test/glog-logfile-resume-test.sh

@@ -0,0 +1,191 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Run after rebuilding the patched glog with build-thirdparty.sh glog.
+# TP_INSTALL_DIR can point to an isolated install of either patched glog version.
+# Exercise the real static library, including fdopen failure via the GNU linker.
+
+set -eo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." &>/dev/null && pwd)"
+TP_INSTALL_DIR="${TP_INSTALL_DIR:-${ROOT}/installed}"
+
+if [[ "$(uname -s)" != 'Linux' ]]; then
+    echo "SKIP: this test requires /proc/self/fd and GNU ld --wrap"
+    exit 0
+fi
+
+[[ -f "${TP_INSTALL_DIR}/lib/libglog.a" ]] || {
+    echo "FAIL: ${TP_INSTALL_DIR}/lib/libglog.a not found; rebuild glog first" >&2
+    exit 1
+}
+
+tmpdir="$(mktemp -d)"
+trap 'rm -rf "${tmpdir}"' EXIT
+
+cat >"${tmpdir}/test.cc" <<'EOF'
+#include <dirent.h>
+#include <fcntl.h>
+#include <glog/logging.h>
+#include <limits.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#include <cerrno>
+#include <cstdio>
+#include <cstdlib>
+#include <fstream>
+#include <string>
+
+static bool fail_next_fdopen = false;
+static int failed_fdopens = 0;
+
+extern "C" FILE* __real_fdopen(int fd, const char* mode);
+extern "C" FILE* __wrap_fdopen(int fd, const char* mode) {
+    if (fail_next_fdopen) {
+        fail_next_fdopen = false;
+        ++failed_fdopens;
+        errno = ENOMEM;
+        return nullptr;
+    }
+    return __real_fdopen(fd, mode);
+}
+
+static void require(bool condition, const char* message) {
+    if (!condition) {
+        fprintf(stderr, "FAIL: %s\n", message);
+        exit(1);
+    }
+}
+
+static int check_fds(const std::string& log_dir, const char* stage, int expected) {
+    DIR* dir = opendir("/proc/self/fd");
+    require(dir != nullptr, "cannot inspect /proc/self/fd");
+    int held = 0;
+    int seeds = 0;
+    int deleted = 0;
+    bool close_on_exec = true;
+    while (dirent* entry = readdir(dir)) {
+        const std::string path = std::string("/proc/self/fd/") + entry->d_name;
+        char target[PATH_MAX];
+        const ssize_t len = readlink(path.c_str(), target, sizeof(target) - 1);
+        if (len < 0) continue;
+        target[len] = '\0';
+        const std::string name(target);
+        if (name.find(log_dir + "/be.") != 0) continue;
+        const int fd = atoi(entry->d_name);
+        struct stat st;
+        require(fstat(fd, &st) == 0, "fstat failed");
+        if (st.st_nlink == 0) ++deleted;
+        close_on_exec = close_on_exec && (fcntl(fd, F_GETFD) & FD_CLOEXEC) != 0;
+        ++held;
+        if (name.find(".seed") != std::string::npos) ++seeds;
+    }
+    closedir(dir);
+    printf("%s: held=%d expected=%d deleted=%d\n", stage, held, expected, deleted);
+    require(held == expected, "unexpected number of log fds");
+    require(deleted == 0, "deleted log still has an open fd");
+    require(close_on_exec, "log fd lost FD_CLOEXEC");
+    return seeds;
+}
+
+int main(int argc, char** argv) {
+    require(argc == 3, "usage: test LOG_DIRECTORY MODE");
+    const std::string log_dir = argv[1];
+    const std::string mode = argv[2];
+    const bool seeded = mode != "empty";
+    const google::LogSeverity severities[] = {google::GLOG_INFO, google::GLOG_WARNING};
+    const char* names[] = {"INFO", "WARNING"};
+    const std::string large(1024 * 1024, 's');
+    const std::string marker = "resume\n";
+    std::string seeds[2];
+    for (int i = 0; i < 2; ++i) {
+        seeds[i] = log_dir + "/be." + names[i] + ".log.seed";
+        if (seeded) {
+            std::ofstream seed(seeds[i]);
+            seed << large;
+            require(seed.good(), "cannot seed log");
+        }
+    }
+
+    FLAGS_log_dir = log_dir;
+    FLAGS_log_split_method = "size";
+    FLAGS_log_filenum_quota = mode == "quota-one" ? 1 : 2;
+    FLAGS_max_log_size = 1;
+    google::InitGoogleLogging("be");
+    time_t timestamp = 1800000000;
+    for (int i = 0; i < 2; ++i) {
+        fail_next_fdopen = mode == "fdopen-failure";
+        google::base::GetLogger(severities[i])
+                ->Write(true, timestamp, marker.data(), marker.size());
+        if (mode == "resume") {
+            struct stat st;
+            require(stat(seeds[i].c_str(), &st) == 0, "existing log disappeared");
+            require(st.st_size == static_cast<off_t>(large.size() + marker.size()),
+                    "existing log was not appended");
+        }
+    }
+    const int held_seeds = check_fds(log_dir, "initial write", 2);
+    if (mode == "resume") {
+        require(held_seeds == 2, "did not resume both existing logs");
+        // The seeded files already exceed max_log_size. Their saved length must
+        // trigger rotation on the next write, even though this message is tiny.
+        for (auto severity : severities) {
+            google::base::GetLogger(severity)->Write(true, timestamp + 1, marker.data(),
+                                                     marker.size());
+        }
+        require(check_fds(log_dir, "rotation from saved length", 2) == 0,
+                "resumed file length was not preserved");
+    }
+    if (mode == "fdopen-failure") {
+        require(failed_fdopens == 2, "did not inject both resume failures");
+    }
+
+    // Rotate enough times for the quota to unlink both seeded inodes. A leaked
+    // raw fd would keep the deleted files open even after closing each FILE*.
+    for (int round = 0; round < 4; ++round) {
+        timestamp += 2;
+        for (auto severity : severities) {
+            google::base::GetLogger(severity)->Write(true, timestamp, large.data(), large.size());
+            google::base::GetLogger(severity)->Write(true, timestamp + 1, marker.data(),
+                                                     marker.size());
+        }
+        check_fds(log_dir, "quota rotation", 2);
+    }
+    for (const auto& seed : seeds) {
+        require(access(seed.c_str(), F_OK) != 0 && errno == ENOENT,
+                "quota did not remove seeded log");
+    }
+    google::ShutdownGoogleLogging();
+    check_fds(log_dir, "shutdown", 0);
+    printf("PASS: %s\n", mode.c_str());
+}
+EOF
+
+extra_libs=()
+[[ -f "${TP_INSTALL_DIR}/lib/libgflags.a" ]] && extra_libs+=("${TP_INSTALL_DIR}/lib/libgflags.a")
+
+"${CXX:-c++}" -std=c++11 -O0 -g -Wall -Wextra -Werror \
+    -I"${TP_INSTALL_DIR}/include" "${tmpdir}/test.cc" \
+    "${TP_INSTALL_DIR}/lib/libglog.a" "${extra_libs[@]}" \
+    -Wl,--wrap=fdopen -pthread -ldl -o "${tmpdir}/test"
+
+for mode in resume empty quota-one fdopen-failure; do
+    mkdir "${tmpdir}/${mode}"
+    "${tmpdir}/test" "${tmpdir}/${mode}" "${mode}"
+done