current position:Home>How to send alarm notification to nail in Python?

How to send alarm notification to nail in Python?

2022-02-02 02:22:16 cousin

This is an old watch 12 Month's Day 3 More articles to share ~ Remember to follow me , Insist on sharing programming development 、 Data analysis 、 Learning notes such as machine learning .

Learn cloud server development with Lao Biao Related articles ( If you are reading this series for the first time , It is strongly recommended to study the following articles first ):

Forerunner : Having a server , I'm so cool ?

Alternative items :10 Line of code to write a resume page !

And insecure access Say goodbye, Teach you how to apply for a domain name for free SSL certificate

Linux Pagoda in , The real pagoda ! Detailed tutorial

Finally, there is a website that everyone can visit

How to use Python Send alarm notification to nail ?

One 、 Preface

Not long ago , I saw how Mingge wrote it Python Send warning notice to enterprise wechat , Remember I wrote before Pytho Send data in the specified format to the nailing service , This article will refactor the previous code , To become a : utilize Python Monitoring server data , Then an exception is sent to the user through the nail .

The outline of the project is as follows , The project has been open source to GitHub La , You can directly click to read the original text or visit the browser… Download project .

Project environment description :

  • Python 3.6.8 ( theory 3.6 And above can certainly )
  • Third party dependency Library :
    • requests send out post request , send data
    • psutil Obtain data related to the operation of the operating system
    • apscheduler Set timing task

Because there is less dependence , You can install and use directly in your local environment , You can also create a virtual environment to install and use (Python Recommended for virtual environments pipenv Conduct management ).

After entering the environment , Type below pip Order to install :

pip3 install requests psutil apscheduler
 Copy code 

Two 、 Start thinking

2.1 Create a nail robot

Nail robot personal version is only for group chat , So we need to build a group first , Open the nails , Then create a group chat , Just pull 2 personal , Once created , You can remove these two friends from group chat ( It's a little damaging ?!), Click... In group settings Intelligent group assistant .

Enter the robot management page , Click... After adding robot Three point button , Enter the robot selection page .

We're going down the page , choice Customize robot .

Give the robot a name , Then you need to make security settings , Select the signature method ( Data transmission requires it as a parameter , Avoid safety issues ), Copy the contents . Click finish , that will do complete establish .

Copy Webhook link , Then we just pass Python To this url send out post Request data transmission , You can click Setting instructions Check the relevant functions and configuration methods of the robot .

If you forget the previous settings The signature of the Key or Webhook Address , Sure Group management -> Intelligent group assistant -> Click the corresponding robot Three point button View or modify .

In this way, we have created the nail robot , Next , We just need to make it up Python The code can be .

2.2 Write a simple nail message transmission assistant

2.2.1 Calculate the number and add the signature content , Prepare for automatic message sending

We set up The signature of the Safety protection methods , So before we transmit data , First, we have to calculate the digital signature content of the nailing robot , There are very detailed instructions on the nail document , The calculation methods of various languages are given , Let's just use it for debugging .

Official document address…

You don't have to understand the calculation process ( The following code ), modify secret Just for your own .

import hmac
import hashlib
import base64
import urllib.parse
from time import time

'''  Nail robot digital signature calculation  '''
def get_digest():
    #  Take the millisecond timestamp ,round(x, n)  take x After the decimal point n The result of a , Rounding by default 
    timestamp = str(round(time() * 1000))
    secret = ' Your own signature '
    secret_enc = secret.encode('utf-8')  # utf-8 code 
    string_to_sign = '{}\n{}'.format(timestamp, secret)  #  String formatting splicing 
    string_to_sign_enc = string_to_sign.encode('utf-8')  # utf-8 code 
    hmac_code =, string_to_sign_enc, digestmod=hashlib.sha256).digest()  # HmacSHA256 Algorithmic computation signature 
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))  # Base64 After coding urlEncode
    #  Returns a timestamp and a computed encoded concatenation string , The back is directly spliced to Webhook that will do 
    return f"&timestamp={timestamp}&sign={sign}"
 Copy code 

2.2.2 Post Request to send data , Realize automatic message sending to nail

We go directly to our own Webhook The address to send post Request to transmit data , I used it here markdown data type , There are many other data formats to choose from , Text 、 Jump card 、 Information flow cards, etc , A very rich .

If you are interested, you can go to the official website to check…

It should be noted that , If you need a robot to send messages @ Designated person , Then you need to be in the content ( In the following code @xxxxxx) Also add @ The designated person's nail mobile phone number .

#  Simply send markdown news 
def warning_bot():
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "【 Jane said Python】 Benefits today ",
            "text": """###  Welfare Introduction  @xxxxxx  Please pay attention to :** cousin **, reply 2021, You can get programming learning materials .\n >![]( **[ Lao Biao's personal blog , Online ](**\n """
        "at": {
          "atMobiles": [
              "xxxxxx"  #  want @ The mobile phone number of the object 
    #  Robot link address , Hair post request   Send instructions to the nailing robot 
    webhook_url = ' Yours Webhookurl'
    #  utilize requests send out post request 
    req =, json=data)
 Copy code 

The display effect is as follows ( Above is the chat message bar , Displays the title we specified ; The following is the group chat , Shows markdown The rendered effect ), good-looking , Remember the praise. ( Can't think of , Come here , There are a thousand words ~ That's great Forward With the support of the author ~):

2.3 Write the function of basic data of statistical system

stay Linux We usually use top Command to see CPU Usage situation , Let's look at the following data : process CPU Usage rate 、 Load condition 、 fictitious / Physical memory usage , So in this section we will use Python Get relevant data .

Here we use psutil,Process and System utilities( Process and system utilities ), Used to retrieve the process and system utilization of the system (CPU, Memory , disk , The Internet , sensor ) Cross platform library of information , Through a few lines of code, you can get the relevant data of the local system ~( This article has praised 50, We will update the issue to introduce psutil The article )~

import psutil as psu
import os

'''  ECS basic data   The server has been running for 、 Load status 、CPU Usage rate 、 Running memory usage 、 Physical memory usage  '''
def get_server_info():
    #  Get the basic data of the system 
    #  The server has been running for = The difference between the current time and the server startup time 
    run_times = str(timedelta(seconds=int(time())-int(psu.boot_time())))
    #  System load status ( lately 1、5、15 minute )
    loadavg = [round(i, 2) for i in os.getloadavg()]
    # CPU Usage rate   Test interval 0.3 second 
    cpu_in_use = psu.cpu_percent(interval=0.3)
    #  System running memory utilization 
    #  Memory usage is greater than 80%  Trigger alarm 
    vm_in_use = psu.virtual_memory().percent
    vm_available = round(psu.virtual_memory().available/(1024**3), 2)
    #  System physical storage utilization 
    disk_in_use = psu.disk_usage('/').percent
    disk_free = round(psu.disk_usage('/').free/(1024**3), 2)
    #  You can also add processes 、 Thread and other information , A special article is arranged later to write 
    base_info = f""">  Your ECS is running -{run_times}, The load condition of the machine is ( lately 1、5、15 minute ):{loadavg} -  at present CPU Usage rate is :{cpu_in_use}%, -  The memory utilization rate of the system is :{vm_in_use}%, -  The remaining available running memory is :{vm_available}GiB, -  The system storage memory utilization is :{disk_in_use}%, -  The remaining available storage memory is :{disk_free}GiB <br>**{' machine CPU The utilization rate is normal ' if cpu_in_use<=80 else ' machine CPU High utilization rate , May trigger an alert '}** """
    return base_info, loadavg, cpu_in_use, vm_in_use, disk_in_use
 Copy code 

The code comments are quite clear , Mainly called psutil Some of the built-in methods , We got some basic data of the operating system , In addition, we use os The built-in method of the module getloadavg Acquired , Finally, the data are spliced , When splicing, we used f-string, Like breaking 50 Issue an article to introduce , Here we mainly give you an additional GB and GiB The difference between :

1GB = 1000MB = 1000*1000 KB = 1000*1000*1000 B
1GiB = 1024MiB = 1024*1024 KiB = 1024*1024*1024 B
#  Have a chance to talk to you   interesting 
 Copy code 

2.4 Write a function to monitor data

Next, we need to write a monitoring function , There are only two simple monitoring ( The project is open source , I will when I have time later ), One is to monitor the load , The other is right CPU Monitor usage , Two thresholds are set :CPU Usage rate 70% And the load (1 Minutes and 15 minute )70%*CPU Number , When the threshold value is exceeded, the corresponding alarm content is returned , If there is no problem, return to ok.

 From teacher Ruan Yifeng's blog

'''  Server alert settings   This article is simple first , Set only the load and CPU Usage alert  '''
def get_warning():
    base_info, loadavg, cpu_in_use, vm_in_use, disk_in_use = get_server_info()
    #  First, determine the server load 
    #  Just look at the last minute and the last 15 minutes   should <= 0.7*CPU Number 
    loadavg_max = psu.cpu_count() * 0.7
    loadavg_max = 0.01   #  Test use , Please comment out the formal environment 
    if loadavg[0] >= loadavg_max and loadavg[2] >= loadavg_max:
        warning1 = f'️<font color="#d30c0c">【 Warning 】</font> The current load rate of your ECS is ( lately 1、5、15 minute )-{loadavg}, The load rate has reached <font color="#d30c0c">{round(loadavg[2]/loadavg_max, 2)*100}%</font>, Please check the system for problems in time , It's fine too @ I , send out : Basic information , View the basic information of ECs .'
        return warning1
    if cpu_in_use >= 80:
        warning2 = f'️<font color="#d30c0c">【 Warning 】</font> Your ECS is currently CPU Usage rate is <font color="#d30c0c">{cpu_in_use}%</font>, Please check the system for problems in time , It's fine too @ I , send out : Basic information , View the basic information of ECs .'
        return warning2
    return 'ok'
 Copy code 

Originally I wanted to write a function , That is to check the data from time to time (@ Get off the robot , Just give the corresponding instruction ), However, it is found here that this function requires a new enterprise robot , Share it with you next time ~

2.5 Write scheduled tasks

stay Linux above , We can directly use the pagoda panel mentioned before to set timing tasks , You can see Linux Pagoda in , The real pagoda ! Detailed tutorial , Here we write our own scheduled tasks , We used apscheduler This third-party library , You can write another long article about this library , I'm not going to expand on that , This article likes if broken 100, I'll arrange... Next week ( Old watch online humble praise !).

from apscheduler.schedulers.blocking import BlockingScheduler

''' 1、 Every day in the morning 9:00  Send the server information to the nail group  '''
def every_day_nine():
    message = get_server_info()[0]
    title = ' Basic server information '
    warning_bot(message, title)

''' 2、 Always alert ( Every time 30 Check once per second ) '''
def every_seconds_30():
    warning = get_warning()
    if warning != 'ok':
        title = '【️ Warning 】 Server failure '
        warning_bot(warning, title)

''' 3、@ robot , Automatic Q & A settings   The next arrangement , You need to create another enterprise robot   It's different from the process of group chat robot ~ '''
#  choice BlockingScheduler Scheduler 
sched = BlockingScheduler(timezone='Asia/Shanghai')

# job_every_nine  Every day in the morning 9 Click to run once   Daily sending 
sched.add_job(every_day_nine, 'cron', hour=21, minute=55)

# every_seconds_30  Every time 30s Do it once   Data monitoring 
sched.add_job(every_seconds_30, 'interval', seconds=3)

#  Start timing task 

 Copy code 

add_job Set... In scheduled tasks interval Indicates a periodic trigger , Like every minute ;cron yes apscheduler The most powerful trigger in , It can be triggered at a certain time of each month , Like every morning 9:00.

2.6 function , Let's see the effect

At present, they are all automatically triggered , And then send a message , The first is the regular morning every day 9:00 Basic information of sending server .

The second is every 30s Perform a server data detection (CPU Usage and load ), When the data exceeds the threshold , Trigger alarm , Send a message reminder .

2.7 Create daemons for programs

After the above, we have completed the function development , But you will find , Once we close the program , The alert monitoring service will also stop , So we need to create a daemon to protect our process .

Take myself for example , After we log in to the pagoda panel , Get into /etc/systemd/system Under the folder , Create a new one ding_bot.service file , And write the following :

Description=Dingding Bot service

ExecStart=/usr/bin/python3 /root/Project/Little_project/DingdingBot/

 Copy code 

A simple explanation Service The meaning set in ,Type=forking Indicates that after the program starts , Will run in the background ;ExecStart Specific execution instructions of the service ( perform File can );KillMode=process It means that when the service stops, it will also kill the main process of the program ;Restart=on-failure Indicates that the program exits due to an accident in the system , The program restarts automatically .

After saving the file , We can start the daemon by directly executing the following instructions in the terminal , After running, it will enter the daemon state , We can press ctrl+c sign out , Will not affect the daemon :

systemctl start ding_bot
 Copy code 

After the code is modified , The daemon needs to be restarted , Modify the code to take effect , The restart command is as follows :

systemctl restart ding_bot
 Copy code 

If you don't want to set up this daemon , perform stop The command can stop the service( The program will also stop ), The instructions are as follows :

systemctl stop ding_bot
 Copy code 

About daemons system Other relevant instructions and operations can be searched by yourself , You can also communicate in the message area , Let's start with another tweet ~ Like breaking 50, arrange ~

3、 ... and 、 Next up

The overall effect feels pretty good , Hey, hey, hey , We will continue to optimize later , What additional ideas and needs do you have , You can also leave a message .

In addition, a related article is expected to be published this week :@ robot Enter keywords It can be realized by automatic reply function , That requires the use of enterprise robots , The process is a little troublesome ~ Try to come out this week ~ In addition, the psutil and apscheduler library 、 Daemon details , I'll share with you when I have time and scenes , Or this tweet likes 50+50+50~

Learn cloud server development with Lao Biao Related articles ( If you are reading this series for the first time , It is strongly recommended to study the following articles first ):

Forerunner : Having a server , I'm so cool ?

Alternative items :10 Line of code to write a resume page !

And insecure access Say goodbye, Teach you how to apply for a domain name for free SSL certificate

Linux Pagoda in , The real pagoda ! Detailed tutorial

Finally, there is a website that everyone can visit

It's not easy to create , Everybody praise more Leaving a message. Forwarding support ~ Always asking for praise , In fact, the message is more important , Because the message can more intuitively let me see your support , Interact with everyone 、 communication , I think it can also trigger a lot of my creative inspiration .

well , See you next time , I love cats and technology , More loving old watch ⁽⁽ଘ( ˙꒳˙ )ଓ⁾⁾

copyright notice
author[cousin],Please bring the original link to reprint, thank you.

Random recommended