scripts/cronjobs/pubsubber.py

#################################################
# Pubsubber.py: a minimalistic svnwcsub program #
#################################################

from threading import Thread
from datetime import datetime
import os
import sys
import logging
import atexit
import signal
import json
import re
import time
import subprocess
import base64

version = 2
if sys.hexversion < 0x03000000:
    print("Using Python 2...")
    import httplib, urllib, urllib2, ConfigParser as configparser, socket
    socket._fileobject.default_bufsize = 0
else:
    print("Using Python 3")
    version = 3
    import http.client, urllib.request, urllib.parse, configparser


debug = False
logger = None
watching={} # dict: key = path to watch, value = set() of paths to update on match

############################################################
# Daemon class, courtesy of an anonymous good-hearted soul #
############################################################
class daemon:
        """A generic daemon class.

        Usage: subclass the daemon class and override the run() method."""

        def __init__(self, pidfile, logfile = None):
            self.pidfile = pidfile
            if logfile == None:
                self.logfile = os.devnull
            else:
                self.logfile = logfile
        
        def daemonize(self):
                """Daemonize class. UNIX double fork mechanism."""

                try: 
                        pid = os.fork() 
                        if pid > 0:
                                # exit first parent
                                sys.exit(0) 
                except OSError as err: 
                        sys.stderr.write('fork #1 failed: {0}\n'.format(err))
                        sys.exit(1)
        
                # decouple from parent environment
                os.chdir('/') 
                os.setsid() 
                os.umask(0) 
        
                # do second fork
                try: 
                        pid = os.fork() 
                        if pid > 0:

                                # exit from second parent
                                sys.exit(0) 
                except OSError as err: 
                        sys.stderr.write('fork #2 failed: {0}\n'.format(err))
                        sys.exit(1) 
        
                # redirect standard file descriptors
                sys.stdout.flush()
                sys.stderr.flush()
                si = open(os.devnull, 'r')
                so = open(self.logfile, 'a+')
                se = open(self.logfile, 'a+')

                os.dup2(si.fileno(), sys.stdin.fileno())
                os.dup2(so.fileno(), sys.stdout.fileno())
                os.dup2(se.fileno(), sys.stderr.fileno())
        
                # write pidfile
                atexit.register(self.delpid)

                pid = str(os.getpid())
                with open(self.pidfile,'w+') as f:
                        f.write(pid + '\n')
                logger.info("Created %s", self.pidfile)
        
        def delpid(self):
                logger.info("Removing %s", self.pidfile)
                os.remove(self.pidfile)

        def start(self):
                """Start the daemon."""

                # Check for a pidfile to see if the daemon already runs
                try:
                        with open(self.pidfile,'r') as pf:

                                pid = int(pf.read().strip())
                except IOError:
                        pid = None
        
                if pid:
                        message = "pidfile {0} already exist. " + \
                                        "Daemon already running?\n"
                        sys.stderr.write(message.format(self.pidfile))
                        sys.exit(1)
                
                # Start the daemon
                self.daemonize()
                self.run()

        def stop(self):
                """Stop the daemon."""

                # Get the pid from the pidfile
                try:
                        with open(self.pidfile,'r') as pf:
                                pid = int(pf.read().strip())
                except IOError:
                        pid = None
        
                if not pid:
                        message = "pidfile {0} does not exist. " + \
                                        "Daemon not running?\n"
                        sys.stderr.write(message.format(self.pidfile))
                        return # not an error in a restart

                # Try killing the daemon process        
                try:
                        # Try gentle stop first
                        os.kill(pid, signal.SIGINT)
                        time.sleep(0.2)
                        while 1:
                                os.kill(pid, signal.SIGTERM)
                                time.sleep(0.1)
                except OSError as err:
                        e = str(err.args)
                        if e.find("No such process") > 0:
                                if os.path.exists(self.pidfile):
                                        os.remove(self.pidfile)
                        else:
                                print (str(err.args))
                                sys.exit(1)

        def restart(self):
                """Restart the daemon."""
                self.stop()
                self.start()

        def run(self):
                """You should override this method when you subclass Daemon.
                
                It will be called after the process has been daemonized by 
                start() or restart()."""



####################
# Helper functions #
####################


# read_chunk: iterator for reading chunks from the stream
# since this is all handled via urllib now, this is quite rudimentary
def read_chunk(req):
    while True:
        try:
            line = req.readline().strip()
            if line:
                yield line
            else:
                print("No more lines?")
                break
        except Exception as info:
            
            break
    return

 
#########################
# Main listener program #
#########################

# PubSub class: handles connecting to a pubsub service and checking commits
class PubSubClient(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.setDaemon(True)

    def run(self):
        logger.info("Watching %s", watching)
        while True:
            
            self.req = None
            while not self.req:
                try:
                    if version == 3:
                        self.req = urllib.request.urlopen(self.url, None, 30)
                    else:
                        self.req = urllib2.urlopen(self.url, None, 30)
                    
                    logger.info("Connected to %s", self.url)
                except:
                    logger.warn("Failed to connect to %s", self.url)
                    time.sleep(30)
                    continue
                
            for line in read_chunk(self.req):
                if version == 3:
                    line = str( line, encoding='ascii' ).rstrip('\r\n,').replace('\x00','') # strip away any old pre-0.9 commas from gitpubsub chunks and \0 in svnpubsub chunks
                else:
                    line = str( line ).rstrip('\r\n,').replace('\x00','') # strip away any old pre-0.9 commas from gitpubsub chunks and \0 in svnpubsub chunks
                try:
                    obj = json.loads(line)
                    if "commit" in obj and "repository" in obj['commit']:
                        # If it's our public svn repo, then...
                        if obj['commit']['repository'] == "13f79535-47bb-0310-9956-ffa450edef68":
                            #Grab some vars
                            commit = obj['commit']
                            # e.g. {"committer": "sebb", "log": "Ensure we exit on control+C", "repository": "13f79535-47bb-0310-9956-ffa450edef68", "format": 1, 
                            # "changed": {"comdev/reporter.apache.org/trunk/scandist.py": {"flags": "U  "}}, 
                            # "date": "2015-07-13 13:38:33 +0000 (Mon, 13 Jul 2015)", "type": "svn", "id": 1690668}
                            svnuser = commit['committer']
                            revision = commit['id']
                            filePaths = set()
                            # e.g. {"comdev/reporter.apache.org/trunk/scandist.py": {"flags": "U  "}}
                            for path in commit['changed']:
                                for watchPath in watching:
                                    # Check if the commit is for our part of the repo
                                    match = re.match("^%s" % watchPath, path)
                                    if match:
                                        filePath = str(watching[watchPath])
                                        if debug:
                                            print("Matched '" + path + "' against '" + watchPath + "'; would run 'svn up " + filePath + "'")
                                        filePaths.update(watching[watchPath])
                            if filePaths:
                                for filePath in filePaths:
                                    if debug:
                                        print("Matched 'r" + str(revision) + "'; would run 'svn up " + filePath + "'")
                                    else:
                                        time.sleep(3)
                                        logger.info("svn up %s", filePath)
                                        subprocess.call(['svn','up', filePath])
                            else:
                                logger.debug("Did not match 'r" + str(revision) + "' against ' " + str(watching.keys()) + "'")
                                if debug:
                                    print("Did not match 'r" + str(revision) + "' against ' " + str(watching.keys()) + "'")
                    

                except ValueError as detail:
                    continue
            




################         
# Main program #
################
"""
According to https://svn.apache.org/repos/asf/subversion/trunk/tools/server-side/svnpubsub/svnpubsub/server.py

#   URLs are constructed from 3 parts:
#       /${notification}/${optional_type}/${optional_repository}
#
#   Notifications can be sent for commits or metadata (e.g., revprop) changes.
#   If the type is included in the URL, you will only get notifications of that type.
#   The type can be * and then you will receive notifications of any type.
#
#   If the repository is included in the URL, you will only receive
#   messages about that repository.  The repository can be * and then you
#   will receive messages about all repositories.

"""

def main():
    if debug:
        print("Foreground test mode enabled, no updates will be made")
        for watchPath in watching:
            print("Watching: '" + watchPath + "' for updates to '" + str(watching[watchPath]) + "'")
        
  
    
    # Start the svn thread
    svn_thread = PubSubClient()
    svn_thread.url = "http://svn-master.apache.org:2069/commits/*"
    svn_thread.start()
    
    while True:
        try:
            time.sleep(60)
        except KeyboardInterrupt:
            logger.info("Detected shutdown interrupt")
            pass

##############
# Daemonizer #
##############
class MyDaemon(daemon):
    def run(self):
        main()

def usage():
    print("usage: %s start|stop|restart|foreground ([repo-path] [file-path])*" % sys.argv[0])
    print("for example: %s start comdev/projects.apache.org /var/www/projects.apache.org"  % sys.argv[0])

def handle_args(args):
    if len(args) %  2 != 0:
        usage()
        sys.exit("Need an even number of repo/file paths, found "+str(len(args)))
    else:
        for i in range(0, len(args), 2):
            try:
                watching[args[i]].update(args[i+1])
            except KeyError:
                watching[args[i]] = set([args[i+1]])

if __name__ == "__main__":
        logger = logging.getLogger('pubsubber')
        logger.setLevel(logging.INFO)
        formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s')
        ch = logging.StreamHandler()
        ch.setFormatter(formatter)
        logger.addHandler(ch)
        logfile = os.environ.get('LOGFILE')
        logger.info("LOGFILE=%s", logfile)
        daemon = MyDaemon('/tmp/pubsubber.pid', logfile)
        if len(sys.argv) >= 2:
                if 'start' == sys.argv[1]:
                    handle_args(sys.argv[2:])
                    daemon.start()
                elif 'stop' == sys.argv[1]:
                    daemon.stop()
                elif 'restart' == sys.argv[1]:
                    daemon.restart()
                elif 'foreground' == sys.argv[1]:
                    debug = True
                    handle_args(sys.argv[2:])
                    main()
                else:
                    usage()
                    sys.exit(2)
                sys.exit(0)
        else:
                usage()
                sys.exit(2)