// 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 "agent/heartbeat_server.h"

#include <gen_cpp/HeartbeatService.h>

#include <gen_cpp/HeartbeatService_types.h>

#include <gen_cpp/Types_types.h>

#include <glog/logging.h>

#include <memory>

#include <ostream>

#include <string>

#include <vector>

#include "cloud/cloud_cluster_info.h"

#include "cloud/cloud_tablet_mgr.h"

#include "cloud/config.h"

#include "common/config.h"

#include "common/status.h"

#include "runtime/cluster_info.h"

#include "runtime/exec_env.h"

#include "runtime/fragment_mgr.h"

#include "runtime/heartbeat_flags.h"

#include "service/backend_options.h"

#include "storage/storage_engine.h"

#include "util/debug_util.h"

#include "util/mem_info.h"

#include "util/network_util.h"

#include "util/thrift_server.h"

#include "util/time.h"

namespace apache {

namespace thrift {

class TProcessor;

} // namespace thrift

} // namespace apache

namespace doris {

HeartbeatServer::HeartbeatServer(ClusterInfo* cluster_info)

        : _engine(ExecEnv::GetInstance()->storage_engine()),

          _cluster_info(cluster_info),

          _fe_epoch(0) {

    _be_epoch = GetCurrentTimeMicros() / 1000;

}

void HeartbeatServer::init_cluster_id() {

    _cluster_info->cluster_id = _engine.effective_cluster_id();

}

void HeartbeatServer::heartbeat(THeartbeatResult& heartbeat_result,

                                const TMasterInfo& master_info) {

    //print heartbeat in every minute

    LOG_EVERY_N(INFO, 12) << "get heartbeat from FE."

                          << "host:" << master_info.network_address.hostname

                          << ", rpc port:" << master_info.network_address.port

                          << ", cluster id:" << master_info.cluster_id

                          << ", frontend_info:" << PrintFrontendInfos(master_info.frontend_infos)

                          << ", counter:" << google::COUNTER << ", BE start time: " << _be_epoch;

    MonotonicStopWatch watch;

    watch.start();

    // do heartbeat

    Status st = _heartbeat(master_info);

    st.to_thrift(&heartbeat_result.status);

    if (st.ok()) {

        heartbeat_result.backend_info.__set_be_port(config::be_port);

        heartbeat_result.backend_info.__set_http_port(config::webserver_port);

        heartbeat_result.backend_info.__set_be_rpc_port(-1);

        heartbeat_result.backend_info.__set_brpc_port(config::brpc_port);

        heartbeat_result.backend_info.__set_arrow_flight_sql_port(config::arrow_flight_sql_port);

        heartbeat_result.backend_info.__set_version(get_short_version());

        heartbeat_result.backend_info.__set_be_start_time(_be_epoch);

        heartbeat_result.backend_info.__set_be_node_role(config::be_node_role);

        // If be is gracefully stop, then k_doris_exist is set to true

        heartbeat_result.backend_info.__set_is_shutdown(doris::k_doris_exit);

        heartbeat_result.backend_info.__set_fragment_executing_count(

                get_fragment_executing_count());

        heartbeat_result.backend_info.__set_fragment_last_active_time(

                get_fragment_last_active_time());

        heartbeat_result.backend_info.__set_be_mem(MemInfo::physical_mem());

    }

    watch.stop();

    if (watch.elapsed_time() > 1000L * 1000L * 1000L) {

        LOG(WARNING) << "heartbeat consume too much time. time=" << watch.elapsed_time()

                     << ", host:" << master_info.network_address.hostname

                     << ", port:" << master_info.network_address.port

                     << ", cluster id:" << master_info.cluster_id

                     << ", frontend_info:" << PrintFrontendInfos(master_info.frontend_infos)

                     << ", counter:" << google::COUNTER << ", BE start time: " << _be_epoch;

    }

}

Status HeartbeatServer::_heartbeat(const TMasterInfo& master_info) {

    std::lock_guard<std::mutex> lk(_hb_mtx);

    // Check cluster id

    if (_cluster_info->cluster_id == -1) {

        LOG(INFO) << "get first heartbeat. update cluster id";

        // write and update cluster id

        RETURN_IF_ERROR(_engine.set_cluster_id(master_info.cluster_id));

        _cluster_info->cluster_id = master_info.cluster_id;

        LOG(INFO) << "record cluster id. host: " << master_info.network_address.hostname

                  << ". port: " << master_info.network_address.port

                  << ". cluster id: " << master_info.cluster_id

                  << ". frontend_infos: " << PrintFrontendInfos(master_info.frontend_infos);

    } else {

        if (_cluster_info->cluster_id != master_info.cluster_id) {

            return Status::InternalError(

                    "invalid cluster id. ignore. Record cluster id ={}, record frontend info {}. "

                    "Invalid cluster_id={}, invalid frontend info {}",

                    _cluster_info->cluster_id,

                    PrintFrontendInfos(ExecEnv::GetInstance()->get_frontends()),

                    master_info.cluster_id, PrintFrontendInfos(master_info.frontend_infos));

        }

    }

    if (master_info.__isset.backend_ip) {

        // master_info.backend_ip may be an IP or domain name, and it should be renamed 'backend_host', as it requires compatibility with historical versions, the name is still 'backend_ ip'

        if (master_info.backend_ip != BackendOptions::get_localhost()) {

            LOG(INFO) << master_info.backend_ip << " not equal to to backend localhost "

                      << BackendOptions::get_localhost();

            // step1: check master_info.backend_ip is IP or FQDN

            if (!is_valid_ip(master_info.backend_ip)) {

                //step2: resolve FQDN to IP

                std::string ip;

                Status status =

                        hostname_to_ip(master_info.backend_ip, ip, BackendOptions::is_bind_ipv6());

                if (!status.ok()) {

                    std::stringstream ss;

                    ss << "can not get ip from fqdn: " << status.to_string();

                    LOG(WARNING) << ss.str();

                    return status;

                }

                LOG(INFO) << "master_info.backend_ip: " << master_info.backend_ip

                          << ", hostname_to_ip: " << ip;

                //step3: get all ips of the interfaces on this machine

                std::vector<InetAddress> hosts;

                status = get_hosts(&hosts);

                if (!status.ok() || hosts.empty()) {

                    return Status::InternalError(

                            "the status was not ok when get_hosts, error is {}",

                            status.to_string());

                }

                //step4: check if the IP of FQDN belongs to the current machine and update BackendOptions._s_localhost

                bool set_new_localhost = false;

                for (auto& addr : hosts) {

                    if (addr.get_host_address() == ip) {

                        BackendOptions::set_localhost(master_info.backend_ip);

                        set_new_localhost = true;

                        break;

                    }

                }

                if (!set_new_localhost) {

                    return Status::InternalError(

                            "the host recorded in master is {}, but we cannot found the local ip "

                            "that mapped to that host. backend={}",

                            master_info.backend_ip, BackendOptions::get_localhost());

                }

            } else {

                // if is ip,not check anything,use it

                BackendOptions::set_localhost(master_info.backend_ip);

            }

            LOG(WARNING) << "update localhost done, the new localhost is "

                         << BackendOptions::get_localhost();

        }

    }

    bool need_report = false;

    if (_cluster_info->master_fe_addr.hostname != master_info.network_address.hostname ||

        _cluster_info->master_fe_addr.port != master_info.network_address.port) {

        if (master_info.epoch > _fe_epoch) {

            _cluster_info->master_fe_addr.hostname = master_info.network_address.hostname;

            _cluster_info->master_fe_addr.port = master_info.network_address.port;

            _fe_epoch = master_info.epoch;

            need_report = true;

            LOG(INFO) << "master change. new master host: "

                      << _cluster_info->master_fe_addr.hostname

                      << ". port: " << _cluster_info->master_fe_addr.port

                      << ". epoch: " << _fe_epoch;

        } else {

            return Status::InternalError(

                    "epoch is not greater than local. ignore heartbeat. host: {}, port: {}, local "

                    "epoch: {}, received epoch: {}",

                    _cluster_info->master_fe_addr.hostname, _cluster_info->master_fe_addr.port,

                    _fe_epoch, master_info.epoch);

        }

    } else {

        // when Master FE restarted, host and port remains the same, but epoch will be increased.

        if (master_info.epoch > _fe_epoch) {

            _fe_epoch = master_info.epoch;

            need_report = true;

            LOG(INFO) << "master restarted. epoch: " << _fe_epoch;

        }

    }

    if (master_info.__isset.token) {

        if (_cluster_info->token == "") {

            _cluster_info->token = master_info.token;

            LOG(INFO) << "get token. token: " << _cluster_info->token;

        } else if (_cluster_info->token != master_info.token) {

            return Status::InternalError("invalid token. local: {}, master: {}",

                                         _cluster_info->token, master_info.token);

        }

    }

    if (master_info.__isset.http_port) {

        _cluster_info->master_fe_http_port = master_info.http_port;

    }

    if (master_info.__isset.heartbeat_flags) {

        HeartbeatFlags* heartbeat_flags = ExecEnv::GetInstance()->heartbeat_flags();

        heartbeat_flags->update(master_info.heartbeat_flags);

    }

    if (master_info.__isset.backend_id) {

        _cluster_info->backend_id = master_info.backend_id;

        BackendOptions::set_backend_id(master_info.backend_id);

    }

    if (master_info.__isset.frontend_infos) {

        ExecEnv::GetInstance()->update_frontends(master_info.frontend_infos);

    } else {

        LOG_EVERY_N(WARNING, 2) << fmt::format(

                "Heartbeat from {}:{} does not have frontend_infos, this may because we are "

                "upgrading cluster",

                master_info.network_address.hostname, master_info.network_address.port);

    }

    if (master_info.__isset.meta_service_endpoint != config::is_cloud_mode()) {

        LOG(WARNING) << "Detected mismatch in cloud mode configuration between FE and BE. "

                     << "FE cloud mode: "

                     << (master_info.__isset.meta_service_endpoint ? "true" : "false")

                     << ", BE cloud mode: " << (config::is_cloud_mode() ? "true" : "false")

                     << ". If fe is earlier than version 3.0.2, the message can be ignored.";

    }

    if (master_info.__isset.meta_service_endpoint) {

        if (config::meta_service_endpoint.empty() && !master_info.meta_service_endpoint.empty()) {

            auto st = config::set_config("meta_service_endpoint", master_info.meta_service_endpoint,

                                         true);

            LOG(INFO) << "set config meta_service_endpoint " << master_info.meta_service_endpoint

                      << " " << st;

        }

        if (master_info.meta_service_endpoint != config::meta_service_endpoint) {

            LOG(WARNING) << "Detected mismatch in meta_service_endpoint configuration between FE "

                            "and BE. "

                         << "FE meta_service_endpoint: " << master_info.meta_service_endpoint

                         << ", BE meta_service_endpoint: " << config::meta_service_endpoint;

            std::vector<std::string> old_endpoints =

                    doris::split(config::meta_service_endpoint, ",");

            std::vector<std::string> new_endpoints =

                    doris::split(master_info.meta_service_endpoint, ",");

            auto has_intersection = false;

            for (auto endpoint : new_endpoints) {

                if (std::find(old_endpoints.begin(), old_endpoints.end(), endpoint) !=

                    old_endpoints.end()) {

                    has_intersection = true;

                }

            }

            if (has_intersection) {

                auto st = config::set_config("meta_service_endpoint",

                                             master_info.meta_service_endpoint, true);

                LOG(INFO) << "change config meta_service_endpoint to "

                          << master_info.meta_service_endpoint << " " << st;

            }

            if (!has_intersection && config::enable_meta_service_endpoint_consistency_check) {

                return Status::InvalidArgument<false>(

                        "fe and be do not work in same mode or meta_service_endpoint mismatch,"

                        "fe meta_service_endpoint: {}, be meta_service_endpoint: {}",

                        master_info.meta_service_endpoint, config::meta_service_endpoint);

            }

        }

    }

    if (master_info.__isset.cloud_unique_id &&

        config::cloud_unique_id != master_info.cloud_unique_id &&

        config::enable_use_cloud_unique_id_from_fe) {

        auto st = config::set_config("cloud_unique_id", master_info.cloud_unique_id, true);

        LOG(INFO) << "set config cloud_unique_id " << master_info.cloud_unique_id << " " << st;

    }

    if (master_info.__isset.cloud_cluster_info &&

        master_info.cloud_cluster_info.__isset.isStandby) {

        auto* cloud_cluster_info = static_cast<CloudClusterInfo*>(_cluster_info);

        cloud_cluster_info->set_is_in_standby(master_info.cloud_cluster_info.isStandby);

    }

    if (master_info.__isset.tablet_report_inactive_duration_ms) {

        doris::g_tablet_report_inactive_duration_ms =

                master_info.tablet_report_inactive_duration_ms;

    }

    if (master_info.__isset.auth_token) {

        if (_cluster_info->curr_auth_token == "") {

            _cluster_info->curr_auth_token = master_info.auth_token;

            LOG(INFO) << "set new auth token: " << master_info.auth_token;

        } else if (_cluster_info->curr_auth_token != master_info.auth_token) {

            LOG(INFO) << "last auth token: " << _cluster_info->last_auth_token

                      << "set new auth token: " << master_info.auth_token;

            _cluster_info->last_auth_token = _cluster_info->curr_auth_token;

            _cluster_info->curr_auth_token = master_info.auth_token;

        }

    }

    if (need_report) {

        LOG(INFO) << "Master FE is changed or restarted. report tablet and disk info immediately";

        _engine.notify_listeners();

    }

    return Status::OK();

}

Status create_heartbeat_server(ExecEnv* exec_env, uint32_t server_port,

                               std::unique_ptr<ThriftServer>* thrift_server,

                               uint32_t worker_thread_num, ClusterInfo* cluster_info) {

    HeartbeatServer* heartbeat_server = new HeartbeatServer(cluster_info);

    if (heartbeat_server == nullptr) {

        return Status::InternalError("Get heartbeat server failed");

    }

    heartbeat_server->init_cluster_id();

    std::shared_ptr<HeartbeatServer> handler(heartbeat_server);

    std::shared_ptr<HeartbeatServiceProcessor::TProcessor> server_processor(

            new HeartbeatServiceProcessor(handler));

    *thrift_server = std::make_unique<ThriftServer>("heartbeat", server_processor, server_port,

                                                    worker_thread_num);

    return Status::OK();

}

} // namespace doris