#!/bin/env bash
##Description: check drainer replicate 

function f_syntax
{
    [ ! -z "$1" ] && echo ">>[$(date "+%Y%m%d.%H%M%S")] $1"
    echo " "
    echo "bash ${ShellName}  [-hostname <hostname>]"
    echo "  [-ip <ip>] [-port <port>] [-user <user>] [-passwd <passwd>]  "
    echo "  [-standbyip <ip>] [-standbyport <port>] [-standbyuser <user>] [-standbypasswd <passwd>]  "
    echo " "
    echo " #sample:"
    echo " #sh ${ShellName} -standbyip 172.16.xxx -standbyport 4100 -standbyuser abcuser1 -standbypasswd xxx -checkpointschema ccb"
    echo "   -primaryip 172.16.xxx -primaryport 4000 -primaryuser abcuser1 -primarypasswd xxx "
    exit 22
}

function f_PrintLog
{
    _LogFlag=$(echo "$1"|tr "a-z" "A-Z")
    [ $(echo "$1"|wc -L) -eq 4 ] && _LogFlag=$(echo "${_LogFlag}_")
    _LogMessage=$2
    # declare LogFile="${LogFile:-"/tmp/${PShellName:-${ShellName}}.log"}"
    # touch "${LogFile}" 2>/dev/null
    # if [ -w "${LogFile}" ]
    # then
    #     echo "$(date "+%Y%m%d.%H%M%S"):[${_LogFlag}]:${UserName}@${HostName}:${2}"|tee -a "${LogFile}";
    # else
        echo "$(date "+%Y%m%d.%H%M%S"):[${_LogFlag}]:${UserName}@${HostName}:${2}";
    # fi
}


###############################################################################
##Init##
ShellName="$(echo $0|awk -F / '{print $NF}')";ShellOption="$@";
WorkDir="$(echo $0|sed s/${ShellName}//g)";[ -z "${WorkDir}" ] && WorkDir=${PWD};cd ${WorkDir};WorkDir=${PWD}
LogDir="${WorkDir%/}/logs";mkdir -p ${LogDir};chmod ugo+rwx ${LogDir} 2>/dev/null
LogFile="${LogDir}/${ShellName}.$(date "+%Y%m%d.%H%M%S").log";LogLevel=3;
touch "${LogFile}" 2>/dev/null;chmod ugo+rw "${LogFile}" 2>/dev/null
declare Shellcksum=$(cksum ${WorkDir%/}/${ShellName} 2>/dev/null|awk '{printf "%d",$1}')
declare -i UserID=$(id -u);declare UserName=$(whoami);declare HostName=$(hostname);
###############################################################################
##Get inopt
[ -z "${ShellOption}" ] && f_syntax;
for inopt in ${ShellOption}
do
    case $(echo $inopt|tr a-z A-Z) in
    -HOSTNAME|-HOST) CurOpt="-HOSTNAME";continue;;
    -CLSNAME|-C) CurOpt="-CLSNAME";continue;;
    -PRIMARYIP) CurOpt="-PRIMARYIP";continue;;
    -PRIMARYPORT) CurOpt="-PRIMARYPORT";continue;;
    -PRIMARYUSER) CurOpt="-PRIMARYUSER";continue;;
    -PRIMARYPASSWORD|-PRIMARYPASSWD) CurOpt="-PRIMARYPASSWORD";continue;;
    -STANDBYIP) CurOpt="-STANDBYIP";continue;;
    -STANDBYPORT) CurOpt="-STANDBYPORT";continue;;
    -STANDBYUSER) CurOpt="-STANDBYUSER";continue;;
    -STANDBYPASSWORD|-STANDBYPASSWD) CurOpt="-STANDBYPASSWORD";continue;;
    -CHECKPOINTSCHEMA|-CKPSCHEMA) CurOpt="-CKPSCHEMA";continue;;
    -HELP|--HELP|-H) f_syntax;;
    -*) CurOpt="";continue;;
    esac
    case "${CurOpt}" in
    -HOSTNAME) declare LocalHost="${inopt}";continue;;
    -CLSNAME) declare LocalCluster="${inopt}";continue;;
    -IGNOREUSERS) declare ignoreusers=$(echo "${ignoreusers} ${inopt}");continue;;
    -PRIMARYIP) declare p_ip="${inopt}";continue;;
    -PRIMARYPORT) declare p_port="${inopt}";continue;;
    -PRIMARYUSER) declare p_user="${inopt}";continue;;
    -PRIMARYPASSWORD) declare p_passwd="${inopt}";continue;;
    -STANDBYIP) declare s_ip="${inopt}";continue;;
    -STANDBYPORT) declare s_port="${inopt}";continue;;
    -STANDBYUSER) declare s_user="${inopt}";continue;;
    -STANDBYPASSWORD) declare s_passwd="${inopt}";continue;;
    -CKPSCHEMA) declare c_schema="${inopt}";continue;;
    esac
done

f_PrintLog "Info" "Start to run Script ${ShellName}, the Shellcksum is ${Shellcksum}"
f_PrintLog "Debug" "Script Options:${ShellOption}"


function f_runsql
{
    _ip=$1
    _port=$2
    _user=$3
    _passwd=$4
    _execsql=$5

    f_PrintLog "Info" "start to run sql:${_execsql}"
    exec_result=$(mysql -h${_ip} -P${_port} -u${_user} -p${_passwd} -NBe "${_execsql}")
}


function f_committs_diff
{
    f_PrintLog "Info" "Start to run f_committs_diff."
    #最多检查3次,每次间隔10s
    for i in 1 2 3 
    do
        #commit ts
        ck_sql="select checkPoint from ${c_schema}.checkpoint"
        f_runsql "${s_ip}" "${s_port}" "${s_user}" "${s_passwd}" "${ck_sql}"
        cmt_ts=$(echo $exec_result  | python -m json.tool | awk '{
        if($1~"commitTS"){gsub(",","",$2);printf "%s\n",$2;exit}
        }')
        echo $exec_result  | python -m json.tool
        
        #time-tso diff
        ck_sql="SELECT timestampdiff(SECOND,TIDB_PARSE_TSO(${cmt_ts}),NOW()) "
        f_runsql "${s_ip}" "${s_port}" "${s_user}" "${s_passwd}" "${ck_sql}"
        f_PrintLog "Info" "Binlog commitTS is ${cmt_ts}, is ${exec_result} seconds away from now!"
        
        [ ${exec_result} -lt 30 ] && return 0
        sleep 10
    done 
    return 1
}

function f_tsmap_stable
{
    #获取checkpoint表中的tsmap信息
    f_PrintLog "Info" "Start to run f_committs_diff."
    #tsmap
    ck_sql="select checkPoint from ${c_schema}.checkpoint"
    f_runsql "${s_ip}" "${s_port}" "${s_user}" "${s_passwd}" "${ck_sql}"
    tsmap_tso=$(echo $exec_result  | python -m json.tool | awk '{if($0~"ry-ts"){printf "%s",$2}}')
    f_PrintLog "Info" "Binlog tsmap is ${tsmap_tso}!"
}


function f_runsqlstmt
{
    _ip=$1
    _port=$2
    _user=$3
    _passwd=$4
    _execsql=$5

    f_PrintLog "Info" "START TO RUN SQL STMT:${_execsql}, DBinfo ${_ip}:${_port}"
    mysql -h${_ip} -P${_port} -u${_user} -p${_passwd:-123} -vvv -e "${_execsql}"
    return $?
}

function f_check_insert
{
    #在主集群执行建表和写入数据,去目标集群查询数据
    f_PrintLog "Info" "Start to run f_check_insert."
    #sql
    create_tab="create table If Not Exists test.drainer_delay(ts datetime)"
    delete_tab="delete from test.drainer_delay"
    insert_tab="insert into test.drainer_delay(ts) values(now(0))"
    select_tab="select max(ts) from test.drainer_delay"
    
    #Primary
    f_runsqlstmt "${p_ip}" "${p_port}" "${p_user}" "${p_passwd}"  "${create_tab}"
    f_runsqlstmt "${p_ip}" "${p_port}" "${p_user}" "${p_passwd}"  "${delete_tab}"
    f_runsqlstmt "${p_ip}" "${p_port}" "${p_user}" "${p_passwd}"  "${insert_tab}"
    f_runsqlstmt "${p_ip}" "${p_port}" "${p_user}" "${p_passwd}"  "${select_tab}"

    
    p_val=$(mysql -h${p_ip} -P${p_port} -u${p_user} -p${p_passwd:-123} -NBe "${select_tab}")
    if [ $? -ne 0 ]
    then
        f_PrintLog "Error" "exec sql failed, stmt: ${select_tab}"
    fi
    
    sleep 1
    #Standby
    # f_PrintLog "Info" "START TO RUN SQL STMT:${select_tab}, DBinfo ${s_ip}:${s_port} "
    f_runsqlstmt "${s_ip}" "${s_port}" "${s_user}" "${s_passwd}"  "${select_tab}"
    for i in 1 2 3
    do
        s_val=$(mysql -h${s_ip} -P${s_port} -u${s_user} -p${s_passwd:-123} -NBe "${select_tab}")
        if [ "${s_val:-0}" = "${p_val:-1}" ]
        then
            # break
            f_PrintLog "Info" "Real time Data replicate successful: ${p_val}=${s_val}, ${i} check pass! "
            return 0
        fi
        sleep 1
    done

    f_PrintLog "Error" "Real time data replicate:${p_val}!=${s_val}, check failed!!!"
    return 1
}


###################
####    Main   ####
###################

#1. 写入数据,确认复制状态
#1. 确认 cluster_b 上 drainer 断点表中 binlog commitTS 与 cluster_b tso 时间差小于 30s; 
#2. tsmap 在两分钟之内不再变化
#方案二：上有写测试数据，观测下游是否消费
function check_main
{

[ -z "${c_schema}" ] && c_schema="tidb_binlog"

f_check_insert
rt_code1=$?

f_tsmap_stable
tsmap_a=${tsmap_tso}

f_committs_diff
rt_code2=$?

rt_code=$((rt_code1+rt_code2))
#
f_PrintLog "Info" "Start sleep 120, and check tsmap do not change !"
sleep 30
f_tsmap_stable
tsmap_b=${tsmap_tso}


if [ ${rt_code} -eq 0 -a "${tsmap_a}" = "${tsmap_b}" ]
then 
    f_PrintLog "Info" "Binlog checkpoint's commitTS and tsmap, ${tsmap_a}=${tsmap_b} check pass!"
elif [ "${tsmap_a}" != "${tsmap_b}" ]
then
    f_PrintLog "Error" "Binlog tsmap changed in 2 minutes, ${tsmap_a}!=${tsmap_b} drainer still working, check failed!!!"
    exit 1
elif [ ${rt_code} -ne 0 ]
then
    f_PrintLog "Error" "Binlog commitTS not close to now, drainer status may abnormal or stopped already, check failed!!!"
    exit 1
else
    f_PrintLog "Error" "Binlog checkpoint's commitTS and tsmap check failed!"
    exit 1
fi

}

check_main | tee -a ${LogFile:-"/tmp/${PShellName:-${ShellName}}.log"}
exit ${PIPESTATUS[0]}