/**
 * Tencent is pleased to support the open source community by making Tars available.
 *
 * Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
 *
 * Licensed under the BSD 3-Clause License (the "License"); you may not use this file except 
 * in compliance with the License. You may obtain a copy of the License at
 *
 * https://opensource.org/licenses/BSD-3-Clause
 *
 * 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 "KeepAliveThread.h"
#include "RegistryProxy.h"
#include "NodeRollLogger.h"
#include "util/tc_timeprovider.h"
#include "util.h"

KeepAliveThread::KeepAliveThread()
: _terminate(false)
, _registryPrx(NULL)
{
 _runTime = time(0);
 _nodeInfo = _platformInfo.getNodeInfo();
 _heartTimeout = TC_Common::strto<int>(g_pconf->get("/tars/node/keepalive<heartTimeout>", "10"));
 _monitorInterval = TC_Common::strto<int>(g_pconf->get("/tars/node/keepalive<monitorInterval>", "2"));

 _monitorIntervalMs = TC_Common::strto<int>(g_pconf->get("/tars/node/keepalive<monitorIntervalMs>", "10"));

 _synInterval = TC_Common::strto<int>(g_pconf->get("/tars/node/keepalive<synStatInterval>", "60"));
 _synStatBatch = g_pconf->get("/tars/node/keepalive<synStatBatch>", "Y");
 _monitorInterval = _monitorInterval > 10 ? 10 : (_monitorInterval < 1 ? 1 : _monitorInterval);

}

KeepAliveThread::~KeepAliveThread()
{
 terminate();
}

void KeepAliveThread::terminate()
{
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << endl;

 _terminate = true;

 if (isAlive())
 {
 _lock.notifyAll();
 getThreadControl().join();
 }

}

bool KeepAliveThread::timedWait(int millsecond)
{
 TC_ThreadLock::Lock lock(_lock);

 if (_terminate)
 {
 return true;
 }

 return _lock.timedWait(millsecond);
}

void KeepAliveThread::run()
{
 bool bLoaded = false;
 bool bRegistered = false;

 while (!_terminate)
 {
 int64_t startMs = TC_TimeProvider::getInstance()->getNowMs();

 try
 {
 //获取主控代理
 if (!_registryPrx)
 {
 _registryPrx = AdminProxy::getInstance()->getRegistryProxy();
 if (!_registryPrx)
 {
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "RegistryProxy init fail ! retry it after " + TC_Common::tostr(_monitorInterval) + " second" << endl;
 }
 }

 //注册node信息
 if (_registryPrx && bRegistered == false)
 {
 bRegistered = registerNode();
 }

 //加载服务
 if (bLoaded == false)
 {
 bLoaded = loadAllServers();
 }

 //检查服务的limit配置是否需要更新
 if (bLoaded)
 {
 ServerFactory::getInstance()->setAllServerResourceLimit();
 }

 //检查服务
 checkAlive();

 // 上报node状态
 if (reportAlive() != 0)
 {
 bRegistered = false; //registry服务重启 需要重新注册
 }
 }
 catch (exception& e)
 {
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "catch exception|" << e.what() << endl;
 }
 catch (...)
 {
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "catch unkown exception|" << endl;
 }

 {
 int64_t useMs = (TC_TimeProvider::getInstance()->getNowMs() - startMs);
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "run use:" << useMs << " ms" << endl;
 }

 timedWait(ServerFactory::getInstance()->getMinMonitorIntervalMs());
 }
}

bool KeepAliveThread::registerNode()
{
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "registerNode begin===============|node name|" << _nodeInfo.nodeName << endl;
 try
 {
 int iRet = _registryPrx->registerNode(_nodeInfo.nodeName, _nodeInfo, _platformInfo.getLoadInfo());

 if (iRet == 0)
 {
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "register node succ" << endl;
 return true;
 }
 }
 catch (exception& e)
 {
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "KeepAliveThread::registerNode catch exception|" << e.what() << endl;
 }

 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "register node fail" << endl;

 return false;
}

bool KeepAliveThread::loadAllServers()
{
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "load server begin===============|node name|" << _nodeInfo.nodeName << endl;

 /**
 * 由于加载失败或者node下没有部署服务,这里就会一直去访问主控
 * 增加这个限制,如果超过5次失败,则不去加载,10分钟重试10次
 * 如果之后有新服务部署,会自动添加到node缓存中,不影响监控流程
 */
 static int iFailNum = 0;
 if (iFailNum >= 5)
 {
 static time_t tLastLoadTime = 0;
 if ((TNOW - tLastLoadTime) < 600)
 {
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "load server fail" << endl;
 return false;
 }
 tLastLoadTime = TNOW;
 iFailNum = 0;
 }

 try
 {
 if (ServerFactory::getInstance()->loadServer())
 {
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "load server succ" << endl;
 return true;
 }
 }
 catch (exception& e)
 {
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "catch exception|" << e.what() << endl;
 }

 iFailNum++;

 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "load server fail" << endl;

 return false;
}

int KeepAliveThread::reportAlive()
{
 try
 {
 static time_t tReport;
 time_t tNow = TNOW;
 if (tNow - tReport > _heartTimeout)
 {
 tReport = tNow;

 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "node keep alive ----------------------------------------------------|" << time(0) << "|" << pthread_self() << '\n' << endl;
 
 int iRet = _registryPrx->keepAlive(_nodeInfo.nodeName, _platformInfo.getLoadInfo());

 return iRet;
 }
 }
 catch (exception& e)
 {
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "KeepAliveThread::reportAlive catch exception|" << e.what() << endl;
 }

 return 0;
}

int KeepAliveThread::synStat()
{
 try
 {
 int iRet = -1;

 vector<ServerStateInfo> v;
 _stat.swap(v);

 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "node syn stat size|" << v.size() << "|" << _synStatBatch << "|" << pthread_self() << endl;

 if (v.size() > 0)
 {
 iRet = _registryPrx->updateServerBatch(v);
 }

 return iRet;
 }
 catch (exception& e)
 {
 string s = e.what();
 if (s.find("server function mismatch exception") != string::npos)
 {
 _synStatBatch = "N";
 }
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << "catch exception|" << s << endl;
 }

 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "fail,set SynStatBatch = " << _synStatBatch << endl;

 return -1;
}

void KeepAliveThread::checkAlive()
{
 int64_t startMs = TC_TimeProvider::getInstance()->getNowMs();

 static time_t tSyn = 0;
 bool bNeedSynServerState = false;

 time_t tNow = TNOW;
 if (tNow - tSyn > _synInterval)
 {
 tSyn = tNow;
 bNeedSynServerState = true;
 }

 string sServerId;

 _stat.clear();

 map<string, ServerGroup> mmServerList = ServerFactory::getInstance()->getAllServers();
 map<string, ServerGroup>::const_iterator it = mmServerList.begin();
 for (; it != mmServerList.end(); it++)
 {
 map<string, ServerObjectPtr>::const_iterator p = it->second.begin();
 for (; p != it->second.end(); p++)
 {
 try
 {
 sServerId = it->first + "." + p->first;
 ServerObjectPtr pServerObjectPtr = p->second;
 if (!pServerObjectPtr)
 {
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << sServerId << "|=NULL|" << endl;
 continue;
 }

 pServerObjectPtr->doMonScript();

 if (TNOW - _runTime < ServantHandle::HEART_BEAT_INTERVAL * 5)
 {
 //等待心跳包
 continue;
 }

 //corelimit checking
 pServerObjectPtr->checkCoredumpLimit();

 //(checkServer时,上报Server所占用的内存给主控)
 pServerObjectPtr->checkServer(_heartTimeout);

 //cache 同步server 状态
 if (bNeedSynServerState == true)
 {
 ServerInfo tServerInfo;
 tServerInfo.application = it->first;
 tServerInfo.serverName = p->first;
 ServerDescriptor tServerDescriptor = pServerObjectPtr->getServerDescriptor();
 tServerDescriptor.settingState = pServerObjectPtr->isEnabled() == true ? "active" : "inactive";
 g_serverInfoHashmap.set(tServerInfo, tServerDescriptor);

 ServerStateInfo tServerStateInfo;
 tServerStateInfo.serverState = (pServerObjectPtr->IsEnSynState() ? pServerObjectPtr->getState() : tars::Inactive);
 tServerStateInfo.processId = pServerObjectPtr->getPid();
 tServerStateInfo.nodeName = _nodeInfo.nodeName;
 tServerStateInfo.application = it->first;
 tServerStateInfo.serverName = p->first;

 if (TC_Common::lower(_synStatBatch) == "y")
 {
 _stat.push_back(tServerStateInfo);
 }
 else
 {
 pServerObjectPtr->asyncSynState();
 }
 }
 }
 catch (exception& e)
 {
 NODE_LOG("KeepAliveThread")->error() << FILE_FUN << sServerId << " catch exception|" << e.what() << endl;
 }
 }
 }

 if (bNeedSynServerState)
 {
 synStat();
 }

 {
 int64_t useMs = (TC_TimeProvider::getInstance()->getNowMs() - startMs);
 NODE_LOG("KeepAliveThread")->debug() << FILE_FUN << "checkAlive use:" << useMs << " ms" << endl;
 }
}