Проверьте, работает ли скрипт python


У меня есть демон python, работающий как часть моего веб-приложения/ как я могу быстро проверить (используя python), если мой демон работает, а если нет, запустите его?

Я хочу сделать это таким образом, чтобы устранить любые сбои демона, и поэтому скрипт не нужно запускать вручную, он будет автоматически запускаться, как только он называется, а затем остаться работать.

Как я могу проверить (с помощью Python), если мой скрипт работает?

19 77

19 ответов:

поместите pidfile где-нибудь (например, /tmp). Затем вы можете проверить, выполняется ли процесс, проверяя, существует ли PID в файле. Не забудьте удалить файл, Когда вы закрыли чисто, и проверить его при запуске.

#/usr/bin/env python

import os
import sys

pid = str(os.getpid())
pidfile = "/tmp/mydaemon.pid"

if os.path.isfile(pidfile):
    print "%s already exists, exiting" % pidfile
    sys.exit()
file(pidfile, 'w').write(pid)
try:
    # Do some actual work here
finally:
    os.unlink(pidfile)

затем вы можете проверить, запущен ли процесс, проверив, есть ли содержимое /tmp/mydaemon.pid-это существующий процесс. Monit (упомянутый выше) может сделать это за вас, или вы можете написать простой скрипт оболочки для проверки это для вас с помощью кода возврата из ps.

ps up `cat /tmp/mydaemon.pid ` >/dev/null && echo "Running" || echo "Not running"

для дополнительного кредита вы можете использовать модуль atexit, чтобы убедиться, что ваша программа очищает свой pidfile при любых обстоятельствах (когда он убит, возникают исключения и т. д.).

метод, который удобен в системе Linux, использует доменные сокеты:

import socket
import sys
import time

def get_lock(process_name):
    # Without holding a reference to our socket somewhere it gets garbage
    # collected when the function exits
    get_lock._lock_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)

    try:
        get_lock._lock_socket.bind('' + process_name)
        print 'I got the lock'
    except socket.error:
        print 'lock exists'
        sys.exit()


get_lock('running_test')
while True:
    time.sleep(3)

это атомарный и позволяет избежать проблемы блокировки файлов, лежащих вокруг, если ваш процесс получает отправлен SIGKILL

Вы можете прочитайте в документации для socket.close что сокеты автоматически закрываются при сборе мусора.

на pid библиотека может сделать точно это.

from pid import PidFile

with PidFile():
  do_something()

Он также автоматически обрабатывает случай, когда pidfile существует, но процесс не выполняется.

есть очень хорошие пакеты для перезапуска процессов в UNIX. Тот, который имеет большой учебник о создании и настройке это monit. С некоторыми настройками вы можете иметь надежную проверенную технологию, поддерживающую ваш демон.

конечно пример от Дана не будет работать так, как должно быть.

действительно, если сценарий аварийно завершает работу, создает исключение или не очищает PID-файл, сценарий будет выполняться несколько раз.

Я предлагаю следующее на основе другого веб-сайта:

Это чтобы проверить, если уже существует файл блокировки

\#/usr/bin/env python
import os
import sys
if os.access(os.path.expanduser("~/.lockfile.vestibular.lock"), os.F_OK):
        #if the lockfile is already there then check the PID number
        #in the lock file
        pidfile = open(os.path.expanduser("~/.lockfile.vestibular.lock"), "r")
        pidfile.seek(0)
        old_pid = pidfile.readline()
        # Now we check the PID from lock file matches to the current
        # process PID
        if os.path.exists("/proc/%s" % old_pid):
                print "You already have an instance of the program running"
                print "It is running as process %s," % old_pid
                sys.exit(1)
        else:
                print "File is there but the program is not running"
                print "Removing lock file for the: %s as it can be there because of the program last time it was run" % old_pid
                os.remove(os.path.expanduser("~/.lockfile.vestibular.lock"))

это часть кода, где мы помещаем файл PID в файл блокировки

pidfile = open(os.path.expanduser("~/.lockfile.vestibular.lock"), "w")
pidfile.write("%s" % os.getpid())
pidfile.close()

этот код будет проверять значение pid по сравнению с существующим запущенным процессом., избегая двойного исполнения.

Я надеюсь, что это поможет.

есть множество вариантов. Одним из методов является использование системных вызовов или библиотек python, которые выполняют такие вызовы для вас. Другой просто породить процесс, как:

ps ax | grep processName

и парсить вывод. Многие люди выбирают такой подход, это не обязательно плохой подход, на мой взгляд.

попробуйте эту другую версию

def checkPidRunning(pid):        
    '''Check For the existence of a unix pid.
    '''
    try:
        os.kill(pid, 0)
    except OSError:
        return False
    else:
        return True

# Entry point
if __name__ == '__main__':
    pid = str(os.getpid())
    pidfile = os.path.join("/", "tmp", __program__+".pid")

    if os.path.isfile(pidfile) and checkPidRunning(int(file(pidfile,'r').readlines()[0])):
            print "%s already exists, exiting" % pidfile
            sys.exit()
    else:
        file(pidfile, 'w').write(pid)

    # Do some actual work here
    main()

    os.unlink(pidfile)

Я большой поклонник руководитель для управления демонами. Он написан на Python, поэтому есть много примеров того, как взаимодействовать с ним или расширять его из Python. Для ваших целей XML-RPC process control API должны работать хорошо.

наткнулся на этот старый вопрос, ища решение сам.

использовать psutil:

import psutil
import sys
from subprocess import Popen

for process in psutil.process_iter():
    if process.cmdline() == ['python', 'your_script.py']:
        sys.exit('Process found: exiting.')

print('Process not found: starting it.')
Popen(['python', 'your_script.py'])

мое решение-проверить аргументы процесса и командной строки Протестировано на windows и ubuntu linux

import psutil
import os

def is_running(script):
    for q in psutil.process_iter():
        if q.name().startswith('python'):
            if len(q.cmdline())>1 and script in q.cmdline()[1] and q.pid !=os.getpid():
                print("'{}' Process is already running".format(script))
                return True

    return False


if not is_running("test.py"):
    n = input("What is Your Name? ")
    print ("Hello " + n)

другие ответы отлично подходят для таких вещей, как задания cron, но если вы используете демон, вы должны контролировать его с помощью чего-то вроде daemontools.

ps ax | grep processName

Если yor debug script в pycharm всегда выходит

pydevd.py --multiproc --client 127.0.0.1 --port 33882 --file processName

вместо того, чтобы разрабатывать собственное решение PID-файла (которое имеет больше тонкостей и угловых случаев, чем вы могли бы подумать), посмотрите на supervisord -- это система управления процессом, которая позволяет легко обернуть управление заданиями и поведение демонов вокруг существующего скрипта Python.

попробуйте это:

#/usr/bin/env python
import os, sys, atexit

try:
    # Set PID file
    def set_pid_file():
        pid = str(os.getpid())
        f = open('myCode.pid', 'w')
        f.write(pid)
        f.close()

    def goodby():
        pid = str('myCode.pid')
        os.remove(pid)

    atexit.register(goodby)
    set_pid_file()
    # Place your code here

except KeyboardInterrupt:
    sys.exit(0)

вот более полезный код (с проверкой, если именно python выполняет скрипт):

#! /usr/bin/env python

import os
from sys import exit


def checkPidRunning(pid):
    global script_name
    if pid<1:
        print "Incorrect pid number!"
        exit()
    try:
        os.kill(pid, 0)
    except OSError:
        print "Abnormal termination of previous process."
        return False
    else:
        ps_command = "ps -o command= %s | grep -Eq 'python .*/%s'" % (pid,script_name)
        process_exist = os.system(ps_command)
        if process_exist == 0:
            return True
        else:
            print "Process with pid %s is not a Python process. Continue..." % pid
            return False


if __name__ == '__main__':
    script_name = os.path.basename(__file__)
    pid = str(os.getpid())
    pidfile = os.path.join("/", "tmp/", script_name+".pid")
    if os.path.isfile(pidfile):
        print "Warning! Pid file %s existing. Checking for process..." % pidfile
        r_pid = int(file(pidfile,'r').readlines()[0])
        if checkPidRunning(r_pid):
            print "Python process with pid = %s is already running. Exit!" % r_pid
            exit()
        else:
            file(pidfile, 'w').write(pid)
    else:
        file(pidfile, 'w').write(pid)

# main programm
....
....

os.unlink(pidfile)

вот строка:

ps_command = "ps -o command= %s | grep -Eq 'python .*/%s'" % (pid,script_name)

возвращает 0, если" grep "успешно, и процесс" python " в настоящее время выполняется с именем вашего скрипта в качестве параметра .

простой пример если вы только ищете имя процесса существуют или нет:

import os

def pname_exists(inp):
    os.system('ps -ef > /tmp/psef')
    lines=open('/tmp/psef', 'r').read().split('\n')
    res=[i for i in lines if inp in i]
    return True if res else False

Result:
In [21]: pname_exists('syslog')
Out[21]: True

In [22]: pname_exists('syslog_')
Out[22]: False

рассмотрим следующий пример, чтобы решить вашу проблему:

#!/usr/bin/python
# -*- coding: latin-1 -*-

import os, sys, time, signal

def termination_handler (signum,frame):
    global running
    global pidfile
    print 'You have requested to terminate the application...'
    sys.stdout.flush()
    running = 0
    os.unlink(pidfile)

running = 1
signal.signal(signal.SIGINT,termination_handler)

pid = str(os.getpid())
pidfile = '/tmp/'+os.path.basename(__file__).split('.')[0]+'.pid'

if os.path.isfile(pidfile):
    print "%s already exists, exiting" % pidfile
    sys.exit()
else:
    file(pidfile, 'w').write(pid)

# Do some actual work here

while running:
  time.sleep(10)

Я предлагаю этот скрипт, потому что он может быть выполнен только один раз.

использование bash для поиска процесса с именем текущего скрипта. Никаких лишних файлов.

import commands
import os
import time
import sys

def stop_if_already_running():
    script_name = os.path.basename(__file__)
    l = commands.getstatusoutput("ps aux | grep -e '%s' | grep -v grep | awk '{print }'| awk '{print }'" % script_name)
    if l[1]:
        sys.exit(0);

чтобы проверить, добавьте

stop_if_already_running()
print "running normally"
while True:
    time.sleep(3)

Это то, что я использую в Linux, чтобы избежать запуска сценария, если он уже запущен:

import os
import sys


script_name = os.path.basename(__file__)
pidfile = os.path.join("/tmp", os.path.splitext(script_name)[0]) + ".pid"


def create_pidfile():
    if os.path.exists(pidfile):
        with open(pidfile, "r") as _file:
            last_pid = int(_file.read())

        # Checking if process is still running
        last_process_cmdline = "/proc/%d/cmdline" % last_pid
        if os.path.exists(last_process_cmdline):
            with open(last_process_cmdline, "r") as _file:
                cmdline = _file.read()
            if script_name in cmdline:
                raise Exception("Script already running...")

    with open(pidfile, "w") as _file:
        pid = str(os.getpid())
        _file.write(pid)


def main():
    """Your application logic goes here"""


if __name__ == "__main__":
    create_pidfile()
    main()

этот подход работает хорошо без какой-либо зависимости от внешнего модуля.