// 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();
}

} // namespace doris