Cleaner.py

from __builtin__ import open, type

__author__ = 'Manuela Kuhn <marnuel.kuhn@desy.de>', 'Marco Strutz <marco.strutz@desy.de>'


import time
import argparse
import zmq
import os
import logging
import sys
import json
import traceback
from multiprocessing import Process, freeze_support
import subprocess
import json
import shutil
import helperScript
from RingBuffer import RingBuffer


DEFAULT_CHUNK_SIZE = 1048576


#
#  --------------------------  class: Cleaner  --------------------------------------
#
class Cleaner():
    """
    * received cleaning jobs via zeromq,
      such as removing a file
    * Does regular checks on the watched directory,
    such as
      - deleting files which have been successfully send
        to target but still remain in the watched directory
      - poll the watched directory and reissue new files
        to fileMover which have not been detected yet
    """
    bindingPortForSocket = None
    bindingIpForSocket   = None
    zmqContextForCleaner = None
    externalContext      = None    # if the context was created outside this class or not
    zmqCleanerSocket     = None

    # to get the logging only handling this class
    log                  = None

    useRealTimeAnalysis  = True    # boolian to inform if the receiver for the realtime analysis is running
    maxRingBufferSize    = None
    ringBuffer           = None

    def __init__(self, targetPath, bindingIp="127.0.0.1", bindingPort="6062", maxRingBufferSize = None, context = None, verbose=False):
        self.bindingPortForSocket = bindingPort
        self.bindingIpForSocket   = bindingIp
        self.targetPath           = targetPath

        if context:
            self.zmqContextForCleaner = context
            self.externalContext      = True
        else:
            self.zmqContextForCleaner = zmq.Context()
            self.externalContext      = False

        if maxRingBufferSize:
            self.maxRingBufferSize    = maxRingBufferSize
#            # TODO remove targetPath?
#            self.ringBuffer           = RingBuffer(self.maxRingBufferSize, self.targetPath)
            self.ringBuffer           = RingBuffer(self.maxRingBufferSize)


        self.log = self.getLogger()
        self.log.debug("Init")

        #bind to local port
        self.zmqCleanerSocket      = self.zmqContextForCleaner.socket(zmq.PULL)
        connectionStrCleanerSocket = "tcp://" + self.bindingIpForSocket + ":%s" % self.bindingPortForSocket
        self.zmqCleanerSocket.bind(connectionStrCleanerSocket)
        self.log.debug("zmqCleanerSocket started for '" + connectionStrCleanerSocket + "'")

        try:
            self.process()
        except zmq.error.ZMQError:
            self.log.error("ZMQError: "+ str(e))
            self.log.debug("Shutting down cleaner.")
        except KeyboardInterrupt:
            self.log.info("KeyboardInterrupt detected. Shutting down cleaner.")
        except:
            trace = traceback.format_exc()
            self.log.error("Stopping cleanerProcess due to unknown error condition.")
            self.log.debug("Error was: " + str(trace))

        self.stop()


    def getLogger(self):
        logger = logging.getLogger("cleaner")
        return logger


    def process(self):
        #processing messaging
        while True:
            #waiting for new jobs
            self.log.debug("Waiting for new jobs")
            try:
                workload = self.zmqCleanerSocket.recv()
            except Exception as e:
                self.log.error("Error in receiving job: " + str(e))

            if workload == "STOP":
                self.log.info("Stopping cleaner")
                self.stop()
                break
            elif workload == "START_REALTIME_ANALYSIS":
                self.useRealTimeAnalysis = True
                self.log.info("Starting realtime analysis")
                break
            elif workload == "STOP_REALTIME_ANALYSIS":
                self.useRealTimeAnalysis = False
                self.log.info("Stopping realtime analysis")
                break

            # transform to dictionary
            # metadataDict = {
            #   "filename"             : filename,
            #   "filesize"             : filesize,
            #   "fileModificationTime" : fileModificationTime,
            #   "sourcePath"           : sourcePath,
            #   "relativePath"         : relativePath,
            #   "chunkSize"            : self.getChunkSize()
            #   }
            try:
                workloadDict = json.loads(str(workload))
            except:
                errorMessage = "invalid job received. skipping job"
                self.log.error(errorMessage)
                self.log.debug("workload=" + str(workload))
                continue

            #extract fileEvent metadata
            try:
                #TODO validate fileEventMessageDict dict
                filename       = workloadDict["filename"]
                sourcePath     = workloadDict["sourcePath"]
                relativePath   = workloadDict["relativePath"]
                if self.useRealTimeAnalysis:
                    modTime        = workloadDict["fileModificationTime"]
                # filesize       = workloadDict["filesize"]
            except Exception, e:
                errorMessage   = "Invalid fileEvent message received."
                self.log.error(errorMessage)
                self.log.debug("Error was: " + str(e))
                self.log.debug("workloadDict=" + str(workloadDict))
                #skip all further instructions and continue with next iteration
                continue

            #source file
            sourceFullpath = None
            try:
                #generate target filepath
                sourcePath = os.path.normpath(sourcePath + os.sep + relativePath)
                sourceFullPath = os.path.join(sourcePath,filename)
                targetFullPath = os.path.normpath(self.targetPath + relativePath)
                self.log.debug("sourcePath: " + str (sourcePath))
                self.log.debug("filename: " + str (filename))
                self.log.debug("targetPath: " + str (targetFullPath))

            except Exception, e:
                self.log.error("Unable to generate file paths")
                trace = traceback.format_exc()
                self.log.error("Error was: " + str(trace))
                #skip all further instructions and continue with next iteration
                continue

            if self.useRealTimeAnalysis:
                # copy file
                try:
                    self.log.debug("Copying source file...")
                    self.copyFile(sourcePath, filename, targetFullPath)
                    self.log.debug("File copied: " + str(sourceFullPath))
                    self.log.debug("Copying source file...success.")

                except Exception, e:
                    self.log.error("Unable to copy source file: " + str (sourceFullPath) )
                    trace = traceback.format_exc()
                    self.log.error("Error was: " + str(trace))
                    self.log.debug("sourceFullpath="+str(sourceFullpath))
                    self.log.debug("Copying source file...failed.")
                    #skip all further instructions and continue with next iteration
                    continue

                # add file to ring buffer
                self.log.debug("Add new file to ring buffer: " + str(sourceFullPath) + ", " + str(modTime))
                self.ringBuffer.add(sourceFullPath, modTime)
            else:
                try:
                    self.log.debug("Moving source file...")
                    self.moveFile(sourcePath, filename, targetFullPath)
#                    self.removeFile(sourceFullpath)

                    # #show filesystem statistics
                    # try:
                    #     self.showFilesystemStatistics(sourcePath)
                    # except Exception, f:
                    #     logging.warning("Unable to get filesystem statistics")
                    #     logging.debug("Error was: " + str(f))
                    self.log.debug("File moved: " + str(sourceFullPath))
                    self.log.debug("Moving source file...success.")

                except Exception, e:
                    self.log.error("Unable to move source file: " + str (sourceFullPath) )
                    trace = traceback.format_exc()
                    self.log.error("Error was: " + str(trace))
                    self.log.debug("sourceFullpath="+str(sourceFullpath))
                    self.log.debug("Moving source file...failed.")
                    #skip all further instructions and continue with next iteration
                    continue


    def copyFile(self, source, filename, target):
        maxAttemptsToCopyFile     = 2
        waitTimeBetweenAttemptsInMs = 500


        iterationCount = 0
        self.log.info("Copying file '" + str(filename) + "' from '" +  str(source) + "' to '" + str(target) + "' (attempt " + str(iterationCount) + ")...success.")
        fileWasCopied = False

        while iterationCount <= maxAttemptsToCopyFile and not fileWasCopied:
            iterationCount+=1
            try:
                # check if the directory exists before moving the file
                if not os.path.exists(target):
                    try:
                        os.makedirs(target)
                    except OSError:
                        pass
                # moving the file
                sourceFile = source + os.sep + filename
                targetFile = target + os.sep + filename
                self.log.debug("sourceFile: " + str(sourceFile))
                self.log.debug("targetFile: " + str(targetFile))
                shutil.copyfile(sourceFile, targetFile)
                fileWasCopied = True
                self.log.debug("Copying file '" + str(filename) + "' from '" + str(source) + "' to '" + str(target) + "' (attempt " + str(iterationCount) + ")...success.")
            except IOError:
                self.log.debug ("IOError: " + str(filename))
            except Exception, e:
                trace = traceback.format_exc()
                warningMessage = "Unable to copy file {FILE}.".format(FILE=str(source) + str(filename))
                self.log.warning(warningMessage)
                self.log.debug("trace=" + str(trace))
                self.log.warning("will try again in {MS}ms.".format(MS=str(waitTimeBetweenAttemptsInMs)))

        if not fileWasCopied:
            self.log.error("Copying file '" + str(filename) + " from " + str(source) + " to " + str(target) + "' (attempt " + str(iterationCount) + ")...FAILED.")
            raise Exception("maxAttemptsToCopyFile reached (value={ATTEMPT}). Unable to move file '{FILE}'.".format(ATTEMPT=str(iterationCount), FILE=filename))


    def moveFile(self, source, filename, target):
        maxAttemptsToMoveFile     = 2
        waitTimeBetweenAttemptsInMs = 500


        iterationCount = 0
        self.log.info("Moving file '" + str(filename) + "' from '" +  str(source) + "' to '" + str(target) + "' (attempt " + str(iterationCount) + ")...success.")
        fileWasMoved = False

        while iterationCount <= maxAttemptsToMoveFile and not fileWasMoved:
            iterationCount+=1
            try:
                # check if the directory exists before moving the file
                if not os.path.exists(target):
                    try:
                        os.makedirs(target)
                    except OSError:
                        pass
                # moving the file
                sourceFile = source + os.sep + filename
                targetFile = target + os.sep + filename
                self.log.debug("sourceFile: " + str(sourceFile))
                self.log.debug("targetFile: " + str(targetFile))
                shutil.move(sourceFile, targetFile)
                fileWasMoved = True
                self.log.debug("Moving file '" + str(filename) + "' from '" + str(source) + "' to '" + str(target) + "' (attempt " + str(iterationCount) + ")...success.")
            except IOError:
                self.log.debug ("IOError: " + str(filename))
            except Exception, e:
                trace = traceback.format_exc()
                warningMessage = "Unable to move file {FILE}.".format(FILE=str(source) + str(filename))
                self.log.warning(warningMessage)
                self.log.debug("trace=" + str(trace))
                self.log.warning("will try again in {MS}ms.".format(MS=str(waitTimeBetweenAttemptsInMs)))

        if not fileWasMoved:
            self.log.error("Moving file '" + str(filename) + " from " + str(source) + " to " + str(target) + "' (attempt " + str(iterationCount) + ")...FAILED.")
            raise Exception("maxAttemptsToMoveFile reached (value={ATTEMPT}). Unable to move file '{FILE}'.".format(ATTEMPT=str(iterationCount), FILE=filename))


    def removeFile(self, filepath):
        maxAttemptsToRemoveFile     = 2
        waitTimeBetweenAttemptsInMs = 500


        iterationCount = 0
        self.log.info("Removing file '" + str(filepath) + "' (attempt " + str(iterationCount) + ")...")
        fileWasRemoved = False

        while iterationCount <= maxAttemptsToRemoveFile and not fileWasRemoved:
            iterationCount+=1
            try:
                os.remove(filepath)
                fileWasRemoved = True
                self.log.debug("Removing file '" + str(filepath) + "' (attempt " + str(iterationCount) + ")...success.")
            except Exception, e:
                trace = traceback.format_exc()
                warningMessage = "Unable to remove file {FILE}.".format(FILE=str(filepath))
                self.log.warning(warningMessage)
                self.log.debug("trace=" + str(trace))
                self.log.warning("will try again in {MS}ms.".format(MS=str(waitTimeBetweenAttemptsInMs)))

        if not fileWasRemoved:
            self.log.error("Removing file '" + str(filepath) + "' (attempt " + str(iterationCount) + ")...FAILED.")
            raise Exception("maxAttemptsToRemoveFile reached (value={ATTEMPT}). Unable to remove file '{FILE}'.".format(ATTEMPT=str(iterationCount), FILE=filepath))


    def stop(self):
        self.log.debug("Closing socket")
        self.zmqCleanerSocket.close(0)
        if not self.externalContext:
            self.log.debug("Destroying context")
            self.zmqContextForCleaner.destroy()