Updated to work on unscaled openshift apps

.gitignore

@@ -0,0 +1,2 @@
+env
+logs

README.md

@@ -1,17 +1,35 @@
 openshift-celery-cartridge
 ==========================
 
-Cartridge to Expose Celery as a Daemon on OpenShift
+Cartridge to Expose Celery as a Daemon on OpenShift.
 
-Environment Variables
+Status
 ---------------------
+This work on unscaled applications but only as a plugin cartridge on scaled applications. This means that it's can't scale independently of your main gear. At this point having a celery cartridge is no better than just starting celery in your app.y or action hooks. Further development can combine this with a python cartridge so it can operate in it's own gears.
+
+Instead of using this I recommend starting celery in your action hooks or as part of you app.py instead. This will give more flexibility, and example is here https://github.com/appsembler/appsembler-launch-openshift 
+
+Configuration
+---------------------
+
+*  <code>OPENSHIFT_CELERY_CONFIG</code> 
+    This is the name of your config file, which by default is `$OPENSHIFT_CELERY_DIR/conf.d/celeryconfig.py` but you can copy this to `$OPENSHIFT_DATA_DIR/config/celeryconfig.py` and edit your own copy.
+
+To install
+---------------------
+
+    rhc cartridge-add https://raw.github.com/wassname/openshift-celery-cartridge/master/metadata/manifest.yml -a "appname"
+
+Any log output will be generated to `${OPENSHIFT_HOMEDIR}logs/celery_log.txt` and will be viewable with the rhc tail "appname" command
+
+This was tested using:
+
+    celery==3.1.11
+    redis==2.10.3
+
+To manage
+---------------------
+
+    $ rhc cartridge-status celeryd -a "yourapp"
 
-- OPENSHIFT_CELERY_BROKER_TRANS : Defines the broker type that celery will use.  Current available options are:
-    -- mongodb : DEFAULT
-    -- amqp
-    -- redis
-- OPENSHIFT_CELERY_BROKER_URL : The connection URL for the broker, omitting the transport.  For example, an amqp value may look like: guest:guest@localhost:5672//
-- 
-- OPENSHIFT_CELERY_IMPORTS : Defines the modules that celery should import.  Currently only supports 1 module import path
 
-Currently only supports MongoDB as backend for queues, but in progress to support more environment variables to allow vairous backends and configurations.

bin/control

@@ -1,32 +1,68 @@
 #!/bin/bash -e
 
-PATH=/bin/:/usr/bin:$PATH
-
-#source $OPENSHIFT_CARTRIDGE_SDK_BASH
+source $OPENSHIFT_CARTRIDGE_SDK_BASH
 
 function start {
-    export PYTHONPATH=$OPENSHIFT_REPO_DIR/.openshift:$OPENSHIFT_REPO_DIR/wsgi::$OPENSHIFT_CELERY_DIR/conf.d:$OPENSHIFT_REPO_DIR:$PYTHONPATH
-    usr/celeryd-multi start --cmd=celeryd --pidfile=etc/celeryd.pid --hostname=$OPENSHIFT_INTERNAL_IP --loglevel=DEBUG
+	LOGPIPE=${OPENSHIFT_HOMEDIR}/app-root/runtime/logshifter-python
+	rm -f $LOGPIPE && mkfifo $LOGPIPE
+	/usr/bin/logshifter -tag python < $LOGPIPE &
+
+    PYTHONPATH=$OPENSHIFT_DATA_DIR:$OPENSHIFT_DATA_DIR/config/:$OPENSHIFT_REPO_DIR/.openshift:$OPENSHIFT_REPO_DIR/config:$OPENSHIFT_REPO_DIR/EXT:$OPENSHIFT_REPO_DIR/wsgi::$OPENSHIFT_CELERY_DIR/conf.d:$OPENSHIFT_REPO_DIR:$PYTHONPATH
+
+    echo  "Starting Celery"
+    #echo "nohup nice -n 10 ${OPENSHIFT_CELERY_DIR}usr/celery multi start worker --config=${OPENSHIFT_CELERY_CONFIG=celeryconfig}  --loglevel=DEBUG --hostname=$OPENSHIFT_APP_DNS --pidfile=${OPENSHIFT_CELERY_DIR}etc/celeryd.pid --logfile=${OPENSHIFT_HOMEDIR}app-root/logs/celery_log.txt &> $LOGPIPE &" > ${OPENSHIFT_LOG_DIR}celery_log.txt
+    nohup nice -n 10 ${OPENSHIFT_CELERY_DIR}usr/celery multi start worker --config=${OPENSHIFT_CELERY_CONFIG=celeryconfig}  --loglevel=DEBUG --hostname=$OPENSHIFT_APP_DNS --pidfile=${OPENSHIFT_CELERY_DIR}etc/celeryd.pid --logfile=${OPENSHIFT_LOG_DIR}celery_log.txt &> $LOGPIPE &
+    if ps -p `cat ${OPENSHIFT_CELERY_DIR}etc/celeryd.pid` > /dev/null; 
+    then
+       echo "Starting celery worked"
+    else
+        echo "Starting celery failed"
+    fi
 }
 
 function stop {
-    if ps -p `cat etc/celeryd.pid` > /dev/null 2>$1
+    celery multi stop worker --pidfile=${OPENSHIFT_CELERY_DIR}etc/celeryd.pid
+    if ps -p `cat ${OPENSHIFT_CELERY_DIR}etc/celeryd.pid` > /dev/null; 
     then
-        kill -9 `cat etc/celeryd.pid`
+        kill -9 `cat ${OPENSHIFT_CELERY_DIR}etc/celeryd.pid`
+        rm -f ${OPENSHIFT_CELERY_DIR}etc/celeryd.pid
     else
         echo "nothing to kill"
     fi
     echo "stopped"
 }
 
 function restart {
-    echo "restarted"
+    celery multi restart worker --config=${OPENSHIFT_CELERY_CONFIG=celeryconfig}  --loglevel=DEBUG --hostname=$OPENSHIFT_APP_DNS --pidfile=${OPENSHIFT_CELERY_DIR}etc/celeryd.pid --logfile=${OPENSHIFT_LOG_DIR}celery_log.txt 
+    echo "Restarted celery"
+
 }
 
+function status() {
+    pid=`cat ${OPENSHIFT_CELERY_DIR}etc/celeryd.pid`
+    PROCESS_COUNT=$(ps -ef | grep "${PROCESS_NAME}" | grep -v "grep" | wc -l)
+    client_result $pwd
+    if [ -f $pid ] && ( kill -0 $(cat $pid) ); then
+        client_result "Celery is running at pid:${pid}, $PROCESS_COUNT instances of celery"
+    else
+        client_result "Celery is not running"
+    fi
+}
+
+
 function catchall {
     echo "not yet implemented"
 }
 
+# Ensure arguments.
+if ! [ $# -gt 0 ]; then
+    echo "Usage: $0 [start|restart|stop|status]"
+    exit 1
+fi
+
+# Source utility functions.
+source $OPENSHIFT_CARTRIDGE_SDK_BASH
+
 case "$1" in
   start)       start ;;
   stop)        stop ;;

bin/setup

@@ -1 +1,9 @@
+#!/bin/bash -e
 
+source $OPENSHIFT_CARTRIDGE_SDK_BASH
+
+for dir in logs pid tmp env; do
+	mkdir -p $dir
+done
+
+mkdir -p $OPENSHIFT_DATA_DIR/.celery

conf.d/celeryconfig.py

@@ -0,0 +1,20 @@
+# Modify this file to configure celery. 
+# You can move it to your data directory and reference it in enviromental variable $OPENSHIFT_CELERY_CONFIG. 
+# You could also insert variables at build using an acton hook.
+# config options here http://docs.celeryproject.org/en/2.5/configuration.html#worker-celeryd
+
+import os
+import sys
+
+sys.path.append('.')
+
+BROKER_HOST = "localhost"
+BROKER_PORT = 5672
+BROKER_USER = "celeryuser"
+BROKER_PASSWORD = "celery"
+BROKER_VHOST = "celeryvhost"
+CELERYD_OPTS="--autoscale=3,1 --time-limit=300 --concurrency=2 "
+CELERY_RESULT_BACKEND = "amqp"
+
+# Here we check the env vars first, otherwise go with "tasks"
+CELERY_IMPORTS = ( os.getenv("OPENSHIFT_CELERY_IMPORTS","tasks") ,)

hooks/publish-celery-connection-info

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+echo "S_CELERY_MASTER=$CELERY_MASTER"
+echo "S_CELERY_HOST=$OPENSHIFT_GEAR_DNS"
+echo "S_CELERY_PORT=$OPENSHIFT_GEAR_PORT"

hooks/publish-db-connection-info

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+echo "OPENSHIFT_CELERY_HOST=$OPENSHIFT_GEAR_DNS"
+echo "OPENSHIFT_CELERY_PORT=$OPENSHIFT_GEAR_PORT"

hooks/set-celery-connection-info

@@ -0,0 +1,87 @@
+#!/usr/bin/ruby
+
+def gear_info(tokens, gear_id, &block)
+  gears = {}
+  if tokens.length == 1
+    gears[gear_id] = tokens.first
+  else
+    while not tokens.empty?
+      gear, delim, data = tokens.shift(3)
+      data = yield data if block_given?
+      raise "Invalid data" unless delim == '='
+      gears[gear] = data
+      tokens.shift if tokens.first == ' '
+    end
+  end
+  gears
+end
+
+def tokenize(s)
+  tokens = []
+  a = ""
+  state = :start
+  s.scan(/([ \t]+)|(\\')|(')|([^ \t']+)/) do |args|
+    space, escaped_delim, delim, text = args
+    case state
+    when :start
+      case
+      when space then " "
+      when escaped_delim then raise "Unexpected delimiter"
+      when delim then state = :within_delim
+      when text then tokens << text
+      else raise "error"
+      end
+    when :within_delim
+      case
+      when space then a << space
+      when escaped_delim then a << '"'
+      when delim then tokens << a; a = ""; state = :start
+      when text then a << text
+      end
+    end
+  end
+  tokens << a if a.length > 0
+  tokens
+end
+
+gear_id = ARGV.shift
+domain = ARGV.shift
+tokens = tokenize(ARGV.shift)
+
+# better if this is in creation order (oldest first)
+gears = gear_info(tokens, gear_id) do |d| 
+  d.split(' ').map{ |s| s.scan(/\A(.+?)=(.*?);?\Z/).first }.inject({}){ |h, (k,v)| h[k] = v if v != ''; h }
+end
+gears.each_pair{ |k,v| puts "Found gear #{k}#{k == gear_id ? '* ' : ''} with data #{v.inspect}" }
+gear_ids = gears.keys.sort.uniq
+
+puts "-------"
+
+was_master = ENV['CELERY_MASTER'] == '1'
+masters = gears.map{ |k,v| v['S_CELERY_MASTER'] == '1' ? k : nil }.compact.uniq.sort
+
+mode = 
+  case ENV['CELERY_MODE']
+  when 'read_replica' 
+    if masters.length > 1
+      masters = masters[0,1]
+    elsif masters.length == 0
+      masters = gears.keys.uniq.sort[0,1]
+    end
+    puts "Running in read replica mode with master #{masters}"
+    :read_replica
+  else
+    masters = gears.keys.uniq.sort
+    puts "Running sharded with masters #{masters.inspect}"
+    :sharded
+  end
+
+# An array of all of the host:port pairs for the cluster
+hosts = gears.map{ |k,v| "#{v['S_CELERY_HOST']}:#{v['S_CELERY_PORT']}" }.compact.uniq.sort
+# A list of key value pairs in <gear id> => <host:port pair> for the cluster
+members = gears.map{ |k,v| "#{k}=#{v['S_CELERY_HOST']}:#{v['S_CELERY_PORT']}" }.compact.uniq.sort
+
+File.open('env/CELERY_CLUSTER_MEMBERS', 'w'){ |f| f.puts members.join("\n") }
+File.open('env/CELERY_CLUSTER', 'w'){ |f| f.puts hosts.join(",") }
+File.open('env/CELERY_CLUSTER_MASTERS', 'w'){ |f| f.puts masters.join(",") }
+File.open('env/CELERY_MASTER', 'w'){ |f| f.puts masters.include?(gear_id) ? "1" : "0" }

metadata/managed_files.yml

@@ -3,5 +3,3 @@ locked_files:
 - env/
 - env/*
 - conf.d/*
-processed_templates:
-- '**/*.erb'