我公司在某項目組維護的OGG（Oracle Goldengate，一款數據同步軟件），由于局方沒有成熟的監控平臺，一直沒有短信告警，導致我方維護人員每天都需要頻繁登錄每個配置過OGG的主機進行巡檢（目前已達20多臺），工作重復量較大，但一直沒有很好的解決辦法。雖然可通過寫shell腳本，并配置crontab定時任務的方式生成巡檢結果，但還是需要人為地查看。前不久，局方給了一個HTTP接口，說是通過調用這個接口便可實現短信監控，但是沒給調用該接口的具體實現代碼。 正好那段時間我所在的項目組離那邊比較近，接到同事的反饋后，就想如何去實現這個監控的需求。
雖然用shell腳本也可以實現http接口的調用（使用curl命令、GET/POST的方式），但是比較復雜，而且后期不好維護。既然已經放棄了使用shell腳本實現（主要還是自己懶。。。），那么就得用一種編程語言來做了。既要比shell編寫起來簡單直觀，還要便于后期維護，最好還能有時間、相關運行日志的記錄，那么好像非Python莫屬了，哈哈！話不多說，下面就大概說一下實現的具體過程。
HTTP接口的調用方式
HTTP接口的調用格式如下：
http://xxx.xxx.0.2/sms/sendsms.php?Action=MSG_ALERT&DestTermID=手機號碼&MsgContent=告警內容
HTTP接口的好處就是，測試的時候可以在瀏覽器內直接調用。我們先拼湊一條告警信息，并直接在瀏覽器中輸入：
http://xxx.xxx.0.2/sms/sendsms.php?Action=MSG_ALERT&DestTermID=180xxxx3808&MsgContent=OGG_ALERT!
很快就能收到返回的告警信息，說明請求的格式沒問題。
初步監控目標
OGG進程狀態共有RUNNING, STOPPED, ABENDED三種，其中RUNNING為正常運行狀態，STOPPED一般是人為停止的狀態，ABENDED是由于數據同步等原因導致的異常中斷狀態。我們初步的監控目標就是能夠發現異常ABENDED的進程，并定時發送告警短信給相關維護人員（即監控圖中的Status列）。
對于監控頻率的設置，通過詢問局方人員得知，目前需要監控的系統有本地和異地容災兩部分系統。異地容災系統由于有集團考核，及時性要求較高，出現問題必須馬上處理，所以需要配置成每5分鐘檢查一次；而本地系統配置成每小時檢查一次即可。
收集告警信息
根據上面的接口調用方式，那么我們現在要做的可以分以下幾步：
1） 收集OGG的運行狀態，并匯總到一起；
2） 過濾出包含ABENDED的進程及對應的業務系統名稱；
3） 發送包含該業務系統名稱的告警短信給相關人員（注：短信接收人員可能有多個）。
其中的第一條，我們的同事已經通過shell腳本實現好了（還加入了對目錄使用率的告警），并統一ftp到了一臺Windows監控主機的目錄下，幫了我的一個大忙。腳本內容如下：
oggcheck.sh （主要收集OGG進程狀態、目錄使用率情況及超過閾值后的關鍵字over_load）
#!/usr/bin/sh
# 設置OGG目錄使用率閾值
OGG_DIR_THRESHOLD=60
# 獲取并輸出操作系統類型
OS_TYPE=`uname -a| awk '{print $1}'`
echo "OS TYPE: $OS_TYPE"
## 獲取OGG安裝目錄
# 使用查找mgr進程的命令查找OGG安裝目錄
VAR1=`ps -ef|grep mgr.prm | grep -v grep`
# 從左邊開始，刪除PARAMFILE左邊所有的字符
VAR2=${VAR1#*PARAMFILE}
# 從右邊開始，刪除dirprm右邊所有的字符
OGG_HOME=${VAR2%dirprm*}
# 獲取OGG運行狀態
cd $OGG_HOME
echo "info all" | ./ggsci
# 獲取OGG目錄使用率（當前維護系統主要有LINUX, AIX和HP-UX三類）
if [ $OS_TYPE = 'Linux' ];then
VAR3=`df -h $OGG_HOME | sed -n '$p'|cut -d '%' -f 1|awk '{print $NF}'`
elif [ $OS_TYPE = 'AIX' ];then
VAR3=`df -g $OGG_HOME | sed -n '$p'|cut -d '%' -f 1|awk '{print $NF}'`
elif [ $OS_TYPE = 'HP-UX' ];then
VAR3=`bdf $OGG_HOME | sed -n '$p'|cut -d '%' -f 1|awk '{print $NF}'`
else
OGG_DIR_USAGE=100 # 如果操作系統無法識別，則輸出100
echo "`hostname`'s OS TYPE IS UNKNOWN!"
fi
OGG_DIR_USAGE=${VAR3#* }
# 輸出OGG目錄的實際使用率及關鍵字over_load（關鍵字用于給后面程序判斷是否發目錄告警時用）
if [ "$OGG_DIR_USAGE" -ge "$OGG_DIR_THRESHOLD" ];then
echo -e "\n`hostname`'s OGG directory usage is: $OGG_DIR_USAGE%, over_load"
fi
put_alert_file.sh（上傳監控文件）
#!/usr/bin/sh
# 獲取IP地址
IP_ADD=$(awk '{print $1,$2}' /etc/hosts|grep "$(hostname)""$"|awk '{print $1}')
# 設置監控文件絕對路徑
FILE_NAME=/home/oracle/ogg_mon/"$IP_ADD"_`hostname`.txt
# 設置監控文件名稱
FILE_NAME_SHORT="$IP_ADD"_`hostname`.txt
# 將oggcheck.sh的執行結果輸出到$FILE_NAME監控文件中
/bin/sh oggcheck.sh>$FILE_NAME 2>&1
# 設置目錄名稱
dirdate=`date "+%Y%m%d"`
# 使用ftp創建當日目錄，并上傳監控文件
ftp -n <<!
open xxx.2.0.11 50191
user admin admin
pwd
binary
cd ogg
# 按當天日期創建目錄
mkdir ${dirdate}
cd ${dirdate}
lcd /home/oracle/ogg_mon
prompt
put ${FILE_NAME_SHORT}
bye
!
rm -f /home/oracle/ogg_mon/*.txt
echo "finish"
創建配置文件
第二步就主要是對告警信息的處理了。在處理之前，我們需要先定義一個配置文件，里面存放例如需要監控的系統信息、告警接收人的手機號碼、告警內容等。如果以后這些內容有變動，直接在配置文件中修改即可，便于維護。
代碼如下：
settings.py
class Settings():
def __init__(self):
""" 所有告警系統的配置(包括異地災備系統)
配置方法：采用鍵值對的方式，鍵為具體的告警文件名稱(帶擴展名)，值為對應的系統名稱(可自定義).
告警內容將使用自定義系統名稱，建議將IP也一并配置進系統名稱。
例如："xxx.0.0.1_xxxdb1.txt":"xxx.0.0.1_xxxdb1" 表示告警文件為 xxx.0.0.1_xxxdb1.txt，
自定義的系統名稱為 xxx.0.0.1_xxxdb1 ，如果有告警，該系統名稱會顯示在告警短信中。
注：修改配置信息后需重啟程序方可生效。
重啟方式：任務管理器中kill pyw.exe進程，再雙擊程序目錄中的alert.pyw文件即可。可查看本地日志確認是否啟動進程。 """
self.all_systems = {"xxx.0.2.1_erpdb1.txt":"xxx.0.0.1_erpdb1","xxx.0.2.32_pmpdb1.txt":"xxx.0.2.32_pmpdb1",
"xxx.1.20.131_xnyx.txt": "xxx.1.20.131_xnyx", "xxx.1.18.79_ods.txt": "xxx.1.18.79_remote"
}
# 異地災備告警系統配置(災備系統告警閾值為5分鐘，需另單獨配置一份，只配置系統名稱即可，必須跟上面配置的系統名稱一致)
self.remote_systems = ['xnyx', 'xxx.1.18.79_remote']
# 配置接收告警的手機號碼
self.acc_nbr = [181xxxx9607, 152xxxx8230, 152xxxx3315, 133xxxx8640, 177xxxx213, 136xxxx6297]
# 設置本地日志路徑
self.logfile_name = 'D:\\ogg_alert\\log\\info.log'
# 設置告警文件路徑
self.alert_file_path = 'D:\\系統軟件\\oggxj\\'
# 設置告警相關的字符串，用于拼接告警信息
self.url1 = "http://xxx.xxx.0.2/sms/sendsms.php?Action=MSG_ALERT&DestTermID="
self.url2 = "&MsgContent="
self.abend_warning = "_Goldengate:ABENDED"
# 為方便起見，這里只提示目錄使用率大于60%。實際值可以從監控文件中提取
self.dict_warning = "_Goldengate_directory_usage_>_60%!"
按照時間生成日志路徑
按照上面shell腳本的內容，每天都會生成一個YYYYMMDD格式的文件夾，然后再將當日的監控文件放入該文件夾中（會覆蓋上次的文件）。所以我們在程序中也要按照日期動態生成路徑：
functions.py
import time
import os
def time_path(sys_settings):
# 生成固定格式的日期字符串并拼接
time_format = time.strftime('%Y%m%d', time.localtime())
alert_path = sys_settings.alert_file_path + time_format
# 改變當前工作目錄至alert_path
os.chdir(alert_path)
按照要求，監控每隔5分鐘就會掃描一次，所以在每次循環中都調用此函數即可。
處理告警信息
我們采用的方法是遍歷每個告警文件，并處理成列表，如果存在ABENDED或者over_load（OGG目錄告警關鍵字），就使用相關信息生成告警內容，并記錄日志。
functions.py
def get_alert_list(sys_settings):
info_list = []
try:
# 遍歷當前目錄下的文件
for filename in os.listdir():
with open(filename) as file:
lines = file.read()
# 使用空格分割成列表
line_list = lines.split()
if 'ABENDED' in line_list:
# 使用settings.py中配置的系統名稱及告警信息進行拼接
info = sys_settings.all_systems[filename] + sys_settings.abend_warning
info_list.append(info)
# 將處理好的告警內容寫入日志
log_write("WARNING: " + info + "\n", sys_settings)
if 'over_load' in line_list:
info = sys_settings.all_systems[filename] + sys_settings.dict_warning
info_list.append(info)
log_write("WARNING: " + info + "\n", sys_settings)
except PermissionError as e:
log_write("Alert file not found!\n")
""" 返回告警內容 """
return info_list
def log_write(log_info, settings):
# 以固定格式寫入日志內容
with open(settings.logfile_name, 'a') as logfile:
logfile.write("[" + time.strftime('%Y/%m/%d %H:%M:%S', time.localtime()) + "] " + log_info)
發送告警
收集并記錄好告警信息后，我們就可以使用這些信息給指定人員發送短信了。能夠成功發送請求的關鍵是對指定的url調用request.get()方法，從而獲得相應的返回碼（一般的接口在處理完成后都會給請求方返回一個字符或數字作為返回碼，可以進一步寫一個處理返回碼的函數，記錄或者發送返回碼對應的信息，這里為了簡便就省略了）。
因為我們要把每條告警信息都發送給所有的告警接收人員，所以在遍歷告警信息的同時，也需要遍歷電話號碼：
functions.py
import requests
from urllib.error import URLError
def send_info(info_list, sys_settings):
for info in info_list:
for phone_num in sys_settings.acc_nbr:
# 按照http接口請求格式拼接字符串
url = sys_settings.url1 + str(phone_num) + sys_settings.url2 + info
try:
# get_code 為對應的返回碼，這里為了簡便不做處理
get_code = requests.get(url)
# 在日志中記錄將告警發送給哪個電話號碼
log_write(info + " send to: " + str(phone_num) + "\n", sys_settings)
except URLError as e:
# 如果調用失敗，則記錄連接錯誤的日志
log_write("connect to the server failed!\n", sys_settings)
針對不同的業務系統發送告警
有了上面的send_info()函數后，我們只要傳入包含告警信息的列表info_list及配置信息sys_settings，就可以向指定人員發送告警了。但是在正式發送前，針對本地系統和異地容災系統告警頻率的差異，需要分別寫兩個函數分別進行處理：
functions.py
def send_remote_only(alert_list, sys_settings):
# 只發送異地容災系統的告警
if len(alert_list) == 0:
log_write("There is no alert, process finished!\n\t"
"Remote process will restart after 5 min ... ...\n", sys_settings)
else:
for remote_system in sys_settings.remote_systems:
abend_warning = remote_system + sys_settings.abend_warning
dict_warning = remote_system + sys_settings.dict_warning
if abend_warning in alert_list:
send_info([abend_warning], sys_settings)
log_write("Remote process will restart after 5 min ... ...\n", sys_settings)
if dict_warning in alert_list:
send_info([dict_warning], sys_settings)
def send_all(alert_list, sys_settings):
# 發送所有系統的告警
if len(alert_list) == 0:
log_write("There is no alert, process finished!\n\t"
"Local process will restart after 60 min ... ...\n", sys_settings)
else:
send_info(alert_list, sys_settings)
log_write("Local process will restart after 60 min ... ...\n", sys_settings)
使用主程序文件
目前，我們已經基本上完成了所有的功能，只差一個不斷調用這些函數的循環結構以及控制定時發送的策略了。循環結構使用while循環即可，對時間的控制可以使用time.sleep()方法。所以我們需要一個主程序文件來統一控制這些動作。
alert.pyw
from settings import Settings
from timepath import TimePath, time
from functions import send_all, send_remote_only, log_write, get_alert_list, time_path
# 初始化配置信息
sys_settings = Settings()
# 初始化用于計數的變量n
n = 0
# 主循環
while True:
try:
# 調用time_path生成日志路徑
time_path(sys_settings)
except FileNotFoundError as e:
# 如果發生FileNotFoundError，說明當日目錄未生成，記錄到日志中
log_write("Alert Directory not exists!\n", sys_settings)
alert_list = get_alert_list(sys_settings)
""" 如果n == 12，就調用send_all()函數，檢查所有系統
否則就調用send_remote_only()函數，只檢查異地容災系統 """
n += 1
if n == 12:
send_all(alert_list, sys_settings)
n = 0
time.sleep(300)
else:
send_remote_only(alert_list, sys_settings)
# 使用time.sleep()方法，每5分鐘進行一次循環
time.sleep(300)
注意到，我們的主程序文件alert.pyw的擴展名是.pyw，這是因為在Windows環境下，.pyw文件可以通過直接雙擊在后臺運行，不用再打開編輯器執行了。
功能測試
程序運行一段時間后查看記錄的日志：
正常運行時的日志：
目錄使用率超過設定的60%時的日志：
本地系統（到達一小時后統一發送告警）：
異地容災系統（5分鐘掃描并發送告警）:
OGG進程異常ABENDED時的日志：
本地系統：
異地容災系統：
可以看到，所有功能都測試正常。
監控延遲
之前實現的是最基本的對進程狀態的監控。在有些對數據實時性要求較高的系統中，只監控這些還不夠，還需要對延遲時間進行監控。
圖中標出的Lag at Chkpt列即在事務能夠正常提交情況下的延遲總時間，如果有延遲，說明這個進程里同步的表或數據過多，需要做拆分；Time Since Chkpt列是指距離上次事務提交的時間，如果該列出現延遲，說明有大事務，或者有索引缺失導致DML操作執行較慢等情況，也需要關注。所以在監控延遲時，只需關注這兩列即可。
假設我們的目標是當延遲大于5分鐘時輸出告警信息，則需要提取這兩列表示分鐘的項并做判斷。這要比上面的只對Status列做判斷要稍微復雜一些，除了同時取多個列之外（進程名稱、兩個延遲列），還涉及分割、數據類型轉換等操作。所以我們計劃使用numpy數組存放多個列的值，方便后續在時間上的判斷（numpy數組可以進行矩陣運算，比循環的效率要高）。寫日志也換一種方式，使用csv.reader()輸出延遲信息。
為了便于閱讀，我們將延遲信息單獨生成一個文件進行判斷。
oggcheck_new.sh
## 獲取OGG安裝目錄
# 使用查找mgr進程的命令查找OGG安裝目錄
VAR1=`ps -ef|grep mgr.prm | grep -v grep`
# 從左邊開始，刪除PARAMFILE左邊所有的字符
VAR2=${VAR1#*PARAMFILE}
# 從右邊開始，刪除dirprm右邊所有的字符
OGG_HOME=${VAR2%dirprm*}
# 獲取OGG運行狀態
cd $OGG_HOME
# 每次覆蓋寫當時的監控信息
date > /acfs/dump/ogginfo.tmp
echo "info all" | /ogg/ggsci |grep -E "MANAGER|EXTRACT|REPLICAT" >> /acfs/dump/ogginfo.tmp
# 將監控信息同時寫入備份監控文件
date >> /oradump/ogg_tmp/ogginfo.bak
echo "info all" | /ogg/ggsci |grep -E "MANAGER|EXTRACT|REPLICAT" >> /acfs/dump/ogginfo.bak
使用crontab將整個腳本設置成5分鐘執行一次：
*/5 * * * */home/ogg/chk_ogg_status.sh >> /home/ogg/chk_oggstatus.log 2>&1
我們先寫主程序文件，取進程名稱和兩個延遲列的數據，并將其放到numpy數組中。
ogg_lag_mon.py
import sys
sys.path.append('/ogg/dirsql/py_mon') # 進入延遲監控腳本所在目錄
import csv
import numpy as np
alert_file = '/acfs/dump/ogginfo.tmp' # 設置監控文件路徑
alert_his_file = '/acfs/dump/ogginfo.bak' # 設置備份文件路徑
lag_min = 5 # 設置延遲時間
error_log = '/ogg/dirsql/py_mon/lag.log' # 設置本地錯誤日志路徑
lines = list(csv.reader(open(alert_file))) # 將監控文件內容轉換成列表
values = lines[2:] # 取實際的進程信息（前兩個元素為時間和MANAGER進程的狀態）
group_name = [] # 進程名稱(GROUP)
lag_time = [] # 延遲時間(Lag at Chkpt)
ckpt_time = [] # 事務提交延遲時間(Time Since Chkpt)
for value in values:
group_name.append(value[0].split()[2])
lag_time.append(value[0].split(':')[1])
ckpt_time.append(value[0].split(':')[-2])
result_arr = np.array((group_name, lag_time, ckpt_time))
注：將監控文件內容轉成lines列表時，我們按慣例就應該查看一下lines列表里到底裝了些什么（為保持代碼的連貫性，在這里給出分析步驟）：
可以看到，這是一個列表中嵌套列表的結構，而且前兩項不是我們要分析的數據（本例中暫不考慮MANAGER進程的狀態），我們需要的元素的格式為：
[‘EXTRACT RUNNING EXT01 00:00:00 00:00:01 ‘]
這個列表元素只有一項，取進程名稱EXT01時可以按照空格分割，取后兩列時間的分鐘數值時可直接按照分號分割，分別添加到指定的列表中。最后將三個列表作為元素轉換成numpy數組result_arr。
查看result_arr的內容：
>>> result_arr
array([['EXT01', 'EXT02', 'EXT03'],
['00', '08', '07'],
['00', '00', '00']],
dtype='|S8')
這實際上是一個3*3的數組，第0行為進程名稱，第1行為延遲時間，第2行為事務提交延遲時間。可以看到，正好已經有超過lag_min的進程了（第1行超過5的元素值）。我們先定義一個記錄日志的函數alert_writer()，然后再定義一個寫延遲信息的函數write_lag()，write_lag()函數可以調用alert_writer()函數記錄延遲信息：
alert_writer.py
import csv
def alert_writer(fileName, *contents):
with open(fileName, 'a') as file_obj:
writer = csv.writer(file_obj)
writer.writerow(contents)
注：因為使用csv.writer()寫數據時，每個元素之間都以逗號分隔，所以通過使用任意數量的實參*contents，來記錄你想記錄的任何信息。
write_lag.py
import numpy as np
from datetime import datetime
from pyfunc.alert_writer import alert_writer
def write_lag(result_arr, rows, lag_min, alert_his_file):
# 使用np.where()獲得對應rows行中大于lag_min的元素的索引
# rows只需取1, 2行，第0行為進程名稱
index_lag = np.where(result_arr[rows].astype(int) > lag_min)
if len(index_lag) != 0:
for indexes in index_lag:
for index in indexes:
# 將延遲信息寫入備份文件
alert_writer(alert_his_file, 'LAG', result_arr[0][index], result_arr[rows][index] + ' minutes', datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
最后，我們再完善主程序ogg_lag_mon.py中的代碼，將收集到的result_arr數組帶入write_lag()函數中進行判斷，并記錄延遲信息：
ogg_lag_mon.py
import sys
sys.path.append('/ogg/dirsql/py_mon')
from pyfunc.alert_writer import alert_writer
from pyfunc.write_lag import write_lag
import csv
from datetime import datetime
import numpy as np
... ...
try:
for value in values:
group_name.append(value[0].split()[2])
lag_time.append(value[0].split(':')[1])
ckpt_time.append(value[0].split(':')[-2])
result_arr = np.array((group_name, lag_time, ckpt_time))
# row1為Lag at Chkpt列
write_lag(result_arr, rows=1, lag_min, alert_his_file)
# row2為Time Since Chkpt
write_lag(result_arr, rows=2, lag_min, alert_his_file)
except IndexError as e:
alert_writer(error_log, 'Lag exists, please try later!', datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
最后對存在延遲的OGG進行測試，日志中記錄的信息為：
以上只是將進程的相關信息輸出到了日志，同樣也可以組織成統一的格式，利用HTTP接口發送延遲的告警短信。具體方法與上面介紹的相似，不再贅述。有相同環境的同學可以自行實現。實際上，像這種比較成熟的應用程序，輸出的日志格式基本上都是固定的，可以針對平時常用的幾個產品，對不同的格式先分別進行格式化處理，然后統一生成相同格式的內容，無論是發送告警，還是做進一步的數據分析，都是非常方便的。
總結
在我們工程師日常運維的過程中，除了做好每天份內的工作外，有時候也經常會接到客戶提出的一些超出我們職責范圍的需求。這時候，如果直接拒絕客戶，肯定是不明智的。作為技術服務人員，一定要以滿足客戶的要求為目標。這時候，如果我們自己剛好有這樣的能力，或者借助同事、公司二線技術人員的力量來完成這些需求，肯定會給客戶留下更好的印象。
短信群發,彩信群發,短信群發軟件,廣州巨象計算機科技發展有限公司是一家致力于為企業提供互聯網、通訊技術應用服務和解決方案的高科技公司，具有良好的國內外資金和技術背景；是國內最早投入研發企業短信應用和企業網絡電視臺系統的公司之一，業已成為廣東地區最大的移動商務產品與解決方案的提供商和優秀的電訊服務品牌企業。其主要業務有：