运维脚本中的面向对象实践

发表于 分类于 阅读次数： Waline： 本文字数： 1.8k 阅读时长 ≈ 2 分钟

背景

在上一篇文章: 把MongoDB的业务数据采集到Elasticsearch中, 笔者在一番摸索之后, 写了一个简单的脚本, 顺利实现了需求. 后来, 在工作上围绕着这个工作场景衍生出了很多扩展需求, 甚至有其他同事需要一起做二次开发, 于是笔者就按照一些python项目中的惯常操作, 对项目进行了改造.

需求分析以及工作场景

先看看原来的脚本:

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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# -*- coding: utf-8 -*-
'''
用于定时监控MongoDB中ZoneService.zones的变化, 并收录到ES

by rondochen
'''

from pymongo import MongoClient
from elasticsearch import Elasticsearch
from time import sleep
import os
import datetime
import pytz

def main():
    # 连接MongoDB
    myclient = MongoClient(host=os.getenv('mongodb_url'), tz_aware=True)

    # 连接ES
    es = Elasticsearch(hosts=os.getenv('es_url'))

    while True:
        # 记录一个collect_date(采集时间), 便于监控
        collect_date = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
        print(collect_date)
        try:
            zones = myclient['ZoneService']['zones'].find({})
        except:
            print('some thing wrong about mongodb query')
        
        for zone in zones:
            zone['mongo_id'] = str(zone['_id'])
            del zone['_id']
            zone['timestamp'] = collect_date
            try:
                print(es.index(index=os.getenv('zone_monitor_index_name'),body=zone))
                print('----')
            except:
                print('something wrong in es')
        sleep(int(os.getenv('check_interval') or 30))

if __name__ == '__main__':
    # 检查环境变量
    if os.getenv('es_url') and os.getenv('mongodb_url')and os.getenv('zone_monitor_index_name'):
        main()
    else:
        print('require environment variable $es_url and $mongodb_url $zone_monitor_index_name')
        exit()

已实现的需求:

  • 定时扫描MongoDB中的ZoneService.zones

  • 扫描时间支持自定义

  • 把查询到的MongoDB数据保存到ES, 并附加时间戳记录扫描时间

  • 通过环境变量读取必要的运行参数

需要增加的需求:

  • 通过MongoDB集群的changeStream功能, 实时监控MongoDB

  • 可以通过参数自定义MongoDB的监控对象

  • 增加数据库链接检查功能

  • 支持MongoDB搜索语句

  • 尽可能多地复用代码

  • …等等

面向对象设计

大多数时候, 尤其是运维同僚们, 在写一些单一场景的运维脚本的时候, 哪怕是使用了python, 也很少会考虑使用面向对象的思路去编程. 一来需求单一, 也许是几行简单的操作就能完成, 就没必要把代码写得太复杂; 二来很多时候我们在使用现成的库的时候, 已经在使用很多面向对象的方法了, 也不需要自己去写.

面向过程一些痛点

以上面的需求为例子, 也是可以直接面向过程, 写几个函数, 就把功能实现了, 随便起稿一下大概是这样:

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
def checkVariable():
	# 检查必要的环境变量是否存在
	pass

def runCheck(mongodb_client, mongodb_database_name, mongodb_collection_name):
	# 检查mongodb的数据库和集合是否存在
	pass

def connectionCheck(mongodb_client, es_client):
	# 检查数据库连接
	pass

def mongoMonitor(mongodb_url, mongodb_database_name, mongodb_collection_name, check_interval=60, my_query, es_url, es_index_name):
	while True:
		# do something
		pass

def mongoWatcher(mongodb_url, mongodb_database_name, mongodb_collection_name, es_url, es_index_name):
	pass

def main():
	checkVariable()
	# 创建mongodb和es的客户端连接
	connectionCheck(mongodb_client, es_client)
	runCheck(mongodb_client, mongodb_database_name, mongodb_collection_name)
	# 判断业务场景选择mongoMonitor(), 还是mongoMonitor()

if __name__ == '__main__':
	main()

大概是这样写, 其实也能用, 只是在编码的过程, 会有一些不方便的地方.

  • 在ipython下, 反复调试函数的时候, 每次都要思考传参.

  • 就算用了jypyter工具, 可以很方便复地制粘贴/反复修改/调试代码, 但是每次测试函数调用的时候, 都粘贴一串.

  • 后来有了更加复杂的场景, 比如说, 要把查询MongoDB和入库ES分开成两个函数;要两次查询做对比再入库;要一次执行多个不同条件的搜索…杂乱无章的函数会迅速增多.

  • 大家各写各的, 有时候不同函数之间需要用外部变量传参.

  • 开始有了屎山的雏形.

面向对象的摸索

按照笔者的经验, 在开发运维脚本的时候, 如果要设计对象, 一般是由业务场景着手, 让这个业务场景中需要的运行参数作为对象的属性. 以本文的需求为例子, 先设置一个对象MongoMonitor:

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
class MongoMonitor():
	def __init__(self, mongodb_url, mongodb_database_name, mongodb_collection_name, es_url, es_index_name, check_interval=60, my_query="{}"):
		self.mongodb_url = mongodb_url
		self.mongodb_database_name = mongodb_database_name
		self.mongodb_collection_name = mongodb_collection_name
		self.es_url = es_url
		self.es_index_name = es_index_name
		self.check_interval = check_interval
		self.my_query = my_query
		# 在初始化的时候顺便把参数合法性也检查了
		self.connection_check()
		self.run_check()
	def connection_check(self):
		# 检查数据库连接
		pass
	def run_check(self):
		# 检查MongoDB的数据库和集合是否存在
		pass
	def test_my_query(self):
		# 检查查询语句
		pass
	def mongo_collection_to_es_index(self):
		# 把MongoDB的数据输入到ES
		pass
	def monitor_collection(self):
		# 写一个无限循环去定期扫描MongoDB
		while True:
			self.mongo_collection_to_es_index()
			sleep(self.check_interval)

可以看到, 当开始敲下一个class开始, 思路会瞬间清晰很多. 在为对象开发方法的时候, 就无脑使用self, 因为在初始化的时候已经定义好了属性.

而在调试的时候, 只需要初始化一个对象, 即可方便地进行方法调用, 而不需要考虑诸多参数的传递了:

1
2
3
4
5
6
7
8
9
10
11
12
# 一次繁琐, 终身受益 ;-)
job = MongoMonitor(mongodb_url = getenv('mongodb_url'), \
	mongodb_database_name = getenv('mongodb_database_name'), \
	mongodb_collection_name = getenv('mongodb_collection_name'), \
	es_url = getenv('es_url'), \
	es_index_name = getenv('es_index_name'), \
	check_interval = check_interval, \
	my_query = my_query \
	)
# 开工
job.monitor_collection()

对象属性继承

在写完了定期扫描的那个场景的需求后, 着手进行实时监控的类的设计, 然后发现这个类在初始化的时候, 跟类MongoMonitor使用的几乎是一样的属性, 那就可以考虑一下用一个基类衍生两个针对不同业务场景的子类了. 稍微改造一下, 就是这样:

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
32
33
34
35
36
37
38
39
40
# 基类 mongo2es
class mongo2es():
	def __init__(self, mongodb_url, mongodb_database_name, mongodb_collection_name, es_url, es_index_name):
		self.mongodb_url = mongodb_url
		self.mongodb_database_name = mongodb_database_name
		self.mongodb_collection_name = mongodb_collection_name
		self.es_url = es_url
		self.es_index_name = es_index_name
		self.connection_check()
		self.run_check()

	def run_check(self):
		pass

	def connection_check(self):
		pass

# 再写两个子类, 继承基类的属性

class MongoWatcher(mongo2es):
	def __init__(self, mongodb_url, mongodb_database_name, mongodb_collection_name, es_url, es_index_name):
		super().__init__(mongodb_url, mongodb_database_name, mongodb_collection_name, es_url, es_index_name)
	def watch_collection(self):
		# 实时监控mongodb
		pass

class MongoMonitor(mongo2es):
	def __init__(self, mongodb_url, mongodb_database_name, mongodb_collection_name, es_url, es_index_name, check_interval=60, my_query="{}"):
		super().__init__(mongodb_url, mongodb_database_name, mongodb_collection_name, es_url, es_index_name)
		self.check_interval = check_interval
		self.my_query = my_query
	def mongo_collection_to_es_index(self):
		# 把MongoDB的数据输入到ES
		pass
	def monitor_collection(self):
		# 写一个无限循环去定期扫描MongoDB
		while True:
			self.mongo_collection_to_es_index()
			sleep(self.check_interval)

面向对象的便利

在很多时候, 很多原厂的类已经完成了数据库相关业务的抽象化处理, 我们已经不需要重新思考数据库里面的业务逻辑. 所以在运维脚本中, 针对对象设计, 只需要着眼于具体的业务需求.

当我们需要初始化一个类的时候, 只需要通过一次初始化传参, 就可以通过对象内部的不同属性配合方法调用去进行作业, 在编码的时候可以省下很多关于传参的思考. 也顺带在设计对象的时候, 完成了业务逻辑的整理和规范化.

比如说, 如果我临时想要加一个方法, 要输出MongoDB的集群信息, 只需要在类里面加一个方法:

1
2
3
4
5
6
class mongo2es():
	# ...
	def show_mongodb_info(self):
		return self.mongo_client.server_info()
	# ...

如果继续是用面向过程的老方法, 那就是这样, 就是又要重新做一次传参了:

1
2
3
4
def show_mongodb_info(mongodb_url):
	my_client = MongoClient(mongodb_url)
	return my_client.server_info()

另外, 当涉及到团队合作的时候, 当同事们看到一些提前定义好的类出现在公共库中, 大家在二次开发的时候都自然而然地有了尽可能地维护代码优雅的倾向(防止破窗效应).

这里就会产生一个理想的场景: 大家细心维护一个通用的库, 当有需要的时候, 把公共的库fork下来, 再在自己的项目按需开发.

一些心得

当年笔者还在校园的时候, 就知道面向对象, 但大多数时候都只是知道怎么写, 会读. 自己在专业课做作业的时候, 很多时候也就做一些class student()之类的练习. 后来选了运维方向, 很多时候面对的需求都不像实际业务那样, 有清晰的个体去进行对象设计. 就好比本文的例子, 对一个抽象的东西进行对象设计, 不像课本上的例子那么直观. 经验就是, 得多练习, 多思考, 量变总会引起质变.

纸上得来终觉浅, 绝知此事要躬行.

BTW, 这个项目已经传到Github开源了, 有需要的读者可以自便:

RondoChen/mongo2es-watchdog