Changeset - fb0c3af6031b
[Not reviewed]
celery
1 6 1
Marcin Kuzminski - 15 years ago 2010-09-23 01:08:33
marcin@python-works.com
Implemented locking for task, to prevent for running the same tasks,
moved out pidlock library. Added dirsize display
7 files changed with 71 insertions and 31 deletions:
0 comments (0 inline, 0 general)
pylons_app/lib/celerylib/__init__.py
Show inline comments
 
from pylons_app.lib.pidlock import DaemonLock, LockHeld
 
from vcs.utils.lazy import LazyProperty
 
from decorator import decorator
 
import logging
 
import os
 
import sys
 
import traceback
 

	
 
from hashlib import md5
 
log = logging.getLogger(__name__)
 

	
 
class ResultWrapper(object):
 
    def __init__(self, task):
 
        self.task = task
 
        
 
    @LazyProperty
 
    def result(self):
 
        return self.task
 

	
 
def run_task(task, *args, **kwargs):
 
    try:
 
        t = task.delay(*args, **kwargs)
 
        log.info('running task %s', t.task_id)
 
        return t
 
    except Exception, e:
 
        print e
 
        if e.errno == 111:
 
            log.debug('Unnable to connect. Sync execution')
 
        else:
 
            log.error(traceback.format_exc())
 
        #pure sync version
 
        return ResultWrapper(task(*args, **kwargs))
 

	
 

	
 
class LockTask(object):
 
    """LockTask decorator"""
 
    
 
    def __init__(self, func):
 
        self.func = func
 
        
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 
    
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        params = []
 
        params.extend(fargs)
 
        params.extend(fkwargs.values())
 
        lockkey = 'task_%s' % \
 
           md5(str(self.func) + '-' + '-'.join(map(str, params))).hexdigest()
 
        log.info('running task with lockkey %s', lockkey)
 
        try:
 
            l = DaemonLock(lockkey)
 
            return func(*fargs, **fkwargs)
 
            l.release()
 
        except LockHeld:
 
            log.info('LockHeld')
 
            return 'Task with key %s already running' % lockkey   
 

	
 
            
 
            
 

	
 
        
 
        
 
    
 
    
 
    
 
  
pylons_app/lib/celerylib/tasks.py
Show inline comments
 
from celery.decorators import task
 
from celery.task.sets import subtask
 
from celeryconfig import PYLONS_CONFIG as config
 
from pylons.i18n.translation import _
 
from pylons_app.lib.celerylib import run_task
 
from pylons_app.lib.celerylib import run_task, LockTask
 
from pylons_app.lib.helpers import person
 
from pylons_app.lib.smtp_mailer import SmtpMailer
 
from pylons_app.lib.utils import OrderedDict
 
from operator import itemgetter
 
from vcs.backends.hg import MercurialRepository
 
from time import mktime
 
import traceback
 
import json
 

	
 
__all__ = ['whoosh_index', 'get_commits_stats',
 
           'reset_user_password', 'send_email']
 

	
 
def get_session():
 
    from sqlalchemy import engine_from_config
 
    from sqlalchemy.orm import sessionmaker, scoped_session
 
    engine = engine_from_config(dict(config.items('app:main')), 'sqlalchemy.db1.')
 
    sa = scoped_session(sessionmaker(bind=engine))
 
    return sa
 

	
 
def get_hg_settings():
 
    from pylons_app.model.db import HgAppSettings
 
    try:
 
        sa = get_session()
 
        ret = sa.query(HgAppSettings).all()
 
@@ -47,73 +47,75 @@ def get_hg_ui_settings():
 
        sa.remove()
 
        
 
    if not ret:
 
        raise Exception('Could not get application ui settings !')
 
    settings = {}
 
    for each in ret:
 
        k = each.ui_key
 
        v = each.ui_value
 
        if k == '/':
 
            k = 'root_path'
 
        
 
        if k.find('.') != -1:
 
            k = k.replace('.', '_')
 
        
 
        if each.ui_section == 'hooks':
 
            v = each.ui_active
 
        
 
        settings[each.ui_section + '_' + k] = v  
 
    
 
    return settings   
 

	
 
@task
 
def whoosh_index(repo_location, full_index):
 
    log = whoosh_index.get_logger()
 
    from pylons_app.lib.indexers import DaemonLock
 
    from pylons_app.lib.pidlock import DaemonLock
 
    from pylons_app.lib.indexers.daemon import WhooshIndexingDaemon, LockHeld
 
    try:
 
        l = DaemonLock()
 
        WhooshIndexingDaemon(repo_location=repo_location)\
 
            .run(full_index=full_index)
 
        l.release()
 
        return 'Done'
 
    except LockHeld:
 
        log.info('LockHeld')
 
        return 'LockHeld'    
 

	
 

	
 
@task
 
@LockTask('get_commits_stats')
 
def get_commits_stats(repo_name, ts_min_y, ts_max_y):
 
    author_key_cleaner = lambda k: person(k).replace('"', "") #for js data compatibilty
 
        
 
    from pylons_app.model.db import Statistics, Repository
 
    log = get_commits_stats.get_logger()
 
    commits_by_day_author_aggregate = {}
 
    commits_by_day_aggregate = {}
 
    repos_path = get_hg_ui_settings()['paths_root_path'].replace('*', '')
 
    repo = MercurialRepository(repos_path + repo_name)
 

	
 
    skip_date_limit = True
 
    parse_limit = 500 #limit for single task changeset parsing
 
    parse_limit = 350 #limit for single task changeset parsing
 
    last_rev = 0
 
    last_cs = None
 
    timegetter = itemgetter('time')
 
    
 
    sa = get_session()
 
    
 
    dbrepo = sa.query(Repository)\
 
        .filter(Repository.repo_name == repo_name).scalar()
 
    cur_stats = sa.query(Statistics)\
 
        .filter(Statistics.repository == dbrepo).scalar()
 
    if cur_stats:
 
        last_rev = cur_stats.stat_on_revision
 
    
 
    if last_rev == repo.revisions[-1]:
 
        #pass silently without any work
 
        return True
 
    
 
    if cur_stats:
 
        commits_by_day_aggregate = OrderedDict(
 
                                       json.loads(
 
                                        cur_stats.commit_activity_combined))
 
        commits_by_day_author_aggregate = json.loads(cur_stats.commit_activity)
 
    
 
    for cnt, rev in enumerate(repo.revisions[last_rev:]):
 
@@ -184,49 +186,51 @@ def get_commits_stats(repo_name, ts_min_
 
        overview_data.append([k, v])
 
    overview_data = sorted(overview_data, key=itemgetter(0))
 
        
 
    if not commits_by_day_author_aggregate:
 
        commits_by_day_author_aggregate[author_key_cleaner(repo.contact)] = {
 
            "label":author_key_cleaner(repo.contact),
 
            "data":[0, 1],
 
            "schema":["commits"],
 
        }
 

	
 
    stats = cur_stats if cur_stats else Statistics()
 
    stats.commit_activity = json.dumps(commits_by_day_author_aggregate)
 
    stats.commit_activity_combined = json.dumps(overview_data)
 
    stats.repository = dbrepo
 
    stats.stat_on_revision = last_cs.revision
 
    stats.languages = json.dumps({'_TOTAL_':0, '':0})
 
    
 
    try:
 
        sa.add(stats)
 
        sa.commit()    
 
    except:
 
        log.error(traceback.format_exc())
 
        sa.rollback()
 
        return False
 
                        
 
    
 
    run_task(get_commits_stats, repo_name, ts_min_y, ts_max_y)
 
                            
 
    return True
 

	
 
@task
 
def reset_user_password(user_email):
 
    log = reset_user_password.get_logger()
 
    from pylons_app.lib import auth
 
    from pylons_app.model.db import User
 
    
 
    try:
 
        try:
 
            sa = get_session()
 
            user = sa.query(User).filter(User.email == user_email).scalar()
 
            new_passwd = auth.PasswordGenerator().gen_password(8,
 
                             auth.PasswordGenerator.ALPHABETS_BIG_SMALL)
 
            if user:
 
                user.password = auth.get_crypt_password(new_passwd)
 
                sa.add(user)
 
                sa.commit()
 
                log.info('change password for %s', user_email)
 
            if new_passwd is None:
 
                raise Exception('unable to generate new password')
 
            
 
        except:
 
            log.error(traceback.format_exc())
pylons_app/lib/indexers/__init__.py
Show inline comments
 
from os.path import dirname as dn, join as jn
 
from pidlock import LockHeld, DaemonLock
 
from pylons_app.config.environment import load_environment
 
from pylons_app.model.hg_model import HgModel
 
from shutil import rmtree
 
from webhelpers.html.builder import escape
 
from vcs.utils.lazy import LazyProperty
 

	
 
from whoosh.analysis import RegexTokenizer, LowercaseFilter, StopFilter
 
from whoosh.fields import TEXT, ID, STORED, Schema, FieldType
 
from whoosh.index import create_in, open_dir
 
from whoosh.formats import Characters
 
from whoosh.highlight import highlight, SimpleFragmenter, HtmlFormatter   
 

	
 
import os
 
import sys
 
import traceback
 

	
 
#to get the pylons_app import
 
sys.path.append(dn(dn(dn(os.path.realpath(__file__)))))
 

	
 

	
 
#LOCATION WE KEEP THE INDEX
 
IDX_LOCATION = jn(dn(dn(dn(dn(os.path.abspath(__file__))))), 'data', 'index')
 

	
 
#EXTENSIONS WE WANT TO INDEX CONTENT OFF
pylons_app/lib/indexers/daemon.py
Show inline comments
 
@@ -11,49 +11,49 @@
 
# This program is distributed in the hope that it will be useful,
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
# GNU General Public License for more details.
 
# 
 
# You should have received a copy of the GNU General Public License
 
# along with this program; if not, write to the Free Software
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
 
# MA  02110-1301, USA.
 
"""
 
Created on Jan 26, 2010
 

	
 
@author: marcink
 
A deamon will read from task table and run tasks
 
"""
 
import sys
 
import os
 
from os.path import dirname as dn
 
from os.path import join as jn
 

	
 
#to get the pylons_app import
 
project_path = dn(dn(dn(dn(os.path.realpath(__file__)))))
 
sys.path.append(project_path)
 

	
 
from pidlock import LockHeld, DaemonLock
 
from pylons_app.lib.pidlock import LockHeld, DaemonLock
 
from pylons_app.model.hg_model import HgModel
 
from pylons_app.lib.helpers import safe_unicode
 
from whoosh.index import create_in, open_dir
 
from shutil import rmtree
 
from pylons_app.lib.indexers import INDEX_EXTENSIONS, IDX_LOCATION, SCHEMA, IDX_NAME
 

	
 
import logging
 

	
 
log = logging.getLogger('whooshIndexer')
 
# create logger
 
log.setLevel(logging.DEBUG)
 
log.propagate = False
 
# create console handler and set level to debug
 
ch = logging.StreamHandler()
 
ch.setLevel(logging.DEBUG)
 

	
 
# create formatter
 
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 

	
 
# add formatter to ch
 
ch.setFormatter(formatter)
 

	
 
# add ch to logger
 
log.addHandler(ch)
pylons_app/lib/pidlock.py
Show inline comments
 
file renamed from pylons_app/lib/indexers/pidlock.py to pylons_app/lib/pidlock.py
 
import os, time
 
import sys
 
from warnings import warn
 

	
 
class LockHeld(Exception):pass
 

	
 

	
 
class DaemonLock(object):
 
    '''daemon locking
 
    """daemon locking
 
    USAGE:
 
    try:
 
        l = lock()
 
        main()
 
        l.release()
 
    except LockHeld:
 
        sys.exit(1)
 
    '''
 
    """
 

	
 
    def __init__(self, file=None, callbackfn=None,
 
                 desc='daemon lock', debug=False):
 

	
 
        self.pidfile = file if file else os.path.join(os.path.dirname(__file__),
 
                                                      'running.lock')
 
        self.callbackfn = callbackfn
 
        self.desc = desc
 
        self.debug = debug
 
        self.held = False
 
        #run the lock automatically !
 
        self.lock()
 

	
 
    def __del__(self):
 
        if self.held:
 

	
 
#            warn("use lock.release instead of del lock",
 
#                    category = DeprecationWarning,
 
#                    stacklevel = 2)
 

	
 
            # ensure the lock will be removed
 
            self.release()
 

	
 

	
 
    def lock(self):
 
        '''
 
        """
 
        locking function, if lock is present it will raise LockHeld exception
 
        '''
 
        """
 
        lockname = '%s' % (os.getpid())
 

	
 
        self.trylock()
 
        self.makelock(lockname, self.pidfile)
 
        return True
 

	
 
    def trylock(self):
 
        running_pid = False
 
        try:
 
            pidfile = open(self.pidfile, "r")
 
            pidfile.seek(0)
 
            running_pid = pidfile.readline()
 
            if self.debug:
 
                print 'lock file present running_pid: %s, checking for execution'\
 
                % running_pid
 
            # Now we check the PID from lock file matches to the current
 
            # process PID
 
            if running_pid:
 
                if os.path.exists("/proc/%s" % running_pid):
 
                        print "You already have an instance of the program running"
 
                        print "It is running as process %s" % running_pid
 
                        raise LockHeld
 
                else:
 
                        print "Lock File is there but the program is not running"
 
                        print "Removing lock file for the: %s" % running_pid
 
                        self.release()
 
        except IOError, e:
 
            if e.errno != 2:
 
                raise
 

	
 

	
 
    def release(self):
 
        '''
 
        """
 
        releases the pid by removing the pidfile
 
        '''
 
        """
 
        if self.callbackfn:
 
            #execute callback function on release
 
            if self.debug:
 
                print 'executing callback function %s' % self.callbackfn
 
            self.callbackfn()
 
        try:
 
            if self.debug:
 
                print 'removing pidfile %s' % self.pidfile
 
            os.remove(self.pidfile)
 
            self.held = False
 
        except OSError, e:
 
            if self.debug:
 
                print 'removing pidfile failed %s' % e
 
            pass
 

	
 
    def makelock(self, lockname, pidfile):
 
        '''
 
        """
 
        this function will make an actual lock
 
        @param lockname: acctual pid of file
 
        @param pidfile: the file to write the pid in
 
        '''
 
        """
 
        if self.debug:
 
            print 'creating a file %s and pid: %s' % (pidfile, lockname)
 
        pidfile = open(self.pidfile, "wb")
 
        pidfile.write(lockname)
 
        pidfile.close
 
        self.held = True
 

	
 

	
 
def main():
 
    print 'func is running'
 
    cnt = 20
 
    while 1:
 
        print cnt
 
        if cnt == 0:
 
            break
 
        time.sleep(1)
 
        cnt -= 1
 

	
 

	
 
if __name__ == "__main__":
 
    try:
 
        l = DaemonLock(desc='test lock')
 
        main()
 
        l.release()
pylons_app/lib/utils.py
Show inline comments
 
@@ -353,51 +353,50 @@ class OrderedDict(dict, DictMixin):
 
    @classmethod
 
    def fromkeys(cls, iterable, value=None):
 
        d = cls()
 
        for key in iterable:
 
            d[key] = value
 
        return d
 

	
 
    def __eq__(self, other):
 
        if isinstance(other, OrderedDict):
 
            return len(self) == len(other) and self.items() == other.items()
 
        return dict.__eq__(self, other)
 

	
 
    def __ne__(self, other):
 
        return not self == other
 

	
 

	
 
#===============================================================================
 
# TEST FUNCTIONS
 
#===============================================================================
 
def create_test_index(repo_location, full_index):
 
    """Makes default test index
 
    @param repo_location:
 
    @param full_index:
 
    """
 
    from pylons_app.lib.indexers import daemon
 
    from pylons_app.lib.indexers.daemon import WhooshIndexingDaemon
 
    from pylons_app.lib.indexers.pidlock import DaemonLock, LockHeld
 
    from pylons_app.lib.pidlock import DaemonLock, LockHeld
 
    from pylons_app.lib.indexers import IDX_LOCATION
 
    import shutil
 
    
 
    if os.path.exists(IDX_LOCATION):
 
        shutil.rmtree(IDX_LOCATION)
 
         
 
    try:
 
        l = DaemonLock()
 
        WhooshIndexingDaemon(repo_location=repo_location)\
 
            .run(full_index=full_index)
 
        l.release()
 
    except LockHeld:
 
        pass    
 
    
 
def create_test_env(repos_test_path, config):
 
    """Makes a fresh database and 
 
    install test repository into tmp dir
 
    """
 
    from pylons_app.lib.db_manage import DbManage
 
    import tarfile
 
    import shutil
 
    from os.path import dirname as dn, join as jn, abspath
 
    
 
    log = logging.getLogger('TestEnvCreator')
pylons_app/templates/files/files_browser.html
Show inline comments
 
@@ -8,68 +8,69 @@
 
<div id="body" class="browserblock">
 
	<div class="browser-header">
 
		${h.form(h.url.current())}
 
		<div class="info_box">
 
          <span >${_('view')}@rev</span> 
 
          <a href="${c.url_prev}">&laquo;</a>
 
          ${h.text('at_rev',value=c.rev_nr,size=3)}
 
          <a href="${c.url_next}">&raquo;</a>
 
          ${h.submit('view','view')}
 
	    </div>           
 
		${h.end_form()}
 
	</div>
 
	<div class="browser-body">
 
		<table class="code-browser">
 
		         <thead>
 
		             <tr>
 
		                 <th>${_('Name')}</th>
 
		                 <th>${_('Size')}</th>
 
		                 <th>${_('Mimetype')}</th>
 
		                 <th>${_('Revision')}</th>
 
		                 <th>${_('Last modified')}</th>
 
		                 <th>${_('Last commiter')}</th>
 
		             </tr>
 
		         </thead>
 
		         	<tr class="parity0">
 
		          		<td>
 
		          		% if c.files_list.parent:
 
		          			${h.link_to('..',h.url('files_home',repo_name=c.repo_name,revision=c.cur_rev,f_path=c.files_list.parent.path),class_="browser-dir")}
 
		          		%endif
 
		          		</td>
 
		          		<td></td>
 
		          		<td></td>
 
		          		<td></td>
 
		          		<td></td>
 
		         	</tr>
 

	
 
          		% if c.files_list.parent:
 
         		<tr class="parity0">
 
	          		<td>		          		
 
	          			${h.link_to('..',h.url('files_home',repo_name=c.repo_name,revision=c.cur_rev,f_path=c.files_list.parent.path),class_="browser-dir")}
 
	          		</td>
 
	          		<td></td>
 
	          		<td></td>
 
	          		<td></td>
 
	          		<td></td>
 
	          		<td></td>
 
				</tr>	          		
 
          		%endif
 
		         	
 
		    %for cnt,node in enumerate(c.files_list,1):
 
				<tr class="parity${cnt%2}">
 
		             <td>
 
						${h.link_to(node.name,h.url('files_home',repo_name=c.repo_name,revision=c.cur_rev,f_path=node.path),class_=file_class(node))}
 
		             </td>
 
		             <td>
 
		                %if node.is_file():
 
		             		${h.format_byte_size(node.size,binary=True)}
 
		             	%endif
 
		             	${h.format_byte_size(node.size,binary=True)}
 
		             </td>
 
		             <td>
 
		              %if node.is_file():
 
		                  ${node.mimetype}
 
		              %endif
 
		             </td>
 
		             <td>
 
		             	%if node.is_file():
 
		             		${node.last_changeset.revision}
 
		             	%endif
 
		             </td>
 
		             <td>
 
		             	%if node.is_file():
 
		             		${h.age(node.last_changeset._ctx.date())} - ${node.last_changeset.date}
 
		             	%endif
 
		             </td>
 
		             <td>
 
		             	%if node.is_file():
 
		             		${node.last_changeset.author}
 
		             	%endif                    
 
		             </td>
 
				</tr>
 
			%endfor
 
		</table>
0 comments (0 inline, 0 general)