/**
 * 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 "StatImp.h"
#include "StatServer.h"

///////////////////////////////////////////////////////////
TC_ThreadMutex StatImpThreadData::_mutex;
pthread_key_t StatImpThreadData::_key = 0;
size_t StatImpThreadData::_iNo=0;

///////////////////////////////////////////////////////////
StatImpThreadData::StatImpThreadData()
: _iThreadIndex(0)
{
}
void StatImpThreadData::destructor(void* p)
{
 StatImpThreadData * pSptd = (StatImpThreadData*)p;
 if(pSptd)
 { 
 delete pSptd;
 pSptd = NULL;
 }
}

StatImpThreadData * StatImpThreadData::getData()
{
 if(_key == 0)
 {
 TC_LockT<TC_ThreadMutex> lock(_mutex);
 if(_key == 0)
 {
 int iRet = ::pthread_key_create(&_key, StatImpThreadData::destructor);

 if (iRet != 0)
 {
 TLOGERROR("StatImpThreadData pthread_key_create fail:"<< errno << ":" << strerror(errno) << endl);
 return NULL;
 }
 }
 }

 StatImpThreadData * pSptd = (StatImpThreadData*)pthread_getspecific(_key);

 if(!pSptd)
 {
 TC_LockT<TC_ThreadMutex> lock(_mutex);

 pSptd = new StatImpThreadData();
 pSptd->_iThreadIndex = _iNo;
 ++_iNo;

 int iRet = pthread_setspecific(_key, (void *)pSptd);

 assert(iRet == 0);
 }
 return pSptd;
}

//////////////////////////////////////////////////////////
void StatImp::initialize()
{
 StatImpThreadData * td = StatImpThreadData::getData();

 if(td)
 {
 _threadIndex = td->_iThreadIndex;
 }
 else
 {
 _threadIndex = 0;
 TLOGERROR("StatImp::initialize StatImpThreadData::getData error." << endl);
 }
}

///////////////////////////////////////////////////////////
//
int StatImp::reportMicMsg( const map<tars::StatMicMsgHead, tars::StatMicMsgBody>& statmsg,bool bFromClient, tars::TarsCurrentPtr current )
{
 TLOGINFO("report---------------------------------access size:" << statmsg.size() << "|bFromClient:" <<bFromClient << endl);

 for ( map<StatMicMsgHead, StatMicMsgBody>::const_iterator it = statmsg.begin(); it != statmsg.end(); it++ )
 {
 StatMicMsgHead head = it->first;
 const StatMicMsgBody &body = it->second;

 if(bFromClient)
 {
 head.masterIp = current->getIp(); //以前是自己获取主调ip,现在从proxy直接

 head.slaveName = getSlaveName(head.slaveName);
 }
 else
 {
 head.slaveIp = current->getIp();//现在从proxy直接
 }

 string sMasterName = head.masterName;
 string::size_type pos = sMasterName.find("@");
 if (pos != string::npos)
 {
 head.masterName = sMasterName.substr(0, pos);
 head.tarsVersion = sMasterName.substr(pos+1);
 }

 map<string, string>::iterator it_vip;
 it_vip = g_app.getVirtualMasterIp().find(getSlaveName(head.slaveName));
 if( it_vip != g_app.getVirtualMasterIp().end())
 {
 head.masterIp = it_vip->second; //按 slaveName来匹配,填入假的主调ip,减小入库数据量
 }

 //如果不是info等级的日志级别,就别往里走了
 ostringstream os;
 if(LOG->IsNeedLog(TarsRollLogger::INFO_LOG))
 {
 os.str("");
 head.displaySimple(os);
 body.displaySimple(os);
 }

 //三个数据都为0时不入库
 if(body.count == 0 && body.execCount == 0 && body.timeoutCount == 0)
 {
 TLOGINFO(os.str()<<"|zero"<<endl);
 continue;
 }

 int iAddHash = addHashMap(head,body);

 TLOGINFO(os.str()<<"|"<<iAddHash<<endl);
 }

 return 0;
}

int StatImp::reportSampleMsg(const vector<StatSampleMsg> &msg,tars::TarsCurrentPtr current )
{
 TLOGINFO("sample---------------------------------access size:" << msg.size() << endl);

 for(unsigned i=0; i<msg.size();i++)
 {
 StatSampleMsg sample = msg[i];
 sample.masterIp = current->getIp();

 ostringstream os;
 sample.displaySimple(os);
 FDLOG()<<os.str()<<endl;
 }

 return 0;
}

///////////////////////////////////////////////////////////
//
int StatImp::addHashMap(const StatMicMsgHead &head, const StatMicMsgBody &body )
{
 size_t iIndex = _threadIndex;

 //dump数据到文件
 int iBufferIndex = g_app.getSelectBufferIndex();

 map<int,vector<int64_t> >& mBuffer = g_app.getBuffer();
 map<int,vector<int64_t> >::iterator iter = mBuffer.find(iBufferIndex);
 iter->second[iIndex] = TNOW;

 dump2file();

 iBufferIndex = g_app.getSelectBufferIndex();

 string sKey = head.slaveName;
 sKey += head.masterName;
 sKey += head.interfaceName;
 sKey += head.masterIp;
 sKey += head.slaveIp;

 int iHashKey = _hashf(sKey) % g_app.getBuffNum();

 StatHashMap *pHashMap = g_app.getHashMapBuff(iBufferIndex, iHashKey);
 
 //////////////////////////////////////////////////////////////////////////////////////

 float rate = (pHashMap->getMapHead()._iUsedChunk) * 1.0/pHashMap->allBlockChunkCount();

 if(rate >0.9)
 {
 TLOGERROR("StatImp::addHashMap hashmap will full|_iMemSize:" << pHashMap->getMapHead()._iMemSize << endl);
 FDLOG("HashMap")<<"StatImp::addHashMap hashmap will full|_iMemSize:" << pHashMap->getMapHead()._iMemSize << endl;
 return -1;
 }

 int iRet = pHashMap->add(head, body);
 if(iRet != 0)
 {
 TLOGDEBUG("StatImp::addHashMap set g_hashmap recourd erro|" << iRet << endl);
 return iRet;
 }
 return iRet;
}

///////////////////////////////////////////////////////////
string StatImp::getSlaveName(const string& sSlaveName)
{
 string::size_type pos = sSlaveName.find(".");
 if(pos != string::npos)
 {
 pos = sSlaveName.find(".",pos+1);
 if(pos != string::npos)
 {
 return sSlaveName.substr(0,pos);
 }
 }
 return sSlaveName;
}

void StatImp::dump2file()
{
 static string g_sDate;
 static string g_sFlag;
 static time_t g_tLastDumpTime = 0;

 time_t tTimeNow = TC_TimeProvider::getInstance()->getNow();
 time_t tTimeInterv = g_app.getInserInterv() *60;//second

 if(g_tLastDumpTime == 0)
 {
 g_app.getTimeInfo(g_tLastDumpTime,g_sDate,g_sFlag);
 }

 if(tTimeNow - g_tLastDumpTime > tTimeInterv)
 {
 static TC_ThreadLock g_mutex;
 TC_ThreadLock::Lock lock( g_mutex );
 if(tTimeNow - g_tLastDumpTime > tTimeInterv)
 {
 g_app.getTimeInfo(g_tLastDumpTime,g_sDate,g_sFlag);

 int iSelectBuffer = g_app.getSelectBufferIndex();
 iSelectBuffer = !iSelectBuffer;

 g_app.setSelectBufferIndex(iSelectBuffer);

 TLOGDEBUG("StatImp::dump2file select buffer:" << iSelectBuffer << "|TimeInterv:" << tTimeInterv << "|now:" << tTimeNow << "|last:" << g_tLastDumpTime << endl);
 FDLOG("CountStat") << "stat ip:" << ServerConfig::LocalIp << "|StatImp::dump2file select buffer:" << iSelectBuffer << "|TimeInterv:" << tTimeInterv << "|now:" << tTimeNow << "|last:" << g_tLastDumpTime << endl;
 }
 }
}