这次再来给大家分享一波我工作中用到的几个脚本，主要分为：Python和Shell两个部分。
Python 脚本部分实例：企业微信告警、FTP 客户端、SSH 客户端、Saltstack 客户端、vCenter 客户端、获取域名 ssl 证书过期时间、发送今天的天气预报以及未来的天气趋势图；
Shell 脚本部分实例：SVN 完整备份、Zabbix 监控用户密码过期、构建本地 YUM 以及上篇文章中有读者的需求（负载高时，查出占用比较高的进程脚本并存储或推送通知）；
篇幅有些长，还请大家耐心翻到文末，毕竟有彩蛋。
Python 脚本部分
企业微信告警
此脚本通过企业微信应用，进行微信告警，可用于 Zabbix 监控。
# -*- coding: utf-8 -*-
import requests
import json
class DLF:
def __init__(self, corpid, corpsecret):
self.url="https://qyapi.weixin.qq.com/cgi-bin"
self.corpid=corpid
self.corpsecret=corpsecret
self._token=self._get_token
def _get_token(self):
'''
获取企业微信API接口的access_token
:return:
'''
token_url=self.url + "/gettoken?corpid=%s&corpsecret=%s" %(self.corpid, self.corpsecret)
try:
res=requests.get(token_url).json
token=res['access_token']
return token
except Exception as e:
return str(e)
def _get_media_id(self, file_obj):
get_media_url=self.url + "/media/upload?access_token={}&type=file".format(self._token)
data={"media": file_obj}
try:
res=requests.post(url=get_media_url, files=data)
media_id=res.json['media_id']
return media_id
except Exception as e:
return str(e)
def send_text(self, agentid, content, touser=None, toparty=None):
send_msg_url=self.url + "/message/send?access_token=%s" % (self._token)
send_data={
"touser": touser,
"toparty": toparty,
"msgtype": "text",
"agentid": agentid,
"text": {
"content": content
}
}
try:
res=requests.post(send_msg_url, data=json.dumps(send_data))
except Exception as e:
return str(e)
def send_image(self, agentid, file_obj, touser=None, toparty=None):
media_id=self._get_media_id(file_obj)
send_msg_url=self.url + "/message/send?access_token=%s" % (self._token)
send_data={
"touser": touser,
"toparty": toparty,
"msgtype": "image",
"agentid": agentid,
"image": {
"media_id": media_id
}
}
try:
res=requests.post(send_msg_url, data=json.dumps(send_data))
except Exception as e:
return str(e)
FTP 客户端
通过 ftplib 模块操作 ftp 服务器，进行上传下载等操作。
# -*- coding: utf-8 -*-
from ftplib import FTP
from os import path
import copy
class FTPClient:
def __init__(self, host, user, passwd, port=21):
self.host=host
self.user=user
self.passwd=passwd
self.port=port
self.res={'status': True, 'msg': None}
self._ftp=None
self._login
def _login(self):
'''
登录FTP服务器
:return: 连接或登录出现异常时返回错误信息
'''
try:
self._ftp=FTP
self._ftp.connect(self.host, self.port, timeout=30)
self._ftp.login(self.user, self.passwd)
except Exception as e:
return e
def upload(self, localpath, remotepath=None):
'''
上传ftp文件
:param localpath: local file path
:param remotepath: remote file path
:return:
'''
if not localpath: return 'Please select a local file. '
# 读取本地文件
# fp=open(localpath, 'rb')
# 如果未传递远程文件路径，则上传到当前目录，文件名称同本地文件
if not remotepath:
remotepath=path.basename(localpath)
# 上传文件
self._ftp.storbinary('STOR ' + remotepath, localpath)
# fp.close
def download(self, remotepath, localpath=None):
'''
localpath
:param localpath: local file path
:param remotepath: remote file path
:return:
'''
if not remotepath: return 'Please select a remote file. '
# 如果未传递本地文件路径，则下载到当前目录，文件名称同远程文件
if not localpath:
localpath=path.basename(remotepath)
# 如果localpath是目录的话就和remotepath的basename拼接
if path.isdir(localpath):
localpath=path.join(localpath, path.basename(remotepath))
# 写入本地文件
fp=open(localpath, 'wb')
# 下载文件
self._ftp.retrbinary('RETR ' + remotepath, fp.write)
fp.close
def nlst(self, dir='/'):
'''
查看目录下的内容
:return: 以列表形式返回目录下的所有内容
'''
files_list=self._ftp.nlst(dir)
return files_list
def rmd(self, dir=None):
'''
删除目录
:param dir: 目录名称
:return: 执行结果
'''
if not dir: return 'Please input dirname'
res=copy.deepcopy(self.res)
try:
del_d=self._ftp.rmd(dir)
res['msg']=del_d
except Exception as e:
res['status']=False
res['msg']=str(e)
return res
def mkd(self, dir=None):
'''
创建目录
:param dir: 目录名称
:return: 执行结果
'''
if not dir: return 'Please input dirname'
res=copy.deepcopy(self.res)
try:
mkd_d=self._ftp.mkd(dir)
res['msg']=mkd_d
except Exception as e:
res['status']=False
res['msg']=str(e)
return res
def del_file(self, filename=None):
'''
删除文件
:param filename: 文件名称
:return: 执行结果
'''
if not filename: return 'Please input filename'
res=copy.deepcopy(self.res)
try:
del_f=self._ftp.delete(filename)
res['msg']=del_f
except Exception as e:
res['status']=False
res['msg']=str(e)
return res
def get_file_size(self, filenames=[]):
'''
获取文件大小,单位是字节
判断文件类型
:param filename: 文件名称
:return: 执行结果
'''
if not filenames: return {'msg': 'This is an empty directory'}
res_l=
for file in filenames:
res_d={}
# 如果是目录或者文件不存在就会报错
try:
size=self._ftp.size(file)
type='f'
except:
# 如果是路径的话size显示 - , file末尾加/ （/dir/）
size='-'
type='d'
file=file + '/'
res_d['filename']=file
res_d['size']=size
res_d['type']=type
res_l.append(res_d)
return res_l
def rename(self, old_name=None, new_name=None):
'''
重命名
:param old_name: 旧的文件或者目录名称
:param new_name: 新的文件或者目录名称
:return: 执行结果
'''
if not old_name or not new_name: return 'Please input old_name and new_name'
res=copy.deepcopy(self.res)
try:
rename_f=self._ftp.rename(old_name, new_name)
res['msg']=rename_f
except Exception as e:
res['status']=False
res['msg']=str(e)
return res
def close(self):
'''
退出ftp连接
:return:
'''
try:
# 向服务器发送quit命令
self._ftp.quit
except Exception:
return 'No response from server'
finally:
# 客户端单方面关闭连接
self._ftp.close
SSH 客户端
此脚本仅用于通过 key连接，如需要密码连接，简单修改下即可。
# -*- coding: utf-8 -*-
import paramiko
class SSHClient:
def __init__(self, host, port, user, pkey):
self.ssh_host=host
self.ssh_port=port
self.ssh_user=user
self.private_key=paramiko.RSAKey.from_private_key_file(pkey)
self.ssh=None
self._connect
def _connect(self):
self.ssh=paramiko.SSHClient
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)
try:
self.ssh.connect(hostname=self.ssh_host, port=self.ssh_port, username=self.ssh_user, pkey=self.private_key, timeout=10)
except:
return 'ssh connect fail'
def execute_command(self, command):
stdin, stdout, stderr=self.ssh.exec_command(command)
out=stdout.read
err=stderr.read
return out, err
def close(self):
self.ssh.close
Saltstack 客户端
通过 api 对 Saltstack 服务端进行操作，执行命令。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import requests
import json
import copy
class SaltApi:
"""
定义salt api接口的类
初始化获得token
"""
def __init__(self):
self.url="http://172.85.10.21:8000/"
self.username="saltapi"
self.password="saltapi"
self.headers={"Content-type": "application/json"}
self.params={'client': 'local', 'fun': None, 'tgt': None, 'arg': None}
self.login_url=self.url + "login"
self.login_params={'username': self.username, 'password': self.password, 'eauth': 'pam'}
self.token=self.get_data(self.login_url, self.login_params)['token']
self.headers['X-Auth-Token']=self.token
def get_data(self, url, params):
'''
请求url获取数据
:param url: 请求的url地址
:param params: 传递给url的参数
:return: 请求的结果
'''
send_data=json.dumps(params)
request=requests.post(url, data=send_data, headers=self.headers)
response=request.json
result=dict(response)
return result['return'][0]
def get_auth_keys(self):
'''
获取所有已经认证的key
:return:
'''
data=copy.deepcopy(self.params)
data['client']='wheel'
data['fun']='key.list_all'
result=self.get_data(self.url, data)
try:
return result['data']['return']['minions']
except Exception as e:
return str(e)
def get_grains(self, tgt, arg='id'):
"""
获取系统基础信息
:tgt: 目标主机
:return:
"""
data=copy.deepcopy(self.params)
if tgt:
data['tgt']=tgt
else:
data['tgt']='*'
data['fun']='grains.item'
data['arg']=arg
result=self.get_data(self.url, data)
return result
def execute_command(self, tgt, fun='cmd.run', arg=None, tgt_type='list', salt_async=False):
"""
执行saltstack 模块命令，类似于salt '*' cmd.run 'command'
:param tgt: 目标主机
:param fun: 模块方法 可为空
:param arg: 传递参数 可为空
:return: 执行结果
"""
data=copy.deepcopy(self.params)
if not tgt: return {'status': False, 'msg': 'target host not exist'}
if not arg:
data.pop('arg')
else:
data['arg']=arg
if tgt !='*':
data['tgt_type']=tgt_type
if salt_async: data['client']='local_async'
data['fun']=fun
data['tgt']=tgt
result=self.get_data(self.url, data)
return result
def jobs(self, fun='detail', jid=None):
"""
任务
:param fun: active, detail
:param jod: Job ID
:return: 任务执行结果
"""
data={'client': 'runner'}
data['fun']=fun
if fun=='detail':
if not jid: return {'success': False, 'msg': 'job id is none'}
data['fun']='jobs.lookup_jid'
data['jid']=jid
else:
return {'success': False, 'msg': 'fun is active or detail'}
result=self.get_data(self.url, data)
return resultvCenter 客户端
通过官方 SDK 对 vCenter 进行日常操作，此脚本是我用于 cmdb 平台的，自动获取主机信息，存入数据库。
from pyVim.connect import SmartConnect, Disconnect, SmartConnectNoSSL
from pyVmomi import vim
from asset import models
import atexit
class Vmware:
def __init__(self, ip, user, password, port, idc, vcenter_id):
self.ip=ip
self.user=user
self.password=password
self.port=port
self.idc_id=idc
self.vcenter_id=vcenter_id
def get_obj(self, content, vimtype, name=None):
'''
列表返回,name 可以指定匹配的对象
'''
container=content.viewManager.CreateContainerView(content.rootFolder, vimtype, True)
obj=[ view for view in container.view ]
return obj
def get_esxi_info(self):
# 宿主机信息
esxi_host={}
res={"connect_status": True, "msg": None}
try:
# connect this thing
si=SmartConnectNoSSL(host=self.ip, user=self.user, pwd=self.password, port=self.port, connectionPoolTimeout=60)
except Exception as e:
res['connect_status']=False
try:
res['msg']=("%s Caught vmodl fault : " + e.msg) % (self.ip)
except Exception as e:
res['msg']='%s: connection error' % (self.ip)
return res
# disconnect this thing
atexit.register(Disconnect, si)
content=si.RetrieveContent
esxi_obj=self.get_obj(content, [vim.HostSystem])
for esxi in esxi_obj:
esxi_host[esxi.name]={}
esxi_host[esxi.name]['idc_id']=self.idc_id
esxi_host[esxi.name]['vcenter_id']=self.vcenter_id
esxi_host[esxi.name]['server_ip']=esxi.name
esxi_host[esxi.name]['manufacturer']=esxi.summary.hardware.vendor
esxi_host[esxi.name]['server_model']=esxi.summary.hardware.model
for i in esxi.summary.hardware.otherIdentifyingInfo:
if isinstance(i, vim.host.SystemIdentificationInfo):
esxi_host[esxi.name]['server_sn']=i.identifierValue
# 系统名称
esxi_host[esxi.name]['system_name']=esxi.summary.config.product.fullName
# cpu总核数
esxi_cpu_total=esxi.summary.hardware.numCpuThreads
# 内存总量 GB
esxi_memory_total=esxi.summary.hardware.memorySize / 1024 / 1024 /1024
# 获取硬盘总量 GB
esxi_disk_total=0
for ds in esxi.datastore:
esxi_disk_total +=ds.summary.capacity / 1024 / 1024 /1024
# 默认配置4核8G100G，根据这个配置计算剩余可分配虚拟机
default_configure={
'cpu': 4,
'memory': 8,
'disk':100
}
esxi_host[esxi.name]['vm_host']=
vm_usage_total_cpu=0
vm_usage_total_memory=0
vm_usage_total_disk=0
# 虚拟机信息
for vm in esxi.vm:
host_info={}
host_info['vm_name']=vm.name
host_info['power_status']=vm.runtime.powerState
host_info['cpu_total_kernel']=str(vm.config.hardware.numCPU) + '核'
host_info['memory_total']=str(vm.config.hardware.memoryMB) + 'MB'
host_info['system_info']=vm.config.guestFullName
disk_info=''
disk_total=0
for d in vm.config.hardware.device:
if isinstance(d, vim.vm.device.VirtualDisk):
disk_total +=d.capacityInKB / 1024 /1024
disk_info +=d.deviceInfo.label + ": " + str((d.capacityInKB) / 1024 / 1024) + ' GB' + ','
host_info['disk_info']=disk_info
esxi_host[esxi.name]['vm_host'].append(host_info)
# 计算当前宿主机可用容量：总量 - 已分配的
if host_info['power_status']=='poweredOn':
vm_usage_total_cpu +=vm.config.hardware.numCPU
vm_usage_total_disk +=disk_total
vm_usage_total_memory +=(vm.config.hardware.memoryMB / 1024)
esxi_cpu_free=esxi_cpu_total - vm_usage_total_cpu
esxi_memory_free=esxi_memory_total - vm_usage_total_memory
esxi_disk_free=esxi_disk_total - vm_usage_total_disk
esxi_host[esxi.name]['cpu_info']='Total: %d核, Free: %d核' % (esxi_cpu_total, esxi_cpu_free)
esxi_host[esxi.name]['memory_info']='Total: %dGB, Free: %dGB' % (esxi_memory_total, esxi_memory_free)
esxi_host[esxi.name]['disk_info']='Total: %dGB, Free: %dGB' % (esxi_disk_total, esxi_disk_free)
# 计算cpu 内存 磁盘按照默认资源分配的最小值，即为当前可分配资源
if esxi_cpu_free < 4 or esxi_memory_free < 8 or esxi_disk_free < 100:
free_allocation_vm_host=0
else:
free_allocation_vm_host=int(min(
[
esxi_cpu_free / default_configure['cpu'],
esxi_memory_free / default_configure['memory'],
esxi_disk_free / default_configure['disk']
]
))
esxi_host[esxi.name]['free_allocation_vm_host']=free_allocation_vm_host
esxi_host['connect_status']=True
return esxi_host
def write_to_db(self):
esxi_host=self.get_esxi_info
# 连接失败
if not esxi_host['connect_status']:
return esxi_host
del esxi_host['connect_status']
for machine_ip in esxi_host:
# 物理机信息
esxi_host_dict=esxi_host[machine_ip]
# 虚拟机信息
virtual_host=esxi_host[machine_ip]['vm_host']
del esxi_host[machine_ip]['vm_host']
obj=models.EsxiHost.objects.create(**esxi_host_dict)
obj.save
for host_info in virtual_host:
host_info['management_host_id']=obj.id
obj2=models.virtualHost.objects.create(**host_info)
obj2.save
获取域名 ssl 证书过期时间
用于 zabbix 告警
import re
import sys
import time
import subprocess
from datetime import datetime
from io import StringIO
def main(domain):
f=StringIO
comm=f"curl -Ivs https://{domain} --connect-timeout 10"
result=subprocess.getstatusoutput(comm)
f.write(result[1])
try:
m=re.search('start date: (.*?)\n.*?expire date: (.*?)\n.*?common name: (.*?)\n.*?issuer: CN=(.*?)\n', f.getvalue, re.S)
start_date=m.group(1)
expire_date=m.group(2)
common_name=m.group(3)
issuer=m.group(4)
except Exception as e:
return 999999999
# time 字符串转时间数组
start_date=time.strptime(start_date, "%b %d %H:%M:%S %Y GMT")
start_date_st=time.strftime("%Y-%m-%d %H:%M:%S", start_date)
# datetime 字符串转时间数组
expire_date=datetime.strptime(expire_date, "%b %d %H:%M:%S %Y GMT")
expire_date_st=datetime.strftime(expire_date,"%Y-%m-%d %H:%M:%S")
# 剩余天数
remaining=(expire_date-datetime.now).days
return remaining
if __name__=="__main__":
domain=sys.argv[1]
remaining_days=main(domain)
print(remaining_days)发送今天的天气预报以及未来的天气趋势图
此脚本用于给老婆大人发送今天的天气预报以及未来的天气趋势图，现在微信把网页端禁止了，没法发送到微信了，我是通过企业微信进行通知的，需要把你老婆大人拉到企业微信，无兴趣的小伙伴跳过即可。
# -*- coding: utf-8 -*-
import requests
import json
import datetime
def weather(city):
url="http://wthrcdn.etouch.cn/weather_mini?city=%s" % city
try:
data=requests.get(url).json['data']
city=data['city']
ganmao=data['ganmao']
today_weather=data['forecast'][0]
res="老婆今天是{}\n今天天气概况\n城市: {:<10}\n时间: {:<10}\n高温: {:<10}\n低温: {:<10}\n风力: {:<10}\n风向: {:<10}\n天气: {:<10}\n\n稍后会发送近期温度趋势图，请注意查看。\
".format(
ganmao,
city,
datetime.datetime.now.strftime('%Y-%m-%d'),
today_weather['high'].split[1],
today_weather['low'].split[1],
today_weather['fengli'].split('[')[2].split(']')[0],
today_weather['fengxiang'],today_weather['type'],
)
return {"source_data": data, "res": res}
except Exception as e:
return str(e)
```
+ 获取天气预报趋势图
```python
# -*- coding: utf-8 -*-
import matplotlib.pyplot as plt
import re
import datetime
def Future_weather_states(forecast, save_path, day_num=5):
'''
展示未来的天气预报趋势图
:param forecast: 天气预报预测的数据
:param day_num: 未来几天
:return: 趋势图
'''
future_forecast=forecast
dict={}
for i in range(day_num):
data=
date=future_forecast[i]["date"]
date=int(re.findall("\d+",date)[0])
data.append(int(re.findall("\d+", future_forecast[i]["high"])[0]))
data.append(int(re.findall("\d+", future_forecast[i]["low"])[0]))
data.append(future_forecast[i]["type"])
dict[date]=data
data_list=sorted(dict.items)
date=
high_temperature=
low_temperature=
for each in data_list:
date.append(each[0])
high_temperature.append(each[1][0])
low_temperature.append(each[1][1])
fig=plt.plot(date,high_temperature,"r",date,low_temperature,"b")
current_date=datetime.datetime.now.strftime('%Y-%m')
plt.rcParams['font.sans-serif']=['SimHei']
plt.rcParams['axes.unicode_minus']=False
plt.xlabel(current_date)
plt.ylabel("℃")
plt.legend(["高温", "低温"])
plt.xticks(date)
plt.title("最近几天温度变化趋势")
plt.savefig(save_path)
```
+ 发送到企业微信
```python
# -*- coding: utf-8 -*-
import requests
import json
class DLF:
def __init__(self, corpid, corpsecret):
self.url="https://qyapi.weixin.qq.com/cgi-bin"
self.corpid=corpid
self.corpsecret=corpsecret
self._token=self._get_token
def _get_token(self):
'''
获取企业微信API接口的access_token
:return:
'''
token_url=self.url + "/gettoken?corpid=%s&corpsecret=%s" %(self.corpid, self.corpsecret)
try:
res=requests.get(token_url).json
token=res['access_token']
return token
except Exception as e:
return str(e)
def _get_media_id(self, file_obj):
get_media_url=self.url + "/media/upload?access_token={}&type=file".format(self._token)
data={"media": file_obj}
try:
res=requests.post(url=get_media_url, files=data)
media_id=res.json['media_id']
return media_id
except Exception as e:
return str(e)
def send_text(self, agentid, content, touser=None, toparty=None):
send_msg_url=self.url + "/message/send?access_token=%s" % (self._token)
send_data={
"touser": touser,
"toparty": toparty,
"msgtype": "text",
"agentid": agentid,
"text": {
"content": content
}
}
try:
res=requests.post(send_msg_url, data=json.dumps(send_data))
except Exception as e:
return str(e)
def send_image(self, agentid, file_obj, touser=None, toparty=None):
media_id=self._get_media_id(file_obj)
send_msg_url=self.url + "/message/send?access_token=%s" % (self._token)
send_data={
"touser": touser,
"toparty": toparty,
"msgtype": "image",
"agentid": agentid,
"image": {
"media_id": media_id
}
}
try:
res=requests.post(send_msg_url, data=json.dumps(send_data))
except Exception as e:
return str(e)
+ main脚本
# -*- coding: utf-8 -*-
from plugins.weather_forecast import weather
from plugins.trend_chart import Future_weather_states
from plugins.send_wechat import DLF
import os
# 企业微信相关信息
corpid="xxx"
corpsecret="xxx"
agentid="xxx"
# 天气预报趋势图保存路径
_path=os.path.dirname(os.path.abspath(__file__))
save_path=os.path.join(_path ,'./tmp/weather_forecast.jpg')
# 获取天气预报信息
content=weather("大兴")
# 发送文字消息
dlf=DLF(corpid, corpsecret)
dlf.send_text(agentid=agentid, content=content['res'], toparty='1')
# 生成天气预报趋势图
Future_weather_states(content['source_data']['forecast'], save_path)
# 发送图片消息
file_obj=open(save_path, 'rb')
dlf.send_image(agentid=agentid, toparty='1', file_obj=file_obj)
Shell 脚本部分
SVN 完整备份
通过 hotcopy进行 SVN 完整备份，备份保留 7 天。
#!/bin/bash
# Filename : svn_backup_repos.sh
# Date : 2020/12/14
# Author : JakeTian
# Email : JakeTian@***.com
# Crontab : 59 23 * * * /bin/bash $BASE_PATH/svn_backup_repos.sh >/dev/ 2>&1
# Notes : 将脚本加入crontab中，每天定时执行
# Description: SVN完全备份
set -e
SRC_PATH="/opt/svndata"
DST_PATH="/data/svnbackup"
LOG_FILE="$DST_PATH/logs/svn_backup.log"
SVN_BACKUP_C="/bin/svnadmin hotcopy"
SVN_LOOK_C="/bin/svnlook youngest"
TODAY=$(date +'%F')
cd $SRC_PATH
ALL_REPOS=$(find ./ -maxdepth 1 -type d ! -name 'httpd' -a ! -name 'bak' | tr -d './')
# 创建备份目录，备份脚本日志目录
test -d $DST_PATH || mkdir -p $DST_PATH
test -d $DST_PATH/logs || mkdir $DST_PATH/logs
test -d $DST_PATH/$TODAY || mkdir $DST_PATH/$TODAY
# 备份repos文件
for repo in $ALL_REPOS
do
$SVN_BACKUP_C $SRC_PATH/$repo $DST_PATH/$TODAY/$repo
# 判断备份是否完成
if $SVN_LOOK_C $DST_PATH/$TODAY/$repo;then
echo "$TODAY: $repo Backup Success" >> $LOG_FILE
else
echo "$TODAY: $repo Backup Fail" >> $LOG_FILE
fi
done
# # 备份用户密码文件和权限文件
cp -p authz access.conf $DST_PATH/$TODAY
# 日志文件转储
mv $LOG_FILE $LOG_FILE-$TODAY
# 删除七天前的备份
seven_days_ago=$(date -d "7 days ago" +'%F')
rm -rf $DST_PATH/$seven_days_ago
zabbix 监控用户密码过期
用于 Zabbix 监控 Linux 系统用户（shell 为 /bin/bash 和 /bin/sh）密码过期，密码有效期剩余 7 天触发加自动发现用户。
#!/bin/bash
diskarray=(`awk -F':' '$NF ~ /\/bin\/bash/||/\/bin\/sh/{print $1}' /etc/passwd`)
length=${#diskarray[@]}
printf "{\n"
printf '\t'"\"data\":["
for ((i=0;i<$length;i++))
do
printf '\n\t\t{'
printf "\"{#USER_NAME}\":\"${diskarray[$i]}\"}"
if [ $i -lt $[$length-1] ];then
printf ','
fi
done
printf "\n\t]\n"
printf "}\n"
检查用户密码过期
#!/bin/bash
export LANG=en_US.UTF-8
SEVEN_DAYS_AGO=$(date -d '-7 day' +'%s')
user="$1"
# 将Sep 09, 2018格式的时间转换成unix时间
expires_date=$(sudo chage -l $user | awk -F':' '/Password expires/{print $NF}' | sed -n 's/^ //p')
if [[ "$expires_date" !="never" ]];then
expires_date=$(date -d "$expires_date" +'%s')
if [ "$expires_date" -le "$SEVEN_DAYS_AGO" ];then
echo "1"
else
echo "0"
fi
else
echo "0"
fi
构建本地YUM
通过 rsync 的方式同步 yum，通过 nginx 只做 http yum 站点；
但是 centos6 的镜像最近都不能用了，国内貌似都禁用了，如果找到合适的自行更换地址。
#!/bin/bash
# 更新yum镜像
RsyncCommand="rsync -rvutH -P --delete --delete-after --delay-updates --bwlimit=1000"
DIR="/app/yumData"
LogDir="$DIR/logs"
Centos6Base="$DIR/Centos6/x86_64/Base"
Centos7Base="$DIR/Centos7/x86_64/Base"
Centos6Epel="$DIR/Centos6/x86_64/Epel"
Centos7Epel="$DIR/Centos7/x86_64/Epel"
Centos6Salt="$DIR/Centos6/x86_64/Salt"
Centos7Salt="$DIR/Centos7/x86_64/Salt"
Centos6Update="$DIR/Centos6/x86_64/Update"
Centos7Update="$DIR/Centos7/x86_64/Update"
Centos6Docker="$DIR/Centos6/x86_64/Docker"
Centos7Docker="$DIR/Centos7/x86_64/Docker"
Centos6Mysql5_7="$DIR/Centos6/x86_64/Mysql/Mysql5.7"
Centos7Mysql5_7="$DIR/Centos7/x86_64/Mysql/Mysql5.7"
Centos6Mysql8_0="$DIR/Centos6/x86_64/Mysql/Mysql8.0"
Centos7Mysql8_0="$DIR/Centos7/x86_64/Mysql/Mysql8.0"
MirrorDomain="rsync://rsync.mirrors.ustc.edu.cn"
# 目录不存在就创建
check_dir{
for dir in $*
do
test -d $dir || mkdir -p $dir
done
}
# 检查rsync同步结果
check_rsync_status{
if [ $? -eq 0 ];then
echo "rsync success" >> $1
else
echo "rsync fail" >> $1
fi
}
check_dir $DIR $LogDir $Centos6Base $Centos7Base $Centos6Epel $Centos7Epel $Centos6Salt $Centos7Salt $Centos6Update $Centos7Update $Centos6Docker $Centos7Docker $Centos6Mysql5_7 $Centos7Mysql5_7 $Centos6Mysql8_0 $Centos7Mysql8_0
# Base yumrepo
#$RsyncCommand "$MirrorDomain"/repo/centos/6/os/x86_64/ $Centos6Base >> "$LogDir/centos6Base.log" 2>&1
# check_rsync_status "$LogDir/centos6Base.log"
$RsyncCommand "$MirrorDomain"/repo/centos/7/os/x86_64/ $Centos7Base >> "$LogDir/centos7Base.log" 2>&1
check_rsync_status "$LogDir/centos7Base.log"
# Epel yumrepo
# $RsyncCommand "$MirrorDomain"/repo/epel/6/x86_64/ $Centos6Epel >> "$LogDir/centos6Epel.log" 2>&1
# check_rsync_status "$LogDir/centos6Epel.log"
$RsyncCommand "$MirrorDomain"/repo/epel/7/x86_64/ $Centos7Epel >> "$LogDir/centos7Epel.log" 2>&1
check_rsync_status "$LogDir/centos7Epel.log"
# SaltStack yumrepo
# $RsyncCommand "$MirrorDomain"/repo/salt/yum/redhat/6/x86_64/ $Centos6Salt >> "$LogDir/centos6Salt.log" 2>&1
# ln -s $Centos6Salt/archive/$(ls $Centos6Salt/archive | tail -1) $Centos6Salt/latest
# check_rsync_status "$LogDir/centos6Salt.log"
$RsyncComman "$MirrorDomain"/repo/salt/yum/redhat/7/x86_64/ $Centos7Salt >> "$LogDir/centos7Salt.log" 2>&1
check_rsync_status "$LogDir/centos7Salt.log"
# ln -s $Centos7Salt/archive/$(ls $Centos7Salt/archive | tail -1) $Centos7Salt/latest
# Docker yumrepo
$RsyncCommand "$MirrorDomain"/repo/docker-ce/linux/centos/7/x86_64/stable/ $Centos7Docker >> "$LogDir/centos7Docker.log" 2>&1
check_rsync_status "$LogDir/centos7Docker.log"
# centos update yumrepo
# $RsyncCommand "$MirrorDomain"/repo/centos/6/updates/x86_64/ $Centos6Update >> "$LogDir/centos6Update.log" 2>&1
# check_rsync_status "$LogDir/centos6Update.log"
$RsyncCommand "$MirrorDomain"/repo/centos/7/updates/x86_64/ $Centos7Update >> "$LogDir/centos7Update.log" 2>&1
check_rsync_status "$LogDir/centos7Update.log"
# mysql 5.7 yumrepo
# $RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-5.7-community/el/6/x86_64/ "$Centos6Mysql5_7" >> "$LogDir/centos6Mysql5.7.log" 2>&1
# check_rsync_status "$LogDir/centos6Mysql5.7.log"
$RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-5.7-community/el/7/x86_64/ "$Centos7Mysql5_7" >> "$LogDir/centos7Mysql5.7.log" 2>&1
check_rsync_status "$LogDir/centos7Mysql5.7.log"
# mysql 8.0 yumrepo
# $RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-8.0-community/el/6神奇的工作室/x86_64/ "$Centos6Mysql8_0" >> "$LogDir/centos6Mysql8.0.log" 2>&1
# check_rsync_status "$LogDir/centos6Mysql8.0.log"
$RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-8.0-community/el/7/x86_64/ "$Centos7Mysql8_0" >> "$LogDir/centos7Mysql8.0.log" 2>&1
check_rsync_status "$LogDir/centos7Mysql8.0.log"读者需求解答
负载高时，查出占用比较高的进程脚本并存储或推送通知
这部分内容是上篇 Shell 脚本实例中底部读者留言的需求，如下：
神奇的工作室#!/bin/bash
# 物理cpu个数
physical_cpu_count=$(egrep 'physical id' /proc/cpuinfo | sort | uniq | wc -l)
# 单个物理cpu核数
physical_cpu_cores=$(egrep 'cpu cores' /proc/cpuinfo | uniq | awk '{print $NF}')
# 总核数
total_cpu_cores=$((physical_cpu_count*physical_cpu_cores))
# 分别是一分钟、五分钟、十五分钟负载的阈值，其中有一项超过阈值才会触发
one_min_load_threshold="$total_cpu_cores"
five_min_load_threshold=$(awk 'BEGIN {print '"$total_cpu_cores"' * "0.8"}')
fifteen_min_load_threshold=$(awk 'BEGIN {print '"$total_cpu_cores"' * "0.7"}')
# 分别是分钟、五分钟、十五分钟负载平均值
one_min_load=$(uptime | awk '{print $(NF-2)}' | tr -d ',')
five_min_load=$(uptime | awk '{print $(NF-1)}' | tr -d ',')
fifteen_min_load=$(uptime | awk '{print $NF}' | tr -d ',')
# 获取当前cpu 内存 磁盘io信息，并写入日志文件
# 如果需要发送消息或者调用其他，请自行编写函数即可
get_info{
log_dir="cpu_high_script_log"
test -d "$log_dir" || mkdir "$log_dir"
ps -eo user,pid,%cpu,stat,time,command --sort -%cpu | head -10 > "$log_dir"/cpu_top10.log
ps -eo user,pid,%mem,rss,vsz,stat,time,command --sort -%mem | head -10 > "$log_dir"/mem_top10.log
iostat -dx 1 10 > "$log_dir"/disk_io_10.log
}
export -f get_info
echo "$one_min_load $one_min_load_threshold $five_min_load $five_min_load_threshold $fifteen_min_load $fifteen_min_load_threshold" | \
awk '{ if ($1>=$2 || $3>=$4 || $5>=$6) system("get_info") }'
以上，就是今天分享的全部内容了。
希望大家通过这些案例能够学以致用，结合自身的实际场景进行运用，从而提高自己的工作效率。
- 功能
- 范围
- 透视
- 只要
- 游戏
- 我们
- 一个
- ME
- 比赛
- 操作
- 如果
- 网站
- 脚本
- 直播
- 队友
- 元素
- 版本
- 生命
- 角色
- 领取
- 套装
- 武器
- 火线
- 打击
- 法律
- 环境
- 案件
- 目前
- 释放
- 蓝色
- 可以
- 生化
- 幽灵
- 司令
- 马里
- 系列
- 天堂
- 体验
- 驾驶
- 记者
- 财经
- 贝壳
- 铁路
- 红色
- 发现
- 起义
- 服务
- 鼠标
- 世界
- 核能
- 核电
- 供热
- 大师
- 最高
- 青铜
- 小时
- 加速器
- 推荐
- 阵营
- 求生
- 选手
- 没有
- 使用
- 怎么
- 转移
- 雷达
- 曝光
- 手机
- me
- 英雄
- 模式
- 卡车
- 联盟
- 司机
- 精英
- 和平
- 辅助
- 下载
- 登录
- 分享
- 平台
- 市场
- 程序
- 销售
- 骗局
- 这些
- 业务
- 大家
- 敌人
- 自动
- 技能
- 德国
- 火车
- 游客
- 工具
- http
- 关键
- com
- 产品
- 职业
- 这个
- 奖励
- 钻石
- 活动
- 7月
- 6月
- in
- 注册
- 认证
- 冒险
- 阵容
- 装备
- 水平
- 什么
- 现在
- 其他
- 中国
- 那些
- 制作
- 生物
- 失落
- 免费
- 盔甲
- 恐龙
- 飞机
- 地图
- 特殊
- 属性
- 增加
- 介绍
- 安全
- 自己
- 工作
- 他们
- 测试
- 十年
- 导致
- 情况
- 医院
- 核酸
- 阴性
- 经验
- 非常
- 选择
- 皮肤
- 时代
- 汽车
- 淄博
- 市民
- 房产
- 秋季
- 犯罪
- 嫌疑
- 专案
- 主机
- 韩国
- 软件
- 官方
- 鲨鱼
- 深海
- 潮汐
- 之王
- 配合
- 也是
- 人员
- 内部
- 特种
- 中的
- 图案
- 给你
- 学生
- 中心
- 媒体
- 特权
- 金色
- 地道
- 助手
- 资讯
- 进行
- 病毒
- 这种
- 剧情
- 空投
- 物资
- 考生
- 作弊
- 考试
- 团伙
- 公务员
- 内存
- 你的
- 响应
- 加载
- 战场
- 刺激
- 最新
- 这位
- 几万
- 素质
- 原因
- 9月
- 大全
- 地址
- 钥匙
- 就可以
- 抽取
- 运行
- 技术
- 协议
- 儿子
- 互助
- 治疗
- is
- he
- be
- 空间
- 客厅
- 厨房
- 博士
- Mine
- 迁移
- 邮箱
- 开发
- 但是
- 就是
- 穿越
- 那么
- 还有
- 原价
- 觉得
- 问题
- 态度
- Be
- 当中
- 里面
- 朋友
- 公众
- 头部
- 射击
- 减少
- 子弹
- 更新
- 物品
- 离谱
- 高价
- 东南
- 系统
- 诈骗
- 集团
- 被告人
- 万元
- 故事
- 小说
- 对于
- 手感
- 除了
- 间谍
- 控制
- 提供
- 攻击
- 单机
- 评论
- 研究
- 国家
- 磨损
- 球队
- 赛季
- 以及
- 广州
- 少女
- 中文
- 收费
- 国际
- 航海
- 也会
- 内容
- 会员
- 负责
- 离职
- 容易
- 夏季
- 时候
- 来源
- 叶子
- 任务
- 修改
- 地铁
- 模糊
- 动态
- 逃离
- 赛事
- 联名
- 年轻
- 银行
- 申请
- 熊猫
- 太极
- 管理员
- 权限
- 笑脸
- 最大
- 搭载
- 俱乐部
- 去年
- 赛场
- 获奖
- 荣耀
- 看起来
- 所有
- 数量
- 人称
- 火种
- 小弟
- 聊天
- 监狱
- 手段
- 你们
- 那个
- 福利
- 平滑
- 方法
- 用户
- 肖像
- 位置
- 公差
- 符号
- 还款
- 不少
- 逾期
- 交易
- 估价
- 蜜蜂
- 胜者
- 投诉
- 举报
- 取暖
- 欧洲
- 天然气
- 今年
- 网络
- 应用
- 文件
- 微软
- 某某
- 策划
- 状态
- 真的
- 队伍
- 神奇
- 维护
- 清远
- 科技
- 一起
- 永恒
- 催眠
- 过程
- 高度
- 注意力
- 十分
- 新秀
- 热血
- 江湖
- 麒麟
- 拿到
- 永久
- 同时
- 尼泊尔
- 获得
- 大区
- 资格
- 神仙
- 人物
- 当你
- 遇到
- 到了
- 百度
- Windows
- id
- 竟然
- 算了
- 也就
- 城市
- 建筑
- 分解
- 无限
- 爆竹
- 许可证
- 滑梯
- 有趣
- 然而
- 优化
- 配置
- 厂房
- 商业
- 公园
- 重庆
- 成功
- 年后
- 被盗
- 演员
- 战争
- 僵尸
- 防御
- 区域
- 减速
- 11月
- 挑战
- 幸运
- 戒指
- 凤凰
- 电脑
- 帮忙
- 愿意
- 水军
- 处理
- 工作站
- 专业
- 容量
- 视频
- 夺冠
- GM
- 性能
- 频率
- 瞄准
- 步枪
- 宝珠
- 副本
- 智能
- 美元
- 时间
- 圣诞
- 限定
- 梦境
- 带有
- 不仅
- Win
- 海洋
- 蚂蚁
- 答案
- 呼吸
- 电机
- 售价
- 美国
- 小人
- 百里
- 坐标
- 支付
- ing
- 客户
- for
- 数据
- 存储
- 政府
- 共享
- 出售
- 货源
- 一手
- 低价
- 物理
- 别人
- ??
- 字符
- 设备
- 当然
- 依法
- 信息
- 记得
- 上校
- 哪个
- 笔者
- 一些
- 观战
- 都是
- 过关
- 图片
- 肺癌
- 练习
- 移动
- 这样
- 采购
- 项目
- 升级
- 分配
- 我的
- 效果
- 美化
- 安装
- 巨人
- 漂移
- 全民
- 购物
- 商品
- 天下
- 宝贝
- 经常
- 难怪
- 多变
- 仓库
- 战地
- 国产
- 因为
- 兄弟
- 晴空
- 不过
- 屋顶
- 教程
- 代理
- 水库
- 建设
- 工程
- 淮河
- 支持
- 多多
- 稳定
- 工人
- 友好
- 框架
- 还是
- 定位
- 虚拟
- 公司
- 向日葵
- 酱油
- 异常
- 看到
- 公交
- 防范
- 治安
- 隧道
- 断裂
- 施工
- 龙门
- 硬件
- 企业
- 防止
- 方式
- 出行
- 生殖