taosadapter.py 9.4 KB
Newer Older
C
cpwu 已提交
1
import socket
C
cpwu 已提交
2 3 4 5 6 7 8 9 10 11 12
from fabric2 import Connection
from util.log import *
from util.common import *


class TAdapter:
    def __init__(self):
        self.running = 0
        self.deployed = 0
        self.remoteIP = ""
        self.taosadapter_cfg_dict = {
C
cpwu 已提交
13
            "debug"         : True,
C
cpwu 已提交
14 15
            "taosConfigDir" : "",
            "port"          : 6041,
sangshuduo's avatar
sangshuduo 已提交
16
            "logLevel"      : "error",
C
cpwu 已提交
17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
            "cors"          : {
                "allowAllOrigins" : True,
            },
            "pool"          : {
                "maxConnect"    : 4000,
                "maxIdle"       : 4000,
                "idleTimeout"   : "1h"
            },
            "ssl"           : {
                "enable"        : False,
                "certFile"      : "",
                "keyFile"       : "",
            },
            "log"           : {
                "path"                  : "",
                "rotationCount"         : 30,
                "rotationTime"          : "24h",
                "rotationSize"          : "1GB",
C
cpwu 已提交
35
                "enableRecordHttpSql"   : True,
C
cpwu 已提交
36 37 38 39 40 41 42 43 44 45 46 47 48 49
                "sqlRotationCount"      : 2,
                "sqlRotationTime"       : "24h",
                "sqlRotationSize"       : "1GB",
            },
            "monitor"       : {
                "collectDuration"           : "3s",
                "incgroup"                  : False,
                "pauseQueryMemoryThreshold" : 70,
                "pauseAllMemoryThreshold"   : 80,
                "identity"                  : "",
                "writeToTD"                 : True,
                "user"                      : "root",
                "password"                  : "taosdata",
                "writeInterval"             : "30s"
C
cpwu 已提交
50 51
            },
            "opentsdb"      : {
sangshuduo's avatar
sangshuduo 已提交
52
                "enable"        : True
C
cpwu 已提交
53 54
            },
            "influxdb"      : {
sangshuduo's avatar
sangshuduo 已提交
55
                "enable"        : True
C
cpwu 已提交
56 57
            },
            "statsd"      : {
sangshuduo's avatar
sangshuduo 已提交
58
                "enable"        : True
C
cpwu 已提交
59 60
            },
            "collectd"      : {
sangshuduo's avatar
sangshuduo 已提交
61
                "enable"        : True
C
cpwu 已提交
62 63
            },
            "opentsdb_telnet"      : {
sangshuduo's avatar
sangshuduo 已提交
64
                "enable"        : True
C
cpwu 已提交
65 66
            },
            "node_exporter"      : {
sangshuduo's avatar
sangshuduo 已提交
67
                "enable"        : True
C
cpwu 已提交
68 69
            },
            "prometheus"      : {
sangshuduo's avatar
sangshuduo 已提交
70
                "enable"        : True
C
cpwu 已提交
71
            },
C
cpwu 已提交
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113
        }
    # TODO: add taosadapter env:
    # 1. init cfg.toml.dict :OK
    # 2. dump dict to toml : OK
    # 3. update cfg.toml.dict :OK
    # 4. check adapter exists : OK
    # 5. deploy adapter cfg : OK
    # 6. adapter start :   OK
    # 7. adapter stop

    def init(self, path, remoteIP=""):
        self.path = path
        self.remoteIP = remoteIP
        binPath = get_path() + "/../../../"
        binPath = os.path.realpath(binPath)

        if path == "":
            self.path = os.path.abspath(binPath + "../../")
        else:
            self.path = os.path.realpath(path)

        if self.remoteIP:
            try:
                self.config = eval(remoteIP)
                self.remote_conn = Connection(host=self.config["host"], port=self.config["port"], user=self.config["user"], connect_kwargs={'password':self.config["password"]})
            except Exception as e:
                tdLog.notice(e)

    def update_cfg(self, update_dict :dict):
        if not isinstance(update_dict, dict):
            return
        if "log" in update_dict and "path" in update_dict["log"]:
            del update_dict["log"]["path"]
        for key, value in update_dict.items():
            if key in ["cors", "pool", "ssl", "log", "monitor", "opentsdb", "influxdb", "statsd", "collectd", "opentsdb_telnet", "node_exporter", "prometheus"]:
                if  isinstance(value, dict):
                    for k, v in value.items():
                        self.taosadapter_cfg_dict[key][k] = v
            else:
                self.taosadapter_cfg_dict[key] = value

    def check_adapter(self):
C
cpwu 已提交
114
        if get_path(tool="taosadapter"):
C
cpwu 已提交
115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
            return False
        else:
            return True

    def remote_exec(self, updateCfgDict, execCmd):
        remoteCfgDict = copy.deepcopy(updateCfgDict)
        if "log" in remoteCfgDict and "path" in remoteCfgDict["log"]:
            del remoteCfgDict["log"]["path"]

        remoteCfgDictStr = base64.b64encode(toml.dumps(remoteCfgDict).encode()).decode()
        execCmdStr = base64.b64encode(execCmd.encode()).decode()
        with self.remote_conn.cd((self.config["path"]+sys.path[0].replace(self.path, '')).replace('\\','/')):
            self.remote_conn.run(f"python3 ./test.py  -D {remoteCfgDictStr} -e {execCmdStr}" )

    def cfg(self, option, value):
        cmd = f"echo {option} = {value} >> {self.cfg_path}"
        if os.system(cmd) != 0:
            tdLog.exit(cmd)

    def deploy(self, *update_cfg_dict):
        self.log_dir = f"{self.path}/sim/dnode1/log"
        self.cfg_dir = f"{self.path}/sim/dnode1/cfg"
        self.cfg_path = f"{self.cfg_dir}/taosadapter.toml"

        cmd = f"touch {self.cfg_path}"
        if os.system(cmd) != 0:
            tdLog.exit(cmd)

        self.taosadapter_cfg_dict["log"]["path"] = self.log_dir
        if bool(update_cfg_dict):
            self.update_cfg(update_dict=update_cfg_dict)

        if (self.remoteIP == ""):
            dict2toml(self.taosadapter_cfg_dict, self.cfg_path)
        else:
            self.remote_exec(self.taosadapter_cfg_dict, "tAdapter.deploy(update_cfg_dict)")

        self.deployed = 1

        tdLog.debug(f"taosadapter is deployed and configured by {self.cfg_path}")

    def start(self):
        bin_path = get_path(tool="taosadapter")

        if (bin_path == ""):
            tdLog.exit("taosadapter not found!")
        else:
            tdLog.info(f"taosadapter found: {bin_path}")

        if platform.system().lower() == 'windows':
            cmd = f"mintty -h never {bin_path} -c {self.cfg_dir}"
        else:
sangshuduo's avatar
sangshuduo 已提交
167
            cmd = f"nohup {bin_path} -c {self.cfg_path} > /dev/null & "
C
cpwu 已提交
168 169 170 171 172 173 174 175 176 177 178 179

        if  self.remoteIP:
            self.remote_exec(self.taosadapter_cfg_dict, f"tAdapter.deployed=1\ntAdapter.log_dir={self.log_dir}\ntAdapter.cfg_dir={self.cfg_dir}\ntAdapter.start()")
            self.running = 1
        else:
            os.system(f"rm -rf {self.log_dir}/taosadapter*")
            if os.system(cmd) != 0:
                tdLog.exit(cmd)
            self.running = 1
            tdLog.debug(f"taosadapter is running with {cmd} " )

            time.sleep(0.1)
C
cpwu 已提交
180 181 182 183 184 185

            taosadapter_port = self.taosadapter_cfg_dict["port"]
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(3)
            try:
                res = s.connect_ex((self.remoteIP, taosadapter_port))
C
cpwu 已提交
186
                s.shutdown(2)
C
cpwu 已提交
187 188 189 190 191 192
                if res == 0:
                    tdLog.info(f"the taosadapter has been started, using port:{taosadapter_port}")
                else:
                    tdLog.info(f"the taosadapter do not started!!!")
            except socket.error as  e:
                tdLog.notice("socket connect error!")
C
cpwu 已提交
193 194 195
            finally:
                if s:
                    s.close()
C
cpwu 已提交
196 197
            # tdLog.debug("the taosadapter has been started.")
            time.sleep(1)
C
cpwu 已提交
198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215

    def start_taosadapter(self):
        """
            use this method, must deploy taosadapter
        """
        bin_path = get_path(tool="taosadapter")

        if (bin_path == ""):
            tdLog.exit("taosadapter not found!")
        else:
            tdLog.info(f"taosadapter found: {bin_path}")

        if self.deployed == 0:
            tdLog.exit("taosadapter is not deployed")

        if platform.system().lower() == 'windows':
            cmd = f"mintty -h never {bin_path} -c {self.cfg_dir}"
        else:
sangshuduo's avatar
sangshuduo 已提交
216
            cmd = f"nohup {bin_path} -c {self.cfg_path} > /dev/null & "
C
cpwu 已提交
217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240

        if  self.remoteIP:
            self.remote_exec(self.taosadapter_cfg_dict, f"tAdapter.deployed=1\ntAdapter.log_dir={self.log_dir}\ntAdapter.cfg_dir={self.cfg_dir}\ntAdapter.start()")
            self.running = 1
        else:
            if os.system(cmd) != 0:
                tdLog.exit(cmd)
            self.running = 1
            tdLog.debug(f"taosadapter is running with {cmd} " )

            time.sleep(0.1)

    def stop(self, force_kill=False):
        signal = "-SIGKILL" if force_kill else "-SIGTERM"

        if  self.remoteIP:
            self.remote_exec(self.taosadapter_cfg_dict, "tAdapter.running=1\ntAdapter.stop()")
            tdLog.info("stop taosadapter")
            return

        toBeKilled = "taosadapter"

        if self.running != 0:
            psCmd = f"ps -ef|grep -w {toBeKilled}| grep -v grep | awk '{{print $2}}'"
C
cpwu 已提交
241
            # psCmd = f"pgrep {toBeKilled}"
C
cpwu 已提交
242
            processID = subprocess.check_output(
C
cpwu 已提交
243
                psCmd, shell=True)
C
cpwu 已提交
244 245

            while(processID):
sangshuduo's avatar
sangshuduo 已提交
246
                killCmd = f"pkill {signal} {processID} > /dev/null "
C
cpwu 已提交
247 248 249 250 251
                os.system(killCmd)
                time.sleep(1)
                processID = subprocess.check_output(
                    psCmd, shell=True).decode("utf-8")
            if not platform.system().lower() == 'windows':
C
cpwu 已提交
252 253 254 255 256 257
                port = 6041
                fuserCmd = f"fuser -k -n tcp {port} > /dev/null"
                os.system(fuserCmd)
                # for port in range(6030, 6041):
                    # fuserCmd = f"fuser -k -n tcp {port} > /dev/null"
                    # os.system(fuserCmd)
C
cpwu 已提交
258

sangshuduo's avatar
sangshuduo 已提交
259 260
            self.running = 0
            tdLog.debug(f"taosadapter is stopped by kill {signal}")
C
cpwu 已提交
261 262 263



sangshuduo's avatar
sangshuduo 已提交
264
tAdapter = TAdapter()