正文

用python做自动化测试--Python实现远程性能监控

IP云
IP云V管理员/0 阅读
11/30
小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

http://blog.csdn.net/powerccna/article/details/8044222

在性能测试中，监控被测试服务器的性能指标是个重要的工作，包括CPU/Memory/IO/Network，但大多数人估计都是直接在被测试服务器的运行监控程序。我们开始也是这样做的。但这样做带来一个问题是，测试人员需要在每台被测试服务器上部署监控程序，增加了部署的工作量，而且经常因为Python版本的问题，有些模块不兼容，或者第三方模块需要再次安装。

改进性能测试监控工具：

1. 能远程监控被测试服务器，这样测试人员就不需要在每个被测试机器安装监控工具了。

2. 被测试服务器上不需要安装agent，监控结果能取到本地。

3. 本地服务器上的python模块和兼容性问题，可以通过Python  virtualenv解决，每个测试人员维护自己的一套Python环境。

Google了下，找到了pymeter（thttp://pymeter.sourceforge.net/）， 看了下源代码，很多工作还没完成，但这个思路和我的是一样的。而且我在其他项目中已经实现了远程发送命令的模块。 所以不如直接在自己的项目上扩展。

远程发送命令的模块开始是基于Pexpect(http://www.noah.org/wiki/Pexpect)实现的, Pexpect很强大，它是一个用来启动子程序，并使用正则表达式对程序输出做出特定响应，以此实现与其自动交互的 Python 模块。用他来可以很容易实现telnet,ftp,ssh的操作。 但Pexpect无windows下的版本，这是我抛弃它的原因，无法达到测试工具兼容所有系统的要求。 所以就用telent模块替换了Pexpect,实现了远程发送命令和获取结果。

#file name: telnetoperate.py

  1. #!/usr/bin/env python
  2. #coding=utf-8
  3. import time,sys,logging,traceback,telnetlib,socket
  4. class TelnetAction:
  5. def __init__(self,host,prompt,account,accountPasswd,RootPasswd=""):
  6. self.log=logging.getLogger()
  7. self.host=host
  8. self.account=account
  9. self.accountPasswd=accountPasswd
  10. self.RootPasswd=RootPasswd
  11. self.possible_prompt = ["#","$"]
  12. self.prompt=prompt
  13. self.default_time_out=20
  14. self.child =None
  15. self.login()
  16. def expand_expect(self,expect_list):
  17. try:
  18. result=self.child.expect(expect_list,self.default_time_out)
  19. except EOFError:
  20. self.log.error("No text was read, please check reason")
  21. if result[0]==-1:
  22. self.log.error("Expect result"+str(expect_list)+" don't exist")
  23. else:
  24. pass
  25. return result
  26. def login(self):
  27. """Connect to a remote host and login.
  28. """
  29. try:
  30. self.child = telnetlib.Telnet(self.host)
  31. self.expand_expect(['login:'])
  32. self.child.write(self.account+ '\n')
  33. self.expand_expect(['assword:'])
  34. self.child.write(self.accountPasswd + '\n')
  35. self.expand_expect(self.possible_prompt)
  36. self.log.debug("swith to root account on host "+self.host)
  37. if self.RootPasswd!="":
  38. self.child.write('su -'+'\n')
  39. self.expand_expect(['assword:'])
  40. self.child.write(self.RootPasswd+'\n')
  41. self.expand_expect(self.possible_prompt)
  42. #self.child.write('bash'+'\n')
  43. #self.expand_expect(self.possible_prompt)
  44. self.child.read_until(self.prompt)
  45. self.log.info("login host "+self.host+" successfully")
  46. return True
  47. except:
  48. print "Login failed,please check ip address and account/passwd"
  49. self.log.error("log in host "+self.host+" failed, please check reason")
  50. return False
  51. def send_command(self,command,sleeptime=0.5):
  52. """Run a command on the remote host.
  53. @param command: Unix command
  54. @return: Command output
  55. @rtype: String
  56. """
  57. self.log.debug("Starting to execute command: "+command)
  58. try:
  59. self.child.write(command + '\n')
  60. if self.expand_expect(self.possible_prompt)[0]==-1:
  61. self.log.error("Executed command "+command+" is failed, please check it")
  62. return False
  63. else:
  64. time.sleep(sleeptime)
  65. self.log.debug("Executed command "+command+" is successful")
  66. return True
  67. except socket.error:
  68. self.log.error("when executed command "+command+" the connection maybe break, reconnect")
  69. traceback.print_exc()
  70. for i in range(0,3):
  71. self.log.error("Telnet session is broken from "+self.host+ ", reconnecting....")
  72. if self.login():
  73. break
  74. return False
  75. def get_output(self,time_out=2):
  76. reponse=self.child.read_until(self.prompt,time_out)
  77. #print "response:",reponse
  78. self.log.debug("reponse:"+reponse)
  79. return  self.__strip_output(reponse)
  80. def send_atomic_command(self, command):
  81. self.send_command(command)
  82. command_output = self.get_output()
  83. self.logout()
  84. return command_output
  85. def process_is_running(self,process_name,output_string):
  86. self.send_command("ps -ef | grep "+process_name+" | grep -v grep")
  87. output_list=[output_string]
  88. if self.expand_expect(output_list)[0]==-1:
  89. return False
  90. else:
  91. return True
  92. def __strip_output(self, response):
  93. #Strip everything from the response except the actual command output.
  94. #split the response into a list of the lines
  95. lines = response.splitlines()
  96. self.log.debug("lines:"+str(lines))
  97. if len(lines)>1:
  98. #if our command was echoed back, remove it from the output
  99. if self.prompt in lines[0]:
  100. lines.pop(0)
  101. #remove the last element, which is the prompt being displayed again
  102. lines.pop()
  103. #append a newline to each line of output
  104. lines = [item + '\n' for item in lines]
  105. #join the list back into a string and return it
  106. return ''.join(lines)
  107. else:
  108. self.log.info("The response is blank:"+response)
  109. return "Null response"
  110. def logout(self):
  111. self.child.close()

telnetoperate.py代码说明：

1.  __init__(self,host,prompt,account,accountPasswd,RootPasswd="")

这里用到了多个登陆账号（account，root），原因是我们的机器开始不能直接root登陆，需要先用普通用户登陆，才能切换到root账号，所以这里出现了account, rootPasswd这2个参数，如果你的机器可以直接root账号登陆，或者你不需要切换到root账号，可以就用account, accountPasswd就可以了。

prompt是命令行提示符，机器配置不一样，可能是$或者#,用来判断一个命令执行是否完成。

2.  send_command(self,command,sleeptime=0.5)

这里sleeptime=0.5是为了避免很多机器性能不好，命令执行比较慢，命令还没返回，会导致获取命令后的结果失败。如果你嫌这样太慢了，可以调用的时候send_command(command,0)

process_is_running(

process_name

监控远程机器：

#simplemonitor.py

  1. #!/usr/bin/env python
  2. #coding=utf-8
  3. import time
  4. import telnetoperate
  5. remote_server=telnetoperate.TelnetAction("192.168.23.235","#","user","passwd123")
  6. #get cpu information
  7. cpu=remote_server.get_output("sar 1 1 |tail -1")
  8. memory=remote_server.get_output("top | head -5 |grep -i memory")
  9. io=remote_server.get_output("iostat -x 1 2|grep -v '^$' |grep -vi 'dev'")

这样在任何一台机器上就可以实现监控远程多个机器了，信息集中化管理，方便进一步分析。如果你想cpu, memory, io独立的监控，可以多线程或者起多个监控进程，在多线程中需要注意的时候，必须对每个监控实例建立一个telnet连接，get_output是从telnet 建立的socket里面去获取数据，如果多个监控实例用同一个socket会导致数据混乱。

相关阅读

目录[+]