分布式监控开发 分布式监控开发 01 需求分布式监控开发 02 整体架构设计分布式监控开发 04 客户端开发分布式监控开发 05 历史数据存储

1. 监控系统需求讨论
2. 架构设计
3. 表结构设计
4. 客户端开发
5. 客户端获取本地的监控项
6. 客户端项服务端发送数据
7. 服务端接受并保存监控数据
8. 历史数据优化保存
9. 报警触发流程
10. 监控客户端是否正常汇报数据
11. 前端数据展示（highchart）

 

为什么要做监控？ 

zabbix已经这么强大了，为什么要写一个监控

首先来说说zabbix的痛。

1. 性能瓶颈。zabbix是使用MySQL来存放监控历史数据的。一台机器假设有100个监控项，2000台机器，就是20w监控项，监控系统的数据采集没有高峰低谷，是持续性的，周期性的，一般是一分钟采集一次。机器量越来越大，数据量就越来越大，MySQL的写入逐渐成为瓶颈，业界有一些proxy的方案，也只是治标不治本。zabbix有些数据采集是通过pull的方式，也就是server端主动探测的方式，当目标机器量大了之后，这些pull任务也经常出现积压。
2. zabbix有些易用性问题。比如zabbix的模板是不支持继承的，机器分组也是扁平化的，监控策略不容易复用。zabbix要采集哪些数据，是需要在server端做手工配置的。
3. 有些公司还需要业务的监控，比如某个thrift rpc接口，每分钟调用的cps，latency，某些url的5xx、4xx我们也希望做监控，某个开源软件，比redis、openstack、mysql的一些状态统计数据
4. zabbix的大屏是个问题。虽然有些二次开发的界面非常棒

说了这么多不好的地方，只是在某些big的时候不好而已，我们自己写的话，短期内也是不可能超越zabbix的。那么为什么要手写一套监控呢？

 

1、熟悉IT监控系统的设计原理

自己写的时候肯定有很多事更zabbix相匹配的。

 

2、开发一个简版的类Zabbix监控系统。

为以后团队写监控做准备.zabbix在2K以上数量机器的时候，明显会吃力。小米也正是由于这个自己写了open-falcon。那么如果以后我们遇到大数量的服务器的时候，完全也会基于公司的业务去写一个监控。

那么现在练练手也是完全OK的。

 

 3、掌握自动化开发项目的程序设计思路及架构解藕原则。

    

 

监控系统需求讨论

1.可监控常用系统服务、应用、网络设备等

2.一台主机上可监控多个不同服务、不同服务的监控间隔可不同

3.同一个服务在不同主机上的监控间隔、报警阈值可不同

4.可以批量的给一批主机添加、删除、修改要监控的服务

5.告警级别：

• 不同的服务 因为业务重要程度不同，如果出了问题可以设置不同的报警级别
• 可以指定特定的服务或告警级别的事件通知给特定的用户
• 告警的升级设定

6.历史数据 的存储和优化

• 实现用最少的空间占用量存储最多的有效数据
• 做到1s中之内取出一台主机上所有服务的5年的监控数据（采用redis存取模糊点的方式）

7. 数据可视化，做出简洁美观的用户界面

8.实现单机支持5000+机器监控需求

 

9.实现主动以及被动监控方式

 

10.实现监控服务器的水平扩展

 

 

采用什么架构？

1. Mysql存储主机配置项对应关系
2. redis存储历史数据
3. 支持服务端主动的监控方式（SNMP/PING)以及客户端被动的发送数据
4. 采用HTPP的通信方式

 

采用HTTP好处

1.接口设计简单

2.容易水平扩展做分布式

3.Socket稳定成熟，省去较多的通信维护精力。不用自己从socket底层写起

 

Http特性：

1.短连接

2.无状态

3.安全认证

4.被动通信

 

原文：http://www.cnblogs.com/yangliheng/p/6942060.html

整体架构图

 

 

 

1、客户端每五分钟从服务端获取本机需要监控的服务以及每个服务对应的监控间隔

2、客户端在本地定期执行插件脚本去收集数据

3、客户端每项监控服务收集完数据后立即以POST的方式向服务端提交json格式的数据

4、服务端收到客户端发来的数据，会将数据实时存入redis对应的最新数据中。同时也会优化存储到redis的另外的Key中，

5、服务端在收到客户端数据的同时，也会对客户端的数据进行判断（依据设定好的阈值）以及报警。

6、由于报警是基于服务端收到客户端的数据，所以服务端会启一个监听去监听客户端是否有正常发送数据

7、前端从redis中取出数据并进行图形展示

 

表结构设计

#!_*_coding:utf8_*_
from django.db import models

# Create your models here.




class Host(models.Model):
    name =  models.CharField(max_length=64,unique=True)
    ip_addr =  models.GenericIPAddressField(unique=True)
    host_groups = models.ManyToManyField(\'HostGroup\',blank=True) # A B C
    templates = models.ManyToManyField("Template",blank=True) # A D E
    monitored_by_choices = (
        (\'agent\',\'Agent\'),
        (\'snmp\',\'SNMP\'),
        (\'wget\',\'WGET\'),
    )
    monitored_by = models.CharField(u\'监控方式\',max_length=64,choices=monitored_by_choices)
    status_choices= (
        (1,\'Online\'),
        (2,\'Down\'),
        (3,\'Unreachable\'),
        (4,\'Offline\'),
    )
    status = models.IntegerField(u\'状态\',choices=status_choices,default=1)
    memo = models.TextField(u"备注",blank=True,null=True)

    def __unicode__(self):
        return self.name

class HostGroup(models.Model):
    name =  models.CharField(max_length=64,unique=True)
    templates = models.ManyToManyField("Template",blank=True)
    memo = models.TextField(u"备注",blank=True,null=True)
    def __unicode__(self):
        return self.name

class ServiceIndex(models.Model):
    name = models.CharField(max_length=64)
    key =models.CharField(max_length=64)
    data_type_choices = (
        (\'int\',"int"),
        (\'float\',"float"),
        (\'str\',"string")
    )
    data_type = models.CharField(u\'指标数据类型\',max_length=32,choices=data_type_choices,default=\'int\')
    memo = models.CharField(u"备注",max_length=128,blank=True,null=True)
    def __unicode__(self):
        return "%s.%s" %(self.name,self.key)

class Service(models.Model):
    name = models.CharField(u\'服务名称\',max_length=64,unique=True)
    interval = models.IntegerField(u\'监控间隔\',default=60)
    plugin_name = models.CharField(u\'插件名\',max_length=64,default=\'n/a\')
    items = models.ManyToManyField(\'ServiceIndex\',verbose_name=u"指标列表",blank=True)
    memo = models.CharField(u"备注",max_length=128,blank=True,null=True)

    def __unicode__(self):
        return self.name
    #def get_service_items(obj):
    #    return ",".join([i.name for i in obj.items.all()])

class Template(models.Model):
    name = models.CharField(u\'模版名称\',max_length=64,unique=True)
    services = models.ManyToManyField(\'Service\',verbose_name=u"服务列表")
    triggers = models.ManyToManyField(\'Trigger\',verbose_name=u"触发器列表",blank=True)
    def __unicode__(self):
        return self.name
\'\'\'
class TriggerExpression(models.Model):
    name = models.CharField(u"触发器表达式名称",max_length=64,blank=True,null=True)
    service = models.ForeignKey(Service,verbose_name=u"关联服务")
    service_index = models.ForeignKey(ServiceIndex,verbose_name=u"关联服务指标")
    logic_type_choices = ((\'or\',\'OR\'),(\'and\',\'AND\'))
    logic_type = models.CharField(u"逻辑关系",choices=logic_type_choices,max_length=32,blank=True,null=True)
    left_sibling = models.ForeignKey(\'self\',verbose_name=u"左边条件",blank=True,null=True,related_name=\'left_sibling_condition\' )
    operator_type_choices = ((\'eq\',\'=\'),(\'lt\',\'<\'),(\'gt\',\'>\'))
    operator_type = models.CharField(u"运算符",choices=operator_type_choices,max_length=32)
    data_calc_type_choices = (
        (\'avg\',\'Average\'),
        (\'max\',\'Max\'),
        (\'hit\',\'Hit\'),
        (\'last\',\'Last\'),
    )
    data_calc_func= models.CharField(u"数据处理方式",choices=data_calc_type_choices,max_length=64)
    data_calc_args = models.CharField(u"函数传入参数",help_text=u"若是多个参数,则用,号分开,第一个值是时间",max_length=64)
    threshold = models.IntegerField(u"阈值")

    def __unicode__(self):
        return "%s %s(%s(%s))" %(self.service_index,self.operator_type,self.data_calc_func,self.data_calc_args)
\'\'\'


class TriggerExpression(models.Model):
    #name = models.CharField(u"触发器表达式名称",max_length=64,blank=True,null=True)
    trigger = models.ForeignKey(\'Trigger\',verbose_name=u"所属触发器")
    service = models.ForeignKey(Service,verbose_name=u"关联服务")
    service_index = models.ForeignKey(ServiceIndex,verbose_name=u"关联服务指标")
    specified_index_key = models.CharField(verbose_name=u"只监控专门指定的指标key",max_length=64,blank=True,null=True)
    operator_type_choices = ((\'eq\',\'=\'),(\'lt\',\'<\'),(\'gt\',\'>\'))
    operator_type = models.CharField(u"运算符",choices=operator_type_choices,max_length=32)
    data_calc_type_choices = (
        (\'avg\',\'Average\'),
        (\'max\',\'Max\'),
        (\'hit\',\'Hit\'),
        (\'last\',\'Last\'),
    )
    data_calc_func= models.CharField(u"数据处理方式",choices=data_calc_type_choices,max_length=64)
    data_calc_args = models.CharField(u"函数传入参数",help_text=u"若是多个参数,则用,号分开,第一个值是时间",max_length=64)
    threshold = models.IntegerField(u"阈值")


    logic_type_choices = ((\'or\',\'OR\'),(\'and\',\'AND\'))
    logic_type = models.CharField(u"与一个条件的逻辑关系",choices=logic_type_choices,max_length=32,blank=True,null=True)
    #next_condition = models.ForeignKey(\'self\',verbose_name=u"右边条件",blank=True,null=True,related_name=\'right_sibling_condition\' )
    def __unicode__(self):
        return "%s %s(%s(%s))" %(self.service_index,self.operator_type,self.data_calc_func,self.data_calc_args)
    class Meta:
        pass #unique_together = (\'trigger_id\',\'service\')

class Trigger(models.Model):
    name = models.CharField(u\'触发器名称\',max_length=64)
    #expressions= models.TextField(u"表达式")
    severity_choices = (
        (1,\'Information\'),
        (2,\'Warning\'),
        (3,\'Average\'),
        (4,\'High\'),
        (5,\'Diaster\'),
    )
    #expressions = models.ManyToManyField(TriggerExpression,verbose_name=u"条件表达式")
    severity = models.IntegerField(u\'告警级别\',choices=severity_choices)
    enabled = models.BooleanField(default=True)
    memo = models.TextField(u"备注",blank=True,null=True)

    def __unicode__(self):
        return "<serice:%s, severity:%s>" %(self.name,self.get_severity_display())



class Action(models.Model):
    name =  models.CharField(max_length=64,unique=True)
    host_groups = models.ManyToManyField(\'HostGroup\',blank=True)
    hosts = models.ManyToManyField(\'Host\',blank=True)

    conditions = models.TextField(u\'告警条件\')
    interval = models.IntegerField(u\'告警间隔(s)\',default=300)
    operations = models.ManyToManyField(\'ActionOperation\')

    recover_notice = models.BooleanField(u\'故障恢复后发送通知消息\',default=True)
    recover_subject = models.CharField(max_length=128,blank=True,null=True)
    recover_message = models.TextField(blank=True,null=True)

    enabled = models.BooleanField(default=True)

    def __unicode__(self):
        return self.name

class ActionOperation(models.Model):
    name =  models.CharField(max_length=64)
    step = models.SmallIntegerField(u"第n次告警",default=1)
    action_type_choices = (
        (\'email\',\'Email\'),
        (\'sms\',\'SMS\'),
        (\'script\',\'RunScript\'),
    )
    action_type = models.CharField(u"动作类型",choices=action_type_choices,default=\'email\',max_length=64)
    #notifiers= models.ManyToManyField(host_models.UserProfile,verbose_name=u"通知对象",blank=True)
    def __unicode__(self):
        return self.name


class Maintenance(models.Model):
    name =  models.CharField(max_length=64,unique=True)
    hosts = models.ManyToManyField(\'Host\',blank=True)
    host_groups = models.ManyToManyField(\'HostGroup\',blank=True)
    content = models.TextField(u"维护内容")
    start_time = models.DateTimeField()
    end_time = models.DateTimeField()

    def __unicode__(self):
        return self.name

\'\'\'\'
CPU
    idle 80
    usage  90
    system  30
    user
    iowait  50

memory :
    usage
    free
    swap
    cache
    buffer

load:
    load1
    load 5
    load 15
\'\'\'

 

 

客户端需求

 客户端说白了就是存储一些插件脚本。然后通过服务端传来的配置文件（监控项、监控项对应的监控间隔）、

在本地定时的去执行每个监控项对应的监控插件。

并且汇报数据。

客户端目录

MonitorClient.py

启动main

setting.py

配置文件。记录主机的ID号，服务端的地址、端口、API的url、发送数据超时时间、配置文件更新时间。

 

configs ={
    \'HostID\': 1,
    "Server": "localhost",
    "ServerPort": 9000,
    "urls":{

        \'get_configs\' :[\'api/client/config\',\'get\'],  #acquire all the services will be monitored
        \'service_report\': [\'api/client/service/report/\',\'post\'],

    },
    \'RequestTimeout\':30,
    \'ConfigUpdateInterval\': 300, #5 mins as default

}

 

main.py

启动脚本。

 

import client
class command_handler(object):

    def __init__(self, sys_args):
        self.sys_args = sys_args
        if len(self.sys_args)<2:exit(self.help_msg())
        self.command_allowcator()


    def command_allowcator(self):
        \'\'\'分捡用户输入的不同指令\'\'\'
        print(self.sys_args[1])

        if hasattr(self,self.sys_args[1]):
            func= getattr(self,self.sys_args[1])
            return func()
        else:
            print("command does not exist!")
            self.help_msg()

    def help_msg(self):
        valid_commands = \'\'\'
        start       start monitor client
        stop        stop monitor client

        \'\'\'
        exit(valid_commands)


    def start(self):
        print("going to start the monitor client")
        #exit_flag = False

        Client = client.ClientHandle()
        Client.forever_run()

    def stop(self):
        print("stopping the monitor client")

 

client.py

客户端逻辑处理

 

class ClientHandle(object):
    def __init__(self):
        self.monitored_services = {}

    def load_latest_configs(self):
        \'\'\'
        load the latest monitor configs from monitor server
        :return:
        \'\'\'
        request_type = settings.configs[\'urls\'][\'get_configs\'][1]
        url = "%s/%s" %(settings.configs[\'urls\'][\'get_configs\'][0], settings.configs[\'HostID\'])
        latest_configs = self.url_request(request_type,url)
        latest_configs = json.loads(latest_configs)
        self.monitored_services.update(latest_configs)

    def forever_run(self):
        \'\'\'
        start the client program forever
        :return:
        \'\'\'
        exit_flag = False
        config_last_update_time = 0
        while not exit_flag:
              if time.time() - config_last_update_time > settings.configs[\'ConfigUpdateInterval\']:
                  self.load_latest_configs()
                  print("Loaded latest config:", self.monitored_services)
                  config_last_update_time = time.time()
              #start to monitor services

              for service_name,val in self.monitored_services[\'services\'].items():
                  if len(val) == 2:# means it\'s the first time to monitor
                      self.monitored_services[\'services\'][service_name].append(0)
                  monitor_interval = val[1]
                  last_invoke_time = val[2]
                  if time.time() - last_invoke_time > monitor_interval: #needs to run the plugin
                      print(last_invoke_time,time.time())
                      self.monitored_services[\'services\'][service_name][2]= time.time()
                      #start a new thread to call each monitor plugin
                      t = threading.Thread(target=self.invoke_plugin,args=(service_name,val))
                      t.start()
                      print("Going to monitor [%s]" % service_name)

                  else:
                      print("Going to monitor [%s] in [%s] secs" % (service_name,
                                                                                     monitor_interval - (time.time()-last_invoke_time)))

              time.sleep(1)
    def invoke_plugin(self,service_name,val):
        \'\'\'
        invoke the monitor plugin here, and send the data to monitor server after plugin returned status data each time
        :param val: [pulgin_name,monitor_interval,last_run_time]
        :return:
        \'\'\'
        plugin_name = val[0]
        if hasattr(plugin_api,plugin_name):
            func = getattr(plugin_api,plugin_name)
            plugin_callback = func()
            #print("--monitor result:",plugin_callback)

            report_data = {
                \'client_id\':settings.configs[\'HostID\'],
                \'service_name\':service_name,
                \'data\':json.dumps(plugin_callback)
            }

            request_action = settings.configs[\'urls\'][\'service_report\'][1]
            request_url = settings.configs[\'urls\'][\'service_report\'][0]

            #report_data = json.dumps(report_data)
            print(\'---report data:\',report_data)
            self.url_request(request_action,request_url,params=report_data)
        else:
            print("\033[31;1mCannot find service [%s]\'s plugin name [%s] in plugin_api\033[0m"% (service_name,plugin_name ))
        print(\'--plugin:\',val)


    def url_request(self,action,url,**extra_data):
        \'\'\'
        cope with monitor server by url
        :param action: "get" or "post"
        :param url: witch url you want to request from the monitor server
        :param extra_data: extra parameters needed to be submited
        :return:
        \'\'\'
        abs_url = "http://%s:%s/%s" % (settings.configs[\'Server\'],
                                       settings.configs["ServerPort"],
                                       url)
        if action in  (\'get\',\'GET\'):
            print(abs_url,extra_data)
            try:
                req = urllib2.Request(abs_url)
                req_data = urllib2.urlopen(req,timeout=settings.configs[\'RequestTimeout\'])
                callback = req_data.read()
                #print "-->server response:",callback
                return callback
            except urllib2.URLError as e:
                exit("\033[31;1m%s\033[0m"%e)

        elif action in (\'post\',\'POST\'):
            #print(abs_url,extra_data[\'params\'])
            try:
                data_encode = urllib.urlencode(extra_data[\'params\'])
                req = urllib2.Request(url=abs_url,data=data_encode)
                res_data = urllib2.urlopen(req,timeout=settings.configs[\'RequestTimeout\'])
                callback = res_data.read()
                callback = json.loads(callback)
                print "\033[31;1m[%s]:[%s]\033[0m response:\n%s" %(action,abs_url,callback)
                return callback
            except Exception as e:
                print(\'---exec\',e)
                exit("\033[31;1m%s\033[0m"%e)

 

load_latest_configs这个函数是获取服务端的配置文件

forever_run是以多线程的方式定期执行插件脚本

url_request定义了具体的提交方式

invoke_plugin是从插件脚本中获取数据，添加到字典中调用url_request发送给服务端

 

plugins

这个目录是插件目录

windows和linux子目录存有对应系统的插件脚本

plugin_api.py存着监控项（接受自服务端）和插件脚本的对应关系。

 

1、redis LIST 数据类型 

在说历史数据存储的具体方式前，先说说redis的list类型。

Redis目前支持5种数据类型，分别是：

1. String（字符串）
2. List（列表）
3. Hash（字典）
4. Set（集合）
5. Sorted Set（有序集合）

List（列表）

Redis列表是简单的字符串列表，可以类比到C++中的std::list，简单的说就是一个链表或者说是一个队列。可以从头部或尾部向Redis列表添加元素。列表的最大长度为2^32 - 1，也即每个列表支持超过40亿个元素。

Redis list的实现为一个双向链表，即可以支持反向查找和遍历，更方便操作，不过带来了部分额外的内存开销，Redis内部的很多实现，包括发送缓冲队列等也都是用的这个数据结构。

应用场景

Redis list的应用场景非常多，也是Redis最重要的数据结构之一，比如twitter的关注列表、粉丝列表等都可以用Redis的list结构来实现，再比如有的应用使用Redis的list类型实现一个简单的轻量级消息队列，生产者push，消费者pop/bpop。

相关命令

• BLPOP
  BLPOP key1 [key2 ] timeout 取出并获取列表中的第一个元素，或阻塞，直到有可用
• BRPOP
  BRPOP key1 [key2 ] timeout 取出并获取列表中的最后一个元素，或阻塞，直到有可用
• BRPOPLPUSH
  BRPOPLPUSH source destination timeout 从列表中弹出一个值，它推到另一个列表并返回它;或阻塞，直到有可用
• LINDEX
  LINDEX key index 从一个列表其索引获取对应的元素
• LINSERT
  LINSERT key BEFORE|AFTER pivot value 在列表中的其他元素之后或之前插入一个元素
• LLEN
  LLEN key 获取列表的长度
• LPOP
  LPOP key 获取并取出列表中的第一个元素
• LPUSH
  LPUSH key value1 [value2] 在前面加上一个或多个值的列表
• LPUSHX
  LPUSHX key value 在前面加上一个值列表，仅当列表中存在
• LRANGE
  LRANGE key start stop 从一个列表获取各种元素
• LREM
  LREM key count value 从列表中删除元素
• LSET
  LSET key index value 在列表中的索引设置一个元素的值
• LTRIM
  LTRIM key start stop 修剪列表到指定的范围内
• RPOP
  RPOP key 取出并获取列表中的最后一个元素
• RPOPLPUSH
  RPOPLPUSH source destination 删除最后一个元素的列表，将其附加到另一个列表并返回它
• RPUSH
  RPUSH key value1 [value2] 添加一个或多个值到列表
• RPUSHX
  RPUSHX key value 添加一个值列表，仅当列表中存在

使用示例

 

redis 127.0.0.1:6379> lpush list1 redis
(integer) 1
redis 127.0.0.1:6379> lpush list1 hello
(integer) 2
redis 127.0.0.1:6379> rpush list1 world
(integer) 3
redis 127.0.0.1:6379> llen list1
(integer) 3
redis 127.0.0.1:6379> lrange list1 0 3
1) "hello"
2) "redis"
3) "world"
redis 127.0.0.1:6379> lpop list1
"hello"
redis 127.0.0.1:6379> rpop list1
"world"
redis 127.0.0.1:6379> lrange list1 0 3
1) "redis"

 

 

2、历史数据存储需求

 

3、