让我们来做个django小项目之二

上篇内容我们算是来了一个开场，创建了我们的项目，最后把数据库也建立完成了，这篇我们主要完成后台数据入库的部分，根据我们之前的思路，如果已经能实时获得每个站点性能信息了，如果能将每次获得信息插入到数据库中，这个就完成了我们的入库部分，这里说一下我们的数据库操作部分，要操作数据库，首先我们要跟数据库建立连接，然后进行常规的CRUD操作，操作完毕后再关闭数据库连接，这是一个基本流程，所以为了提高数据库的操作速度，我们可以把一些常用的操作封装成一个类，以下是我门数据库操作的类代码：

class database:
    def __init__(self, dbname=None, dbhost=None):
        self._logger = logger
        if dbname is None:
            self._dbname = DBNAME
        else:
            self._dbname = dbname
        if dbhost is None:
            self._dbhost = DBHOST
        else:
            self._dbhost = dbhost
            
        self._dbuser = DBUSER
        self._dbpassword = DBPWD
        self._dbport = int(DBPORT)
        self._conn = self.connectMySQL()
        
        if(self._conn):
            self._cursor = self._conn.cursor()


    def connectMySQL(self):
        conn = False
        try:
            conn = MySQLdb.connect(host=self._dbhost,user=self._dbuser,passwd=self._dbpassword,db=self._dbname,port=self._dbport)
        except Exception,data:
            self._logger.error("connect database failed, %s" % data)
            conn = False
        return conn


    def fetch_all(self, sql):
        res = ''
        if(self._conn):
            try:
                self._cursor.execute(sql)
                res = self._cursor.fetchall()
            except Exception, data:
                res = False
                self._logger.warn("query database exception, %s" % data)
        return res


    def update(self, sql):
        flag = False
        if(self._conn):
            try:
                self._cursor.execute(sql)
                self._conn.commit()
                flag = True
            except Exception, data:
                flag = False
                self._logger.warn("update database exception, %s" % data)

        return flag

    def close(self):
        if(self._conn):
            try:
                if(type(self._cursor)=='object'):
                    self._cursor.close()
                if(type(self._conn)=='object'):
                    self._conn.close()
            except Exception, data:
                self._logger.warn("close database exception, %s,%s,%s" % (data, type(self._cursor), type(self._conn)))

class database:

def __init__(self, dbname=None, dbhost=None):

self._logger = logger

if dbname is None:

self._dbname = DBNAME

else:

self._dbname = dbname

if dbhost is None:

self._dbhost = DBHOST

else:

self._dbhost = dbhost

self._dbuser = DBUSER

self._dbpassword = DBPWD

self._dbport = int(DBPORT)

self._conn = self.connectMySQL()

if(self._conn):

self._cursor = self._conn.cursor()

def connectMySQL(self):

conn = False

try:

conn = MySQLdb.connect(host=self._dbhost,user=self._dbuser,passwd=self._dbpassword,db=self._dbname,port=self._dbport)

except Exception,data:

self._logger.error("connect database failed, %s" % data)

conn = False

return conn

def fetch_all(self, sql):

res = ''

if(self._conn):

try:

self._cursor.execute(sql)

res = self._cursor.fetchall()

except Exception, data:

res = False

self._logger.warn("query database exception, %s" % data)

return res

def update(self, sql):

flag = False

if(self._conn):

try:

self._cursor.execute(sql)

self._conn.commit()

flag = True

except Exception, data:

flag = False

self._logger.warn("update database exception, %s" % data)

return flag

def close(self):

if(self._conn):

try:

if(type(self._cursor)=='object'):

self._cursor.close()

if(type(self._conn)=='object'):

self._conn.close()

except Exception, data:

self._logger.warn("close database exception, %s,%s,%s" % (data, type(self._cursor), type(self._conn)))

在这个类中我们先定义了__init__方法，指定了数据库连接，用户，端口，用户等信息，接下来是定义了三个方法，分别是建立数据库连接，获取数据内容、更新数据库及最后关闭数据库连接，正题逻辑就是这样，代码没特别的内容。

有了这个数据库类后，我们就可以用上我们之前的pycurl模块去获取每个站点的数据了，每次获取的数据插入到数据库中，因为可能非常多的站点，所以我们要考虑运行效率，所以我们这里使用并发去检查站点数据，采用ThreadPool线程池去实现，全部代码如下：

#!/usr/bin/env python
#coding=utf-8


import MySQLdb
import MySQLdb.cursors
import logging
import requests
import pycurl
import StringIO
import sys
import socket
from multiprocessing.dummy import Pool as ThreadPool 


LOGPATH = '/tmp/conndb.log' 
DBNAME = 'dbname'
DBHOST = 'localhost'
DBUSER = 'root'
DBPWD = 'password'
DBPORT = '3306'

logging.basicConfig(level=logging.WARNING, filename=LOGPATH,filemode='w')

logger = logging.getLogger()

class database:
    def __init__(self, dbname=None, dbhost=None):
        self._logger = logger
        if dbname is None:
            self._dbname = DBNAME
        else:
            self._dbname = dbname
        if dbhost is None:
            self._dbhost = DBHOST
        else:
            self._dbhost = dbhost
            
        self._dbuser = DBUSER
        self._dbpassword = DBPWD
        self._dbport = int(DBPORT)
        self._conn = self.connectMySQL()
        
        if(self._conn):
            self._cursor = self._conn.cursor()


    def connectMySQL(self):
        conn = False
        try:
            conn = MySQLdb.connect(host=self._dbhost,user=self._dbuser,passwd=self._dbpassword,db=self._dbname,port=self._dbport)
        except Exception,data:
            self._logger.error("connect database failed, %s" % data)
            conn = False
        return conn


    def fetch_all(self, sql):
        res = ''
        if(self._conn):
            try:
                self._cursor.execute(sql)
                res = self._cursor.fetchall()
            except Exception, data:
                res = False
                self._logger.warn("query database exception, %s" % data)
        return res


    def update(self, sql):
        flag = False
        if(self._conn):
            try:
                self._cursor.execute(sql)
                self._conn.commit()
                flag = True
            except Exception, data:
                flag = False
                self._logger.warn("update database exception, %s" % data)

        return flag

    def close(self):
        if(self._conn):
            try:
                if(type(self._cursor)=='object'):
                    self._cursor.close()
                if(type(self._conn)=='object'):
                    self._conn.close()
            except Exception, data:
                self._logger.warn("close database exception, %s,%s,%s" % (data, type(self._cursor), type(self._conn)))    

def check(urls):
    #if url not in errorurl:
    global HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD
    url = 'http://' + str(urls[0])
    c = pycurl.Curl()
    c.setopt(pycurl.CONNECTTIMEOUT,5)
    c.setopt(pycurl.TIMEOUT,5)
    c.setopt(pycurl.NOPROGRESS,1)
    c.setopt(pycurl.FORBID_REUSE,1)
    c.setopt(pycurl.MAXREDIRS,3)
    c.setopt(pycurl.DNS_CACHE_TIMEOUT,30)
    c.setopt(c.URL, url)
    try:
        b = StringIO.StringIO()
        c.setopt(c.WRITEFUNCTION, b.write)
        c.perform()
        HTTP_CODE = c.getinfo(pycurl.HTTP_CODE)
        DNS_TIME = c.getinfo(c.NAMELOOKUP_TIME) * 1000
        CONNECT_TIME = c.getinfo(c.CONNECT_TIME) * 1000
        PRETRANSFER_TIME = c.getinfo(c.PRETRANSFER_TIME) * 1000
        STARTTRANSFER_TIME = c.getinfo(c.STARTTRANSFER_TIME) * 1000
        TOTAL_TIME = c.getinfo(c.TOTAL_TIME) * 1000
        SPEED_DOWNLOAD = c.getinfo(c.SPEED_DOWNLOAD)
        b.close()
        c.close()
    except pycurl.error, error:
        pass
        #errno, errstr = error
        #print 'An error occurred: ', errstr
    #insert to db
    status = 'ok'
    sqldb = database()
    sql = "update site_info set http_code=%d, dns_time=%f,connect_time=%f,pretransfer_time=%f,starttransfter_time=%f,total_time=%f,spee
d_download=%f,status='%s' where url='%s'" %(HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLO
AD,status,urls[0]) 
    sqldb.update(sql)
    sqldb.close()
    #print HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD


if __name__ == '__main__':
    sqldb = database()
    sql2 = """select url from site_info"""
    res = sqldb.fetch_all(sql2)
    sqldb.close()
    pool = ThreadPool(3)
    pool.map(check,res)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

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

#!/usr/bin/env python

#coding=utf-8

import MySQLdb

import MySQLdb.cursors

import logging

import requests

import pycurl

import StringIO

import sys

import socket

from multiprocessing.dummy import Pool as ThreadPool

LOGPATH = '/tmp/conndb.log'

DBNAME = 'dbname'

DBHOST = 'localhost'

DBUSER = 'root'

DBPWD = 'password'

DBPORT = '3306'

logging.basicConfig(level=logging.WARNING, filename=LOGPATH,filemode='w')

logger = logging.getLogger()

class database:

def __init__(self, dbname=None, dbhost=None):

self._logger = logger

if dbname is None:

self._dbname = DBNAME

else:

self._dbname = dbname

if dbhost is None:

self._dbhost = DBHOST

else:

self._dbhost = dbhost

self._dbuser = DBUSER

self._dbpassword = DBPWD

self._dbport = int(DBPORT)

self._conn = self.connectMySQL()

if(self._conn):

self._cursor = self._conn.cursor()

def connectMySQL(self):

conn = False

try:

conn = MySQLdb.connect(host=self._dbhost,user=self._dbuser,passwd=self._dbpassword,db=self._dbname,port=self._dbport)

except Exception,data:

self._logger.error("connect database failed, %s" % data)

conn = False

return conn

def fetch_all(self, sql):

res = ''

if(self._conn):

try:

self._cursor.execute(sql)

res = self._cursor.fetchall()

except Exception, data:

res = False

self._logger.warn("query database exception, %s" % data)

return res

def update(self, sql):

flag = False

if(self._conn):

try:

self._cursor.execute(sql)

self._conn.commit()

flag = True

except Exception, data:

flag = False

self._logger.warn("update database exception, %s" % data)

return flag

def close(self):

if(self._conn):

try:

if(type(self._cursor)=='object'):

self._cursor.close()

if(type(self._conn)=='object'):

self._conn.close()

except Exception, data:

self._logger.warn("close database exception, %s,%s,%s" % (data, type(self._cursor), type(self._conn)))

def check(urls):

#if url not in errorurl:

global HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD

url = 'http://' + str(urls[0])

c = pycurl.Curl()

c.setopt(pycurl.CONNECTTIMEOUT,5)

c.setopt(pycurl.TIMEOUT,5)

c.setopt(pycurl.NOPROGRESS,1)

c.setopt(pycurl.FORBID_REUSE,1)

c.setopt(pycurl.MAXREDIRS,3)

c.setopt(pycurl.DNS_CACHE_TIMEOUT,30)

c.setopt(c.URL, url)

try:

b = StringIO.StringIO()

c.setopt(c.WRITEFUNCTION, b.write)

c.perform()

HTTP_CODE = c.getinfo(pycurl.HTTP_CODE)

DNS_TIME = c.getinfo(c.NAMELOOKUP_TIME) * 1000

CONNECT_TIME = c.getinfo(c.CONNECT_TIME) * 1000

PRETRANSFER_TIME = c.getinfo(c.PRETRANSFER_TIME) * 1000

STARTTRANSFER_TIME = c.getinfo(c.STARTTRANSFER_TIME) * 1000

TOTAL_TIME = c.getinfo(c.TOTAL_TIME) * 1000

SPEED_DOWNLOAD = c.getinfo(c.SPEED_DOWNLOAD)

b.close()

c.close()

except pycurl.error, error:

pass

#errno, errstr = error

#print 'An error occurred: ', errstr

#insert to db

status = 'ok'

sqldb = database()

sql = "update site_info set http_code=%d, dns_time=%f,connect_time=%f,pretransfer_time=%f,starttransfter_time=%f,total_time=%f,spee

d_download=%f,status='%s' where url='%s'" %(HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLO

AD,status,urls[0])

sqldb.update(sql)

sqldb.close()

#print HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD

if __name__ == '__main__':

sqldb = database()

sql2 = """select url from site_info"""

res = sqldb.fetch_all(sql2)

sqldb.close()

pool = ThreadPool(3)

pool.map(check,res)

以上就是我们全部的代码内容，pycurl部分我们不在多解释，不理解的小伙伴可以参考之前的文章，Dummy就是多进程模块的克隆文件，唯一不同的是，多进程模块使用的是进程，而dummy则使用线程，pool.map类似python的map函数，需要二个参数第一个参数是一个函数名，第二个参数是个迭代对象，在这里迭代对象我们是从数据库中直接获取的，至于数据库的数据从哪里来的呢？当然是用户输入的，那下篇我们实现下前端输入页面，让用户自己可以增加要监控的站点，这篇就到这里，喜欢的小伙伴请帮忙转发哟。

站点性能监控demo

上篇我们简单介绍了pycurl的基本使用，在组后我们留了几个问题需要去解决，因为最终我们要实现的是一个页面版的展示所有站点性能的demo, 那这篇就是正式开始我们这个小项目，废话不多说，首先我们创建我们的项目，在命令行模式下输入：

django-admin.py startproject web_monitor

1	django-admin.py startproject web_monitor

然后进入web_monitor目录中，创建一个app.

python manage.py startapp  sitemonitor

1	python manage.py startapp sitemonitor

创建完成后我们的目录结构是这样的：

├── manage.py
├── sitemonitor
│ ├── admin.py
│ ├── __init__.py
│ ├── migrations
│ │└── __init__.py
│ ├── models.py
│ ├── tests.py
│ └── views.py
└── web_monitor
    ├── __init__.py
    ├── __init__.pyc
    ├── settings.py
    ├── settings.pyc
    ├── urls.py
    └── wsgi.py

├── manage.py

├── sitemonitor

│ ├── admin.py

│ ├── __init__.py

│ ├── migrations

│ │└── __init__.py

│ ├── models.py

│ ├── tests.py

│ └── views.py

└── web_monitor

├── __init__.py

├── __init__.pyc

├── settings.py

├── settings.pyc

├── urls.py

└── wsgi.py

因为数据要需要入库的，所以第一步先设计我们的表结构，也就是我们的Model文件，进入到sitemonitor目录下，打开models.py文件，写入：

from django.db import models


# Create your models here.

class site_info(models.Model):
    url = models.CharField(null=False, blank=False, db_column="url", db_index=True, unique=True,verbose_name="url", max_length=50)
    port = models.CharField(null=False, blank=False, db_column="port", db_index=True, unique=False,verbose_name="port", max_length=50)
    status = models.CharField(null=True, blank=True, db_column="status", db_index=True, unique=False,verbose_name="status", max_length=
50)
    phone = models.CharField(null=True, blank=True, db_column="phone", db_index=True, unique=False,verbose_name="phone", max_length=50)
    http_code = models.SmallIntegerField(null=True,db_column="http_code",blank=True,db_index=True)
    dns_time = models.DecimalField(null=True,db_column="dns_time", blank=True,db_index=True,max_digits=6,decimal_places=3)
    connect_time = models.DecimalField(null=True,db_column="connect_time",blank=True, db_index=True,max_digits=6,decimal_places=3)
    pretransfer_time = models.DecimalField(null=True,db_column="pretransfer_time",blank=True, db_index=True,max_digits=6,decimal_places
=3)
    starttransfter_time = models.DecimalField(null=True,db_column="starttransfter_time",blank=True, db_index=True,max_digits=6,decimal_
places=3)
    total_time = models.DecimalField(null=True,db_column="total_time", blank=True,db_index=True,max_digits=6,decimal_places=3)
    speed_download = models.DecimalField(null=True,db_column="speed_download",blank=True, db_index=True,max_digits=10,decimal_places=2)

    class Meta:
        db_table = 'site_info'

    def __unicode__(self):
        return self.url

from django.db import models

# Create your models here.

class site_info(models.Model):

url = models.CharField(null=False, blank=False, db_column="url", db_index=True, unique=True,verbose_name="url", max_length=50)

port = models.CharField(null=False, blank=False, db_column="port", db_index=True, unique=False,verbose_name="port", max_length=50)

status = models.CharField(null=True, blank=True, db_column="status", db_index=True, unique=False,verbose_name="status", max_length=

50)

phone = models.CharField(null=True, blank=True, db_column="phone", db_index=True, unique=False,verbose_name="phone", max_length=50)

http_code = models.SmallIntegerField(null=True,db_column="http_code",blank=True,db_index=True)

dns_time = models.DecimalField(null=True,db_column="dns_time", blank=True,db_index=True,max_digits=6,decimal_places=3)

connect_time = models.DecimalField(null=True,db_column="connect_time",blank=True, db_index=True,max_digits=6,decimal_places=3)

pretransfer_time = models.DecimalField(null=True,db_column="pretransfer_time",blank=True, db_index=True,max_digits=6,decimal_places

=3)

starttransfter_time = models.DecimalField(null=True,db_column="starttransfter_time",blank=True, db_index=True,max_digits=6,decimal_

places=3)

total_time = models.DecimalField(null=True,db_column="total_time", blank=True,db_index=True,max_digits=6,decimal_places=3)

speed_download = models.DecimalField(null=True,db_column="speed_download",blank=True, db_index=True,max_digits=10,decimal_places=2)

class Meta:

db_table = 'site_info'

def __unicode__(self):

return self.url

这个文件我们定义了表结构和字段，按上篇我们获取的内容，我们存储的内容有http_code, dns_time,connect_time, pretransfer_time, starttransfter_time,total_time, speed_downlaod字段，url,port,phone这三个字段是为用户输入表单时准备的，这三个后续会用到的时候再说，这里只需要注意因为获取的数字很多都是小数，所以像dns_time我们使用了models.DecimalField字段类型，这个字段类型有2个参数：

max_digits：数字允许的最大位数

decimal_places：小数的最大位数

其它字段类型解释如下：

models.CharField：字符串字段，max_length是必须的参数。

models.SmallIntegerField : 数据库中的字段有：tinyint、smallint、int、bigint. 类似 IntegerField，不过只允许某个取值范围内的整数。(依赖数据库)

然后保存退出，打开site_monitor/settings.py文件，修改如下行：

# Application definition

INSTALLED_APPS = (
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'sitemonitor',    #add here
)

# Application definition

INSTALLED_APPS = (

'django.contrib.admin',

'django.contrib.auth',

'django.contrib.contenttypes',

'django.contrib.sessions',

'django.contrib.messages',

'django.contrib.staticfiles',

'sitemonitor', #add here

)

加入的行我已加注释，不多解释，接下来我们创建数据库表，运行：

python manger.py makemigrations

这个会在我们的app下建立 migrations目录，并记录下所有的关于modes.py的改动，比如0001_initial.py， 但是这个改动还没有作用到数据库文件，
后续如果我们改动了models.py这个文件就要执行这个命令

python manger.py makemigrations

这个会在我们的app下建立 migrations目录，并记录下所有的关于modes.py的改动，比如0001_initial.py，但是这个改动还没有作用到数据库文件，

后续如果我们改动了models.py这个文件就要执行这个命令

接着运行：

python manager.py migrate

将该改动作用到数据库文件，比如产生table，修改字段的类型。

python manager.py migrate

将该改动作用到数据库文件，比如产生table，修改字段的类型。

这时候我们的库表结果就建立完成了，那这样前期准备工作就全部完成了，下一篇我们开始写后台实现数据实时插入库。

python+pycurl检查网站性能指标

对于一个网站运维人员来说，一个网站的性能好坏是大家非常关心的内容，我们知道衡量一个网站的好好多方面，但对运维人员来说站点的性能指标是我们比较关注的内容，从用户角度来说，最直观的感受就是用户户输入网址后站点内容打开速度的快慢，那从一个用户发起请求到服务器，然后服务器响应提供给数据给客户端，了解http协议的小伙伴应该知道，这中间要经过多个步骤才能实现，所以说一个站点如果出现访问速度上的问题，要从整条链路上去查原因而不能只看某一方面，访问示意图如下：

在这个访问过程中我们会关注以下几个指标：

DNS解析所消耗时间
从建立连接到准备传输所消耗的时间
从建立连接到传输开始消耗的时间
传输结束所消耗的总时间
平均下载速度
返回的HTTP状态码

通过对以上几个指标的实时监测，当出现问题时就可以快速分析定位问题，那如何获得这些数据呢，下面就要用到我们的pycurl了，Pycurl包是一个libcurl的Python接口，它是由C语言编写的，速度非常的快，下面看看我们的具体实现：

import pycurl
import StringIO

def check(urls):
    #if url not in errorurl:
    global HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD
    url = 'http://' + str(urls)
    c = pycurl.Curl()
    c.setopt(pycurl.CONNECTTIMEOUT,5)
    c.setopt(pycurl.TIMEOUT,5)
    c.setopt(pycurl.NOPROGRESS,1)
    c.setopt(pycurl.FORBID_REUSE,1)
    c.setopt(pycurl.MAXREDIRS,3)
    c.setopt(pycurl.DNS_CACHE_TIMEOUT,30)
    c.setopt(c.URL, url)
    try:
        b = StringIO.StringIO()
        c.setopt(c.WRITEFUNCTION, b.write)
        c.perform()
        HTTP_CODE = c.getinfo(pycurl.HTTP_CODE)    #HTTP状态码
        DNS_TIME = c.getinfo(c.NAMELOOKUP_TIME) * 1000  #DNS消耗时间
        CONNECT_TIME = c.getinfo(c.CONNECT_TIME) * 1000  #建连消耗时间
        PRETRANSFER_TIME = c.getinfo(c.PRETRANSFER_TIME) * 1000 #建连到准备传输消耗
        STARTTRANSFER_TIME = c.getinfo(c.STARTTRANSFER_TIME) * 1000 #建连到传输消耗
        TOTAL_TIME = c.getinfo(c.TOTAL_TIME) * 1000  #传输结束消耗时间
        SPEED_DOWNLOAD = c.getinfo(c.SPEED_DOWNLOAD) #平均下载速度
        b.close()
        c.close()
    except pycurl.error, error:
        pass
        #errno, errstr = error
        #print 'An error occurred: ', errstr
    return HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD



with open('site.txt','a+') as f:
    f.write('HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD' + '\n')
    res = check('www.mindg.cn')
    f.write(str(res))

import pycurl

import StringIO

def check(urls):

#if url not in errorurl:

global HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD

url = 'http://' + str(urls)

c = pycurl.Curl()

c.setopt(pycurl.CONNECTTIMEOUT,5)

c.setopt(pycurl.TIMEOUT,5)

c.setopt(pycurl.NOPROGRESS,1)

c.setopt(pycurl.FORBID_REUSE,1)

c.setopt(pycurl.MAXREDIRS,3)

c.setopt(pycurl.DNS_CACHE_TIMEOUT,30)

c.setopt(c.URL, url)

try:

b = StringIO.StringIO()

c.setopt(c.WRITEFUNCTION, b.write)

c.perform()

HTTP_CODE = c.getinfo(pycurl.HTTP_CODE) #HTTP状态码

DNS_TIME = c.getinfo(c.NAMELOOKUP_TIME) * 1000 #DNS消耗时间

CONNECT_TIME = c.getinfo(c.CONNECT_TIME) * 1000 #建连消耗时间

PRETRANSFER_TIME = c.getinfo(c.PRETRANSFER_TIME) * 1000 #建连到准备传输消耗

STARTTRANSFER_TIME = c.getinfo(c.STARTTRANSFER_TIME) * 1000 #建连到传输消耗

TOTAL_TIME = c.getinfo(c.TOTAL_TIME) * 1000 #传输结束消耗时间

SPEED_DOWNLOAD = c.getinfo(c.SPEED_DOWNLOAD) #平均下载速度

b.close()

c.close()

except pycurl.error, error:

pass

#errno, errstr = error

#print 'An error occurred: ', errstr

return HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD

with open('site.txt','a+') as f:

f.write('HTTP_CODE,DNS_TIME,CONNECT_TIME,PRETRANSFER_TIME,STARTTRANSFER_TIME,TOTAL_TIME,SPEED_DOWNLOAD' + '\n')

res = check('www.mindg.cn')

f.write(str(res))

脚本里我只获取了一些比较核心的指标，当然pycurl功能不止这些，如果需要它还可以获取更细致的指标，例如：平均上传速度，HTTP头部大小等。有需要的小伙伴可以自己再研究下，脚本不做多解释了，关键行我加了注释，最终获取的数据写入到了文本里，可以定时执行这个脚本来实时获取最新数据，这个脚本对要求不高的场景使用没啥问题，但对我们自动化运维来说只迈出了第一步而已，因为最好的方式是通过一个web页面去实时展示各个站点的情况，便于只观的观看，那要实现一个基本的站点，有几个问题我们需要我们思考去解决：

获取的数据要保存下来，就是有一个入库操作
随着站点的增加，后台运行，为效率考虑必须要用并发执行
前台方面：入库的数据要在前台展示，就需要用ajax
前台展示的页面要能主动自动刷新
前台入口需要提供一个用户添加url地址的表单
当然为好看，我们也要使用boostrap等类似的前端框架。

做完这些才算是基本完成我们最终的目标，在后续的文章里我们会逐一完成上面任务，喜欢的小伙伴还请转发，谢谢。

pip安装pycurl报错解决

错误1：Command “python setup.py egg_info” failed with error code 1 in /tmp/pip-install-u_NBeS/pycurl/

解决：

pip install –upgrade

pip install pycurl==7.43.0

错误2：运行时报错：

ImportError: pycurl: libcurl link-time ssl backend (openssl) is different from compile-time ssl backend (none/other)jiej

解决：
pip uninstall pycurl
export PYCURL_SSL_LIBRARY=nss
easy_install pycurl

ImportError: pycurl: libcurl link-time ssl backend (openssl) is different from compile-time ssl backend (none/other)jiej

解决：

pip uninstall pycurl

export PYCURL_SSL_LIBRARY=nss

easy_install pycurl

用Data-pipeline模式将nginx日志存储到elasticsearch中(续)

接上篇的内容，这篇我们要解决第二个问题，就是如何将我们转换完成的json数据发送到es中去，上篇提到了要存储到es中，我们要使用kafka来做消息队列，实现发布和订阅消息流模式，因为涉及kafka内容，所以我们先说一下Kafka的一些基本知识，然后再看代码不然一是一头雾水。

kafka介绍：

1、什么是kafka?

Kafka是最初由Linkedin公司开发，是一个分布式、支
持分区的（partition）、多副本的（replica），基于
zookeeper协调的分布式消息系统，它的最大的特性就是可
以实时的处理大量数据以满足各种需求场景：比如基于
hadoop的批处理系统、低延迟的实时系统、storm/Spark
流式处理引擎，web/nginx日志、访问日志，消息服务等
等，用scala语言编写，Linkedin于2010年贡献给了Apache
基金会并成为顶级开源 项目。

Kafka是最初由Linkedin公司开发，是一个分布式、支

持分区的（partition）、多副本的（replica），基于

zookeeper协调的分布式消息系统，它的最大的特性就是可

以实时的处理大量数据以满足各种需求场景：比如基于

hadoop的批处理系统、低延迟的实时系统、storm/Spark

流式处理引擎，web/nginx日志、访问日志，消息服务等

等，用scala语言编写，Linkedin于2010年贡献给了Apache

基金会并成为顶级开源项目。

2、kafka特性

- 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，
它的延迟最低只有几毫秒，每个topic可以分多个partition,
consumer group 对partition进行consume操作。
- 可扩展性：kafka集群支持热扩展
- 持久性、可靠性：消息被持久化到本地磁盘，并且支持
数据备份防止数据丢失
- 容错性：允许集群中节点失败（若副本数量为n,则允许n1个节点失败）
- 高并发：支持数千个客户端同时读写

- 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，

它的延迟最低只有几毫秒，每个topic可以分多个partition,

consumer group 对partition进行consume操作。

- 可扩展性：kafka集群支持热扩展

- 持久性、可靠性：消息被持久化到本地磁盘，并且支持

数据备份防止数据丢失

- 容错性：允许集群中节点失败（若副本数量为n,则允许n1个节点失败）

- 高并发：支持数千个客户端同时读写

3、kafka解决了什么问题？

kafka能在系统或应用程序之间构建可靠的用于传输实时数据的管道，
是一种高吞吐量的分布式发布订阅消息系统，
将不同的数据通过不同的topic实现发布订阅，
生产者生成的数据发布到对应的topic中，订阅这个topic的消费者
都可以消费这个topic中的数据。

kafka能在系统或应用程序之间构建可靠的用于传输实时数据的管道，

是一种高吞吐量的分布式发布订阅消息系统，

将不同的数据通过不同的topic实现发布订阅，

生产者生成的数据发布到对应的topic中，订阅这个topic的消费者

都可以消费这个topic中的数据。

4、kafka的一些概念和名词

1）Broker
Kafka集群包含一个或多个服务器，这种服务器被称为broker。broker端不维护数据的消费状态，提升了性能。直接使用磁盘进行存储，线
性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，减少耗性能的创建对象和垃圾回收。
2）Producer
负责发布消息到Kafka broke
3）Consumer
消息消费者，向Kafka broker读取消息的客户端，consumer从broker拉取(pull)数据并进行处理。
4）Topic
每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保
存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
5）Partition
Parition是物理上的概念，每个Topic包含一个或多个Partition.
6）Consumer Group
每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）
7）Topic & Partition 关系
Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。为了使
得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个
Partition的所有消息和索引文件

1）Broker

Kafka集群包含一个或多个服务器，这种服务器被称为broker。broker端不维护数据的消费状态，提升了性能。直接使用磁盘进行存储，线

性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，减少耗性能的创建对象和垃圾回收。

2）Producer

负责发布消息到Kafka broke

3）Consumer

消息消费者，向Kafka broker读取消息的客户端，consumer从broker拉取(pull)数据并进行处理。

4）Topic

每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保

存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）

5）Partition

Parition是物理上的概念，每个Topic包含一个或多个Partition.

6）Consumer Group

每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）

7）Topic & Partition 关系

Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。为了使

得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个

Partition的所有消息和索引文件

除了上面的一些内容外，我也谈一下我自己的一些理解，为什么kafka变得如此流行，很大程度上因为的特性决定的，

比如数据的聚合，怎么理解，就是我们可以把产生数据的系统就做生产者，有多少生产者无所谓，只要建立对应的topic，把数据发送到这个topic中就可以，如果有不同部门向要相同的数据，就去订阅这个topic即可，这样最大的好处是使我们系统结构清晰，同时能减少我们很多重复的工作。

另外一个重要的特性就是高并发性，并支持分布式部署，在一个繁忙的系统中，产生的日志或者其它数据是非常庞大的，要对这些数据进行处理，必须要一个高吞吐量、低延迟的处理系统，Kafka正好满足这个需求，每个topic可以建立一个或多个分区，每个分区你可以简单理解为一个公路上的多个车道，每个车就是数据，因为车道多所以它可以加速数据的传输。

经过上面的介绍，相信大家都对Kafka有了一个基本的了解，那接下来回到我们上面的问题，看如何在我们这个例子中使用：

1、安装kafka，这个因为比较简单就不在费篇幅写了。

2、安装完毕后建立一个topic，命令：

bin/kafka-topics.sh –create –zookeeper 127.0.0.1:2181 –replication-factor 1 –partitions 1 –topic www_logs

3、安装：pip install kafka-python

4、实现生产者类：

from kafka import KafkaProducer
import json


class MyKafka(object):

    def __init__(self, kafka_brokers):
        self.producer = KafkaProducer(
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            bootstrap_servers=kafka_brokers
        )

    def send_page_data(self, json_data, topic):
        result = self.producer.send(topic, key=b'log', value=json_data)
        print("kafka send result: {}".format(result.get()))

from kafka import KafkaProducer

import json

class MyKafka(object):

def __init__(self, kafka_brokers):

self.producer = KafkaProducer(

value_serializer=lambda v: json.dumps(v).encode('utf-8'),

bootstrap_servers=kafka_brokers

)

def send_page_data(self, json_data, topic):

result = self.producer.send(topic, key=b'log', value=json_data)

print("kafka send result: {}".format(result.get()))

代码不是很多，先定义了初始化类的__init__方法，需要两个参数，一个就是broker这个没什么可说的，value_serializer的意思是用于将用户提供的消息值转换为字节，send_page_data()方法是将数据发送给对应的topic。

5、修改下上篇中的脚本，使其实现转换后发送到kafka中，修改如下：

import time
import datetime
import socket
import json
from mykafka import MyKafka   #add


def parse_log_line(line):
    strptime = datetime.datetime.strptime
    hostname = socket.gethostname()
    time = line.split(' ')[3][1::]
    entry = {}
    entry['datetime'] = strptime(
        time, "%d/%b/%Y:%H:%M:%S").strftime("%Y-%m-%d %H:%M")
    entry['source'] = "{}".format(hostname)
    entry['type'] = "www_access"
    entry['log'] = "'{}'".format(line.rstrip())
    return entry


def show_entry(entry):
    temp = ",".join([
        entry['datetime'],
        entry['source'],
        entry['type'],
        entry['log']
    ])
    log_entry = {'log': entry}
    temp = json.dumps(log_entry)
    print("{}".format(temp))
    return temp


def follow(syslog_file):
    syslog_file.seek(0, 2)
    pubsub = MyKafka(["10.51.117.28:9092"])   #add
    while True:
        line = syslog_file.readline()
        if not line:
            time.sleep(0.1)
            continue
        else:
            entry = parse_log_line(line)
            if not entry:
                continue
            json_entry = show_entry(entry)
            print(json_entry)
            pubsub.send_page_data(json_entry, 'www_logs')  #add


f = open("access.log", "rt")
follow(f)

import time

import datetime

import socket

import json

from mykafka import MyKafka #add

def parse_log_line(line):

strptime = datetime.datetime.strptime

hostname = socket.gethostname()

time = line.split(' ')[3][1::]

entry = {}

entry['datetime'] = strptime(

time, "%d/%b/%Y:%H:%M:%S").strftime("%Y-%m-%d %H:%M")

entry['source'] = "{}".format(hostname)

entry['type'] = "www_access"

entry['log'] = "'{}'".format(line.rstrip())

return entry

def show_entry(entry):

temp = ",".join([

entry['datetime'],

entry['source'],

entry['type'],

entry['log']

])

log_entry = {'log': entry}

temp = json.dumps(log_entry)

print("{}".format(temp))

return temp

def follow(syslog_file):

syslog_file.seek(0, 2)

pubsub = MyKafka(["10.51.117.28:9092"]) #add

while True:

line = syslog_file.readline()

if not line:

time.sleep(0.1)

continue

else:

entry = parse_log_line(line)

if not entry:

continue

json_entry = show_entry(entry)

print(json_entry)

pubsub.send_page_data(json_entry, 'www_logs') #add

f = open("access.log", "rt")

follow(f)

增加的行都已注释，不多解释了。

到这我再梳理下我们都干了什么。

第一，我们实现了access.log日志的实时读取，并传唤成json格式。

第二，我们讲转换完毕的数据发送到kafka中，topic名词是www_logs

以上我们都已完成，接下来的问题是要怎么把数据存储到es中，不过先别急，先让我们验证下我们之前的工作是否正确吧，要验证是否可以从topic中读取数据，我们还需要一个消费者程序，为简单验证，我这边实现一个最简单的消费者程序，如下：

from kafka import KafkaConsumer
consumer = KafkaConsumer('www_logs',bootstrap_servers=['127.0.0.1:9092'])
for message in consumer:
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key, message.value))

from kafka import KafkaConsumer

consumer = KafkaConsumer('www_logs',bootstrap_servers=['127.0.0.1:9092'])

for message in consumer:

print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key, message.value))

就是从topic中读取数据，然后打印，如果没问题，就可以证明这条通路是通的，首先在一个终端运行，我们的日志分析程序，结果如下：

{"log": {"datetime": "2016-12-19 12:00", "source": "iZ258ml0cx5Z", "type": "www_access", "log": "'10.1.1.2 - - [19/Dec/2016:12:00:30 +0800] \"GET / HTTP/1.1\" 502 602 \"-\" \"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\"'"}}
{"log": {"datetime": "2016-12-19 12:00", "source": "iZ258ml0cx5Z", "type": "www_access", "log": "'10.1.1.2 - - [19/Dec/2016:12:00:30 +0800] \"GET / HTTP/1.1\" 502 602 \"-\" \"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\"'"}}

{"log": {"datetime": "2016-12-19 12:00", "source": "iZ258ml0cx5Z", "type": "www_access", "log": "'10.1.1.2 - - [19/Dec/2016:12:00:30 +0800] \"GET / HTTP/1.1\" 502 602 \"-\" \"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\"'"}}

行太多，我只截取了2行，因为我们有打印，说明转换是没有问题的，但是否发送到了kafka呢，我们在另一个终端运行我们的消费程序，结果如下：

www_logs:0:21: key=b'log' value=b'"{\\"log\\": {\\"datetime\\": \\"2016-12-19 12:00\\", \\"source\\": \\"iZ258ml0cx5Z\\", \\"type\\": \\"www_access\\", \\"log\\": \\"\'10.1.1.2 - - [19/Dec/2016:12:00:30 +0800] \\\\\\"GET / HTTP/1.1\\\\\\" 502 602 \\\\\\"-\\\\\\" \\\\\\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\\\\\\"\'\\"}}"'
www_logs:0:22: key=b'log' value=b'"{\\"log\\": {\\"datetime\\": \\"2016-12-19 12:00\\", \\"source\\": \\"iZ258ml0cx5Z\\", \\"type\\": \\"www_access\\", \\"log\\": \\"\'10.1.1.2 - - [19/Dec/2016:12:00:30 +0800] \\\\\\"GET / HTTP/1.1\\\\\\" 502 602 \\\\\\"-\\\\\\" \\\\\\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\\\\\\"\'\\"}}"'

www_logs:0:21: key=b'log' value=b'"{\\"log\\": {\\"datetime\\": \\"2016-12-19 12:00\\", \\"source\\": \\"iZ258ml0cx5Z\\", \\"type\\": \\"www_access\\", \\"log\\": \\"\'10.1.1.2 - - [19/Dec/2016:12:00:30 +0800] \\\\\\"GET / HTTP/1.1\\\\\\" 502 602 \\\\\\"-\\\\\\" \\\\\\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\\\\\\"\'\\"}}"'

www_logs:0:22: key=b'log' value=b'"{\\"log\\": {\\"datetime\\": \\"2016-12-19 12:00\\", \\"source\\": \\"iZ258ml0cx5Z\\", \\"type\\": \\"www_access\\", \\"log\\": \\"\'10.1.1.2 - - [19/Dec/2016:12:00:30 +0800] \\\\\\"GET / HTTP/1.1\\\\\\" 502 602 \\\\\\"-\\\\\\" \\\\\\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Geco) Chrome/54.0.2840.99 Safari/537.36\\\\\\"\'\\"}}"'

可以看到我们消费者成功的从kafka中取得了数据，说明这条通路我们已经打通了，这条数据流已经没有问题，那接下来就剩下最后存储到es中的问题了，我们后续再接续，这篇已经很长了。

2018年十月
一	二	三	四	五	六	日
« 九
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30	31