be/src/io/fs/hdfs/hdfs_mgr.cpp - doris - Git at Google

 // 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.

 #include "io/fs/hdfs/hdfs_mgr.h"

 #include <bthread/bthread.h>
 #include <bthread/butex.h>

 #include <chrono>
 #include <thread>

 #include "common/config.h"
 #include "common/kerberos/kerberos_ticket_mgr.h"
 #include "common/logging.h"
 #include "core/string_ref.h"
 #include "io/fs/err_utils.h"
 #include "io/hdfs_builder.h"
 #include "io/hdfs_util.h"
 #include "runtime/exec_env.h"

 namespace doris::io {

 HdfsMgr::HdfsMgr() : _should_stop_cleanup_thread(false) {
     _start_cleanup_thread();
 }

 HdfsMgr::~HdfsMgr() {
     _stop_cleanup_thread();
 }

 void HdfsMgr::_start_cleanup_thread() {
     _cleanup_thread = std::make_unique<std::thread>(&HdfsMgr::_cleanup_loop, this);
 }

 void HdfsMgr::_stop_cleanup_thread() {
     if (_cleanup_thread) {
         _should_stop_cleanup_thread = true;
         _cleanup_thread->join();
         _cleanup_thread.reset();
     }
 }

 void HdfsMgr::_cleanup_loop() {
 #ifdef BE_TEST
     static constexpr int64_t CHECK_INTERVAL_SECONDS = 1; // For testing purpose
 #else
     static constexpr int64_t CHECK_INTERVAL_SECONDS = 5; // Check stop flag every 5 seconds
 #endif
     uint64_t last_cleanup_time = std::time(nullptr);

     while (!_should_stop_cleanup_thread) {
         uint64_t current_time = std::time(nullptr);

         // Only perform cleanup if enough time has passed
         if (current_time - last_cleanup_time >= _cleanup_interval_seconds) {
             // Collect expired handlers under lock
             std::vector<std::shared_ptr<HdfsHandler>> handlers_to_cleanup;
             {
                 std::lock_guard<std::mutex> lock(_mutex);
                 std::vector<uint64_t> to_remove;

                 // Find expired handlers
                 for (const auto& entry : _fs_handlers) {
                     bool is_expired = current_time - entry.second->last_access_time >=
                                       _instance_timeout_seconds;
                     // bool is_krb_expired =
                     //         entry.second->is_kerberos_auth &&
                     //         (current_time - entry.second->create_time >=
                     //          entry.second->ticket_cache->get_ticket_lifetime_sec() / 2);
                     if (is_expired) {
                         LOG(INFO) << "Found expired HDFS handler, hash_code=" << entry.first
                                   << ", last_access_time=" << entry.second->last_access_time
                                   << ", is_kerberos=" << entry.second->is_kerberos_auth
                                   << ", principal=" << entry.second->principal
                                   << ", fs_name=" << entry.second->fs_name
                                   << ", is_expired=" << is_expired;
                         // << ", is_krb_expire=" << is_krb_expired;
                         to_remove.push_back(entry.first);
                         handlers_to_cleanup.push_back(entry.second);
                     }
                 }

                 // Remove expired handlers from map under lock
                 for (uint64_t hash_code : to_remove) {
                     _fs_handlers.erase(hash_code);
                 }
             }

             // Cleanup handlers outside lock
             for (const auto& handler : handlers_to_cleanup) {
                 LOG(INFO) << "Start to cleanup HDFS handler"
                           << ", is_kerberos=" << handler->is_kerberos_auth
                           << ", principal=" << handler->principal
                           << ", fs_name=" << handler->fs_name;

                 // The kerberos ticket cache will be automatically cleaned up when the last reference is gone
                 // DO NOT call hdfsDisconnect(), or we will meet "Filesystem closed"
                 // even if we create a new one
                 // hdfsDisconnect(handler->hdfs_fs);

                 LOG(INFO) << "Finished cleanup HDFS handler"
                           << ", fs_name=" << handler->fs_name;
             }

             handlers_to_cleanup.clear();
             last_cleanup_time = current_time;
         }

         // Sleep for a short interval to check stop flag more frequently
         std::this_thread::sleep_for(std::chrono::seconds(CHECK_INTERVAL_SECONDS));
     }
 }

 Status HdfsMgr::get_or_create_fs(const THdfsParams& hdfs_params, const std::string& fs_name,
                                  std::shared_ptr<HdfsHandler>* fs_handler) {
 #ifdef USE_HADOOP_HDFS
     if (!config::enable_java_support) {
         return Status::InvalidArgument(
                 "hdfs file system is not enabled, you can change be config enable_java_support to "
                 "true.");
     }
 #endif
     uint64_t hash_code = _hdfs_hash_code(hdfs_params, fs_name);

     // First check without lock
     {
         std::lock_guard<std::mutex> lock(_mutex);
         auto it = _fs_handlers.find(hash_code);
         if (it != _fs_handlers.end()) {
             LOG(INFO) << "Reuse existing HDFS handler, hash_code=" << hash_code
                       << ", is_kerberos=" << it->second->is_kerberos_auth
                       << ", principal=" << it->second->principal << ", fs_name=" << fs_name;
             it->second->update_access_time();
             *fs_handler = it->second;
             return Status::OK();
         }
     }

     // Create new hdfsFS handler outside the lock
     LOG(INFO) << "Start to create new HDFS handler, hash_code=" << hash_code
               << ", fs_name=" << fs_name;

     std::shared_ptr<HdfsHandler> new_fs_handler;
     RETURN_IF_ERROR(_create_hdfs_fs(hdfs_params, fs_name, &new_fs_handler));

     // Double check with lock before inserting
     {
         std::lock_guard<std::mutex> lock(_mutex);
         auto it = _fs_handlers.find(hash_code);
         if (it != _fs_handlers.end()) {
             // Another thread has created the handler, use it instead
             LOG(INFO) << "Another thread created HDFS handler, reuse it, hash_code=" << hash_code
                       << ", is_kerberos=" << it->second->is_kerberos_auth
                       << ", principal=" << it->second->principal << ", fs_name=" << fs_name;
             it->second->update_access_time();
             *fs_handler = it->second;
             return Status::OK();
         }

         // Store the new handler
         *fs_handler = new_fs_handler;
         _fs_handlers[hash_code] = new_fs_handler;

         LOG(INFO) << "Finished create new HDFS handler, hash_code=" << hash_code
                   << ", is_kerberos=" << new_fs_handler->is_kerberos_auth
                   << ", principal=" << new_fs_handler->principal << ", fs_name=" << fs_name;
     }

     return Status::OK();
 }

 Status HdfsMgr::_create_hdfs_fs_impl(const THdfsParams& hdfs_params, const std::string& fs_name,
                                      std::shared_ptr<HdfsHandler>* fs_handler) {
     HDFSCommonBuilder builder;
     RETURN_IF_ERROR(create_hdfs_builder(hdfs_params, fs_name, &builder));
     hdfsFS hdfs_fs = hdfsBuilderConnect(builder.get());
     if (hdfs_fs == nullptr) {
         return Status::InternalError("failed to connect to hdfs {}: {}", fs_name, hdfs_error());
     }

     bool is_kerberos = builder.is_kerberos();
     *fs_handler = std::make_shared<HdfsHandler>(
             hdfs_fs, is_kerberos, is_kerberos ? hdfs_params.hdfs_kerberos_principal : "",
             is_kerberos ? hdfs_params.hdfs_kerberos_keytab : "", fs_name);
     // builder.get_ticket_cache());
     return Status::OK();
 }

 // https://brpc.apache.org/docs/server/basics/
 // According to the brpc doc, JNI code checks stack layout and cannot be run in
 // bthreads so create a pthread for creating hdfs connection if necessary.
 Status HdfsMgr::_create_hdfs_fs(const THdfsParams& hdfs_params, const std::string& fs_name,
                                 std::shared_ptr<HdfsHandler>* fs_handler) {
     bool is_pthread = bthread_self() == 0;
     LOG(INFO) << "create hdfs fs, is_pthread=" << is_pthread << " fs_name=" << fs_name;
     if (is_pthread) { // running in pthread
         return _create_hdfs_fs_impl(hdfs_params, fs_name, fs_handler);
     }

     // running in bthread, switch to a pthread and wait
     Status st;
     auto btx = bthread::butex_create();
     *(int*)btx = 0;
     std::thread t([&] {
         st = _create_hdfs_fs_impl(hdfs_params, fs_name, fs_handler);
         *(int*)btx = 1;
         bthread::butex_wake_all(btx);
     });
     std::unique_ptr<int, std::function<void(int*)>> defer((int*)0x01, [&t, &btx](...) {
         if (t.joinable()) t.join();
         bthread::butex_destroy(btx);
     });
     timespec tmout {.tv_sec = std::chrono::system_clock::now().time_since_epoch().count() + 60,
                     .tv_nsec = 0};
     if (int ret = bthread::butex_wait(btx, 1, &tmout); ret != 0) {
         std::string msg = "failed to wait create_hdfs_fs finish. fs_name=" + fs_name;
         LOG(WARNING) << msg << " error=" << std::strerror(errno);
         st = Status::Error<ErrorCode::INTERNAL_ERROR, false>(msg);
     }
     return st;
 }

 uint64_t HdfsMgr::_hdfs_hash_code(const THdfsParams& hdfs_params, const std::string& fs_name) {
     uint64_t hash_code = 0;
     // The specified fsname is used first.
     // If there is no specified fsname, the default fsname is used
     if (!fs_name.empty()) {
         hash_code ^= crc32_hash(fs_name);
     } else if (hdfs_params.__isset.fs_name) {
         hash_code ^= crc32_hash(hdfs_params.fs_name);
     }

     if (hdfs_params.__isset.user) {
         hash_code ^= crc32_hash(hdfs_params.user);
     }
     if (hdfs_params.__isset.hdfs_kerberos_principal) {
         hash_code ^= crc32_hash(hdfs_params.hdfs_kerberos_principal);
     }
     if (hdfs_params.__isset.hdfs_kerberos_keytab) {
         hash_code ^= crc32_hash(hdfs_params.hdfs_kerberos_keytab);
     }
     if (hdfs_params.__isset.hdfs_conf) {
         std::map<std::string, std::string> conf_map;
         for (const auto& conf : hdfs_params.hdfs_conf) {
             conf_map[conf.key] = conf.value;
         }
         for (auto& conf : conf_map) {
             hash_code ^= crc32_hash(conf.first);
             hash_code ^= crc32_hash(conf.second);
         }
     }
     return hash_code;
 }

 } // namespace doris::io
	// 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.

	#include "io/fs/hdfs/hdfs_mgr.h"

	#include <bthread/bthread.h>
	#include <bthread/butex.h>

	#include <chrono>
	#include <thread>

	#include "common/config.h"
	#include "common/kerberos/kerberos_ticket_mgr.h"
	#include "common/logging.h"
	#include "core/string_ref.h"
	#include "io/fs/err_utils.h"
	#include "io/hdfs_builder.h"
	#include "io/hdfs_util.h"
	#include "runtime/exec_env.h"

	namespace doris::io {

	HdfsMgr::HdfsMgr() : _should_stop_cleanup_thread(false) {
	_start_cleanup_thread();
	}

	HdfsMgr::~HdfsMgr() {
	_stop_cleanup_thread();
	}

	void HdfsMgr::_start_cleanup_thread() {
	_cleanup_thread = std::make_unique<std::thread>(&HdfsMgr::_cleanup_loop, this);
	}

	void HdfsMgr::_stop_cleanup_thread() {
	if (_cleanup_thread) {
	_should_stop_cleanup_thread = true;
	_cleanup_thread->join();
	_cleanup_thread.reset();
	}
	}

	void HdfsMgr::_cleanup_loop() {
	#ifdef BE_TEST
	static constexpr int64_t CHECK_INTERVAL_SECONDS = 1; // For testing purpose
	#else
	static constexpr int64_t CHECK_INTERVAL_SECONDS = 5; // Check stop flag every 5 seconds
	#endif
	uint64_t last_cleanup_time = std::time(nullptr);

	while (!_should_stop_cleanup_thread) {
	uint64_t current_time = std::time(nullptr);

	// Only perform cleanup if enough time has passed
	if (current_time - last_cleanup_time >= _cleanup_interval_seconds) {
	// Collect expired handlers under lock
	std::vector<std::shared_ptr<HdfsHandler>> handlers_to_cleanup;
	{
	std::lock_guard<std::mutex> lock(_mutex);
	std::vector<uint64_t> to_remove;

	// Find expired handlers
	for (const auto& entry : _fs_handlers) {
	bool is_expired = current_time - entry.second->last_access_time >=
	_instance_timeout_seconds;
	// bool is_krb_expired =
	// entry.second->is_kerberos_auth &&
	// (current_time - entry.second->create_time >=
	// entry.second->ticket_cache->get_ticket_lifetime_sec() / 2);
	if (is_expired) {
	LOG(INFO) << "Found expired HDFS handler, hash_code=" << entry.first
	<< ", last_access_time=" << entry.second->last_access_time
	<< ", is_kerberos=" << entry.second->is_kerberos_auth
	<< ", principal=" << entry.second->principal
	<< ", fs_name=" << entry.second->fs_name
	<< ", is_expired=" << is_expired;
	// << ", is_krb_expire=" << is_krb_expired;
	to_remove.push_back(entry.first);
	handlers_to_cleanup.push_back(entry.second);
	}
	}

	// Remove expired handlers from map under lock
	for (uint64_t hash_code : to_remove) {
	_fs_handlers.erase(hash_code);
	}
	}

	// Cleanup handlers outside lock
	for (const auto& handler : handlers_to_cleanup) {
	LOG(INFO) << "Start to cleanup HDFS handler"
	<< ", is_kerberos=" << handler->is_kerberos_auth
	<< ", principal=" << handler->principal
	<< ", fs_name=" << handler->fs_name;

	// The kerberos ticket cache will be automatically cleaned up when the last reference is gone
	// DO NOT call hdfsDisconnect(), or we will meet "Filesystem closed"
	// even if we create a new one
	// hdfsDisconnect(handler->hdfs_fs);

	LOG(INFO) << "Finished cleanup HDFS handler"
	<< ", fs_name=" << handler->fs_name;
	}

	handlers_to_cleanup.clear();
	last_cleanup_time = current_time;
	}

	// Sleep for a short interval to check stop flag more frequently
	std::this_thread::sleep_for(std::chrono::seconds(CHECK_INTERVAL_SECONDS));
	}
	}

	Status HdfsMgr::get_or_create_fs(const THdfsParams& hdfs_params, const std::string& fs_name,
	std::shared_ptr<HdfsHandler>* fs_handler) {
	#ifdef USE_HADOOP_HDFS
	if (!config::enable_java_support) {
	return Status::InvalidArgument(
	"hdfs file system is not enabled, you can change be config enable_java_support to "
	"true.");
	}
	#endif
	uint64_t hash_code = _hdfs_hash_code(hdfs_params, fs_name);

	// First check without lock
	{
	std::lock_guard<std::mutex> lock(_mutex);
	auto it = _fs_handlers.find(hash_code);
	if (it != _fs_handlers.end()) {
	LOG(INFO) << "Reuse existing HDFS handler, hash_code=" << hash_code
	<< ", is_kerberos=" << it->second->is_kerberos_auth
	<< ", principal=" << it->second->principal << ", fs_name=" << fs_name;
	it->second->update_access_time();
	*fs_handler = it->second;
	return Status::OK();
	}
	}

	// Create new hdfsFS handler outside the lock
	LOG(INFO) << "Start to create new HDFS handler, hash_code=" << hash_code
	<< ", fs_name=" << fs_name;

	std::shared_ptr<HdfsHandler> new_fs_handler;
	RETURN_IF_ERROR(_create_hdfs_fs(hdfs_params, fs_name, &new_fs_handler));

	// Double check with lock before inserting
	{
	std::lock_guard<std::mutex> lock(_mutex);
	auto it = _fs_handlers.find(hash_code);
	if (it != _fs_handlers.end()) {
	// Another thread has created the handler, use it instead
	LOG(INFO) << "Another thread created HDFS handler, reuse it, hash_code=" << hash_code
	<< ", is_kerberos=" << it->second->is_kerberos_auth
	<< ", principal=" << it->second->principal << ", fs_name=" << fs_name;
	it->second->update_access_time();
	*fs_handler = it->second;
	return Status::OK();
	}

	// Store the new handler
	*fs_handler = new_fs_handler;
	_fs_handlers[hash_code] = new_fs_handler;

	LOG(INFO) << "Finished create new HDFS handler, hash_code=" << hash_code
	<< ", is_kerberos=" << new_fs_handler->is_kerberos_auth
	<< ", principal=" << new_fs_handler->principal << ", fs_name=" << fs_name;
	}

	return Status::OK();
	}

	Status HdfsMgr::_create_hdfs_fs_impl(const THdfsParams& hdfs_params, const std::string& fs_name,
	std::shared_ptr<HdfsHandler>* fs_handler) {
	HDFSCommonBuilder builder;
	RETURN_IF_ERROR(create_hdfs_builder(hdfs_params, fs_name, &builder));
	hdfsFS hdfs_fs = hdfsBuilderConnect(builder.get());
	if (hdfs_fs == nullptr) {
	return Status::InternalError("failed to connect to hdfs {}: {}", fs_name, hdfs_error());
	}

	bool is_kerberos = builder.is_kerberos();
	*fs_handler = std::make_shared<HdfsHandler>(
	hdfs_fs, is_kerberos, is_kerberos ? hdfs_params.hdfs_kerberos_principal : "",
	is_kerberos ? hdfs_params.hdfs_kerberos_keytab : "", fs_name);
	// builder.get_ticket_cache());
	return Status::OK();
	}

	// https://brpc.apache.org/docs/server/basics/
	// According to the brpc doc, JNI code checks stack layout and cannot be run in
	// bthreads so create a pthread for creating hdfs connection if necessary.
	Status HdfsMgr::_create_hdfs_fs(const THdfsParams& hdfs_params, const std::string& fs_name,
	std::shared_ptr<HdfsHandler>* fs_handler) {
	bool is_pthread = bthread_self() == 0;
	LOG(INFO) << "create hdfs fs, is_pthread=" << is_pthread << " fs_name=" << fs_name;
	if (is_pthread) { // running in pthread
	return _create_hdfs_fs_impl(hdfs_params, fs_name, fs_handler);
	}

	// running in bthread, switch to a pthread and wait
	Status st;
	auto btx = bthread::butex_create();
	(int)btx = 0;
	std::thread t([&] {
	st = _create_hdfs_fs_impl(hdfs_params, fs_name, fs_handler);
	(int)btx = 1;
	bthread::butex_wake_all(btx);
	});
	std::unique_ptr<int, std::function<void(int)>> defer((int)0x01, [&t, &btx](...) {
	if (t.joinable()) t.join();
	bthread::butex_destroy(btx);
	});
	timespec tmout {.tv_sec = std::chrono::system_clock::now().time_since_epoch().count() + 60,
	.tv_nsec = 0};
	if (int ret = bthread::butex_wait(btx, 1, &tmout); ret != 0) {
	std::string msg = "failed to wait create_hdfs_fs finish. fs_name=" + fs_name;
	LOG(WARNING) << msg << " error=" << std::strerror(errno);
	st = Status::Error<ErrorCode::INTERNAL_ERROR, false>(msg);
	}
	return st;
	}

	uint64_t HdfsMgr::_hdfs_hash_code(const THdfsParams& hdfs_params, const std::string& fs_name) {
	uint64_t hash_code = 0;
	// The specified fsname is used first.
	// If there is no specified fsname, the default fsname is used
	if (!fs_name.empty()) {
	hash_code ^= crc32_hash(fs_name);
	} else if (hdfs_params.__isset.fs_name) {
	hash_code ^= crc32_hash(hdfs_params.fs_name);
	}

	if (hdfs_params.__isset.user) {
	hash_code ^= crc32_hash(hdfs_params.user);
	}
	if (hdfs_params.__isset.hdfs_kerberos_principal) {
	hash_code ^= crc32_hash(hdfs_params.hdfs_kerberos_principal);
	}
	if (hdfs_params.__isset.hdfs_kerberos_keytab) {
	hash_code ^= crc32_hash(hdfs_params.hdfs_kerberos_keytab);
	}
	if (hdfs_params.__isset.hdfs_conf) {
	std::map<std::string, std::string> conf_map;
	for (const auto& conf : hdfs_params.hdfs_conf) {
	conf_map[conf.key] = conf.value;
	}
	for (auto& conf : conf_map) {
	hash_code ^= crc32_hash(conf.first);
	hash_code ^= crc32_hash(conf.second);
	}
	}
	return hash_code;
	}

	} // namespace doris::io