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

#!/usr/bin/env python

#coding=utf-8

import time,sys,logging,traceback,telnetlib,socket

class TelnetAction:

def __init__(self,host,prompt,account,accountPasswd,RootPasswd=""):

self.log=logging.getLogger()

self.host=host

self.account=account

self.accountPasswd=accountPasswd

self.RootPasswd=RootPasswd

self.possible_prompt = ["#","$"]

self.prompt=prompt

self.default_time_out=20

self.child =None

self.login()

def expand_expect(self,expect_list):

try:

result=self.child.expect(expect_list,self.default_time_out)

except EOFError:

self.log.error("No text was read, please check reason")

if result[0]==-1:

self.log.error("Expect result"+str(expect_list)+" don"t exist")

else:

pass

return result

def login(self):

"""Connect to a remote host and login.

"""

try:

self.child = telnetlib.Telnet(self.host)

self.expand_expect(["login:"])

self.child.write(self.account+ " ")

self.expand_expect(["assword:"])

self.child.write(self.accountPasswd + " ")

self.expand_expect(self.possible_prompt)

self.log.debug("swith to root account on host "+self.host)

if self.RootPasswd!="":

self.child.write("su -"+" ")

self.expand_expect(["assword:"])

self.child.write(self.RootPasswd+" ")

self.expand_expect(self.possible_prompt)

#self.child.write("bash"+" ")

#self.expand_expect(self.possible_prompt)

self.child.read_until(self.prompt)

self.log.info("login host "+self.host+" successfully")

return True

except:

print "Login failed,please check ip address and account/passwd"

self.log.error("log in host "+self.host+" failed, please check reason")

return False

def send_command(self,command,sleeptime=0.5):

"""Run a command on the remote host.

@param command: Unix command

@return: Command output

@rtype: String

"""

self.log.debug("Starting to execute command: "+command)

try:

self.child.write(command + " ")

if self.expand_expect(self.possible_prompt)[0]==-1:

self.log.error("Executed command "+command+" is failed, please check it")

return False

else:

time.sleep(sleeptime)

self.log.debug("Executed command "+command+" is successful")

return True

except socket.error:

self.log.error("when executed command "+command+" the connection maybe break, reconnect")

traceback.print_exc()

for i in range(0,3):

self.log.error("Telnet session is broken from "+self.host+ ", reconnecting....")

if self.login():

break

return False

def get_output(self,time_out=2):

reponse=self.child.read_until(self.prompt,time_out)

#print "response:",reponse

self.log.debug("reponse:"+reponse)

return self.__strip_output(reponse)

def send_atomic_command(self, command):

self.send_command(command)

command_output = self.get_output()

self.logout()

return command_output

def process_is_running(self,process_name,output_string):

self.send_command("ps -ef | grep "+process_name+" | grep -v grep")

output_list=[output_string]

if self.expand_expect(output_list)[0]==-1:

return False

else:

return True

def __strip_output(self, response):

#Strip everything from the response except the actual command output.

#split the response into a list of the lines

lines = response.splitlines()

self.log.debug("lines:"+str(lines))

if len(lines)>1:

#if our command was echoed back, remove it from the output

if self.prompt in lines[0]:

lines.pop(0)

#remove the last element, which is the prompt being displayed again

lines.pop()

#append a newline to each line of output

lines = [item + " " for item in lines]

#join the list back into a string and return it

return "".join(lines)

else:

self.log.info("The response is blank:"+response)

return "Null response"

def logout(self):

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

#!/usr/bin/env python

#coding=utf-8

import time

import telnetoperate

remote_server=telnetoperate.TelnetAction("192.168.23.235","#","user","passwd123")

#get cpu information

cpu=remote_server.get_output("sar 1 1 |tail -1")

memory=remote_server.get_output("top | head -5 |grep -i memory")

io=remote_server.get_output("iostat -x 1 2|grep -v "^$" |grep -vi "dev"")

這樣在任何一臺機器上就可以實現監控遠程多個機器了，信息集中化管理，方便進一步分析。如果你想cpu, memory, io獨立的監控，可以多線程或者起多個監控進程，在多線程中需要注意的時候，必須對每個監控實例建立一個telnet連接，get_output是從telnet 建立的socket里面去獲取數據，如果多個監控實例用同一個socket會導致數據混亂。

總結

以上是生活随笔為你收集整理的python做硬件自动化测试-用python做自动化测试--Python实现远程性能监控的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。