GNUer's blog

这几天配置dns集群，为了测试各种场景下的服务影响时间，需要自己写个脚本进行统计。没有什么现成的好的工具，就用python里面的dnspython模块写了个小的脚本
[python]
#!/usr/bin/python
#****************************************************************#
# ScriptName: dnsquery.py
# Author: GNUer
# Create Date: 2013-06-28 12:52
# Modify Author: GNUer
# Modify Date: 2013-06-28 12:52
# Function:
#***************************************************************#
import dns.resolver
import dns.exception
import time
import datetime
import signal
import sys
def get_time():
t=datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return t
def get_mtime():
t=datetime.datetime.now().strftime("%H:%M:%S.%f")
return t
def get_second():
return time.time()
def do_exit(sig,arg):
print "exit dns test"
sys.exit(1)
def dns_test(num,sleep_time):
resolver = dns.resolver.Resolver()
#resolver.timeout = 0.01
resolver.lifetime =0.0013
#change this time accord to reponse time
resolver.nameservers=[‘7.7.7.7′,’6.6.6.6’]
testlist=("search.xxx.com","obs.xxx.com","if.xxx.com")
last_flag=True
fail_start=0
fail_finish=0
for i in range(num):
for test in testlist:
try:
for target in resolver.query(qname=test):
print "\x1b[32m",get_time(),test,target,"\x1b[m"
if last_flag == False:
last_flag= True
fail_finish=get_second()
errortime=fail_finish-fail_start
print "\x1b[1;31m","error time is:",errortime,"\x1b[m"
except dns.exception.Timeout:
print "\x1b[31m",get_time(),test,"failed","\x1b[m"
if last_flag == True:
fail_start=get_second()
last_flag = False
time.sleep(sleep_time)
signal.signal(signal.SIGINT,do_exit)
dns_test(10000000,0.33)
[/python]
需要比较注意的是需要根据自己的实际情况把resolver.lifetime设置为比dns服务器的响应时间稍微大一点的值，但是需要小于平均响应时间的2倍。不然测试的时间不是太准确。因为总共测试的域名是3个，所以我把每次sleep的时间设置为0.32左右，使得1s总共能发3个请求左右。
测试的时候可以先把这个脚本一直跑着，然后去做各种操作，看中间的影响时间是多少。虽然这个脚本是测试dns的，但是也可以修改一下做HTTP请求等等，方便在做各种HA切换的时候测试影响的时间。

我们经常遇到某个页面缺少一个js之类的情况,一般使用firefox的httpfox或者chrome自带的工具都可以查看那些元素有问题.用python写了一个脚本,分析页面里面引用的元素,然后逐个进行请求,查看是否有的元素不能成功获取.不过使用urllib2有的地方异常处理也做的不完善,目前只是自己简单测试了一些,当熟练一下Python的用法了.目前从perl切换到python还是有很多的地方不是太习惯.尤其是很多列表和字符串的操作有点不一样.
[python]
#!/usr/bin/env python
import urllib2
import gzip
import binascii
import re,sys
import string
from StringIO import StringIO
def gunziptxt(data):
buf = StringIO(data)
of =gzip.GzipFile(fileobj=buf,mode="rb")
outdata=of.read()
return outdata
def http_code(url):
request=urllib2.Request(url,headers={‘User-agent’:"python urllib browser","Accept-Encoding":’gzip’})
try:
response=urllib2.urlopen(request,timeout=5)
return response.getcode()
except urllib2.HTTPError,error:
print "url:",error.reason
return error.code
except urllib2.URLError,error:
print url,error.reason
return 1000
def http_client(url):
request=urllib2.Request(url,headers={‘User-agent’:"python urllib browser","Accept-Encoding":’gzip’})
try:
response=urllib2.urlopen(request,timeout=5)
info=response.info()
data=response.read()
except urllib2.HTTPError,error:
print "%s error:%s" %(url,error.reason)
return None
except urllib2.URLError,error:
print error.reason
return None

if info.get("content-encoding",None) == ‘gzip’:
outdata=gunziptxt(data)
else:
outdata=data
return outdata
def get_src(page):
src_re=re.compile(r’src\s*=\s*["|\’]\s*(https?://[^\"\’]+?)["|\’]’)
if page:
link_urls=src_re.findall(page)
return set(link_urls)
else:
return set()
if len(sys.argv)<2:
print "usage:\n\t",sys.argv[0],"url"
exit(1)
if __name__ == "__main__":
urls=sys.argv[1]
pages=http_client(urls)
if pages:
links=get_src(pages)
else:
exit(1)
for link in links:
code=http_code(link)
if code >399:
print "%s \x1B[1;31m%d\x1B[m"%(link,code)
else:
print "%s \x1B[1;32m%d\x1B[m"%(link,code)
else:
print "pagecheck test"

[/python]
用法比较简单:
python pagecheck.py ‘http://www.baidu.com’

仿照之前自己写的一个perl的多线程的脚本,简单试了下python的多线程模块threading.只能说因为之前习惯了perl的很多用法,刚切换到python非常不习惯,一些小的操作上不熟练.

#!/usr/bin/env python

import threading

import time

import random

import sys

import string

def fun_test(x):

    sleep_time=random.random()*1

    print "args is %s sleep time is %.4f s" % (x,sleep_time)

    time.sleep(sleep_time)

def multi_do(lists,concurrent_max,func):

    concurrent=0

    thread_list=set()

    while( len( lists ) > 0 ):

        if len(thread_list) <= concurrent_max:

            ser=lists.pop()

            pid=threading.Thread(target=func,name=ser,args=(ser,))

            thread_list.add(pid)

            pid.start()

        else:

            alive_threads=set(threading.enumerate())

            join_threads=thread_list-alive_threads

            for job in join_threads:

                print "%s is done"% job.getName()

                job.join()

                thread_list.remove(job)

    while(len(thread_list)>0):

        alive_threads=set(threading.enumerate())

        join_threads=thread_list-alive_threads

        for job in join_threads:

            print "%s is done"% job.getName()

            job.join()

            thread_list.remove(job)

    print "all job have been done"

if __name__ == "__main__":

    try:

        fd=open(sys.argv[1],"r")

        print "open %s ok" % sys.argv[1]

        todo_list=fd.readlines()

        todo_list=map(lambda x: x.strip(),todo_list)

        print "to do is:\n","\n".join(todo_list)

    except IOError:

        print "open file failed"

    fd.close()

    multi_do(todo_list,10,fun_test)

之前用tcpdump抓包的时候，只要是gzip压缩过的数据就没有办法直接还原原始的数据。这段时间学了一下python正好看里面有gzip模块。今天先尝试了一下解压web server返回的压缩过的数据。测试了一下OK

#!/usr/bin/env python

import urllib2

import gzip

import binascii

from StringIO import StringIO

def gunziptxt(data):

    buf = StringIO(data)

    of =gzip.GzipFile(fileobj=buf,mode="rb")

    outdata=of.read()

    return outdata

url="http://127.0.0.1/index.html"

request=urllib2.Request(url,headers={'User-agent':"python urllib browser","Accept-Encoding":'gzip'})

try:

    response=urllib2.urlopen(request,timeout=5)

    data=response.read()

except:

    print "get %s response failed" %url

print "headers:\n",response.info()

if response.info()["content-encoding"] == 'gzip':

    print "http resonse is gzip"

    outdata=gunziptxt(data)

    lbuf=StringIO()

    with gzip.GzipFile(mode='wb',fileobj=lbuf) as inf:

        inf.write(data)

    gziplen=len(lbuf.getvalue())

    print "gzip %d and gunzip %d"%(gziplen,len(outdata))

else:

    print "http resonse is not gzip"

    outdata=data

print "http response:\n",outdata

要获取压缩过的数据的长度只能先借助StingIO，把原始的输出放进去后才能用len来得到。