Celery

Justfile

@@ -28,6 +28,12 @@ JUST_DEFAULTIFY_FUNCTIONS+=(terra_caseify)
 function Terra_Pipenv()
 {
   if [[ ${TERRA_LOCAL-} == 1 ]]; then
+    if [ -n "${VIRTUAL_ENV+set}" ]; then
+      echo "Warning: You appear to be in a virtual env" >&2
+      echo "Deactivate external virtual envs before running just" >&2
+      ask_question "Continue?" answer_continue n
+      [ "$answer_continue" == "0" ] && return 1
+    fi
     PIPENV_PIPFILE="${TERRA_CWD}/Pipfile" pipenv ${@+"${@}"} || return $?
   else
     Just-docker-compose -f "${TERRA_CWD}/docker-compose-main.yml" run ${TERRA_PIPENV_IMAGE-terra} pipenv ${@+"${@}"} || return $?
@@ -115,10 +121,6 @@ function terra_caseify()
       extra_args=$#
       ;;
 
-    run_redis) # Run redis
-      Just-docker-compose -f "${TERRA_CWD}/docker-compose.yml" run redis ${@+"${@}"}
-      extra_args=$#
-      ;;
     run_celery) # Starts a celery worker
       local node_name
       if [[ ${TERRA_LOCAL-} == 1 ]]; then
@@ -127,7 +129,22 @@ function terra_caseify()
         node_name="docker@%h"
       fi
 
-      Terra_Pipenv run celery -A terra.executor.celery.app worker --loglevel="${TERRA_CELLER_LOG_LEVEL-INFO}" -n "${node_name}"
+      # Untested
+      if [ "${OS-}" = "Windows_NT" ]; then
+        # https://www.distributedpython.com/2018/08/21/celery-4-windows/
+        local FORKED_BY_MULTIPROCESSING
+        export FORKED_BY_MULTIPROCESSING=1
+      fi
+
+      # We might be able to use CELERY_LOADER to avoid the -A argument
+      Terra_Pipenv run python -m terra.executor.celery -A terra.executor.celery.app worker --loglevel="${TERRA_CELERY_LOG_LEVEL-INFO}" -n "${node_name}"
+      ;;
+
+    run_flower) # Start the flower server
+      Terra_Pipenv run python -m terra.executor.celery -A terra.executor.celery.app flower
+      ;;
+    shutdown_celery) # Shuts down all celery works on all nodes
+      Terra_Pipenv run python -c "from terra.executor.celery import app; app.control.broadcast('shutdown')"
       ;;
 
     ### Run Debugging containers ###
@@ -203,7 +220,8 @@ function terra_caseify()
     terra_test-pep8) # Run pep8 test
       justify terra pep8
       echo "Running flake8..."
-      Terra_Pipenv run bash -c 'flake8 \
+      Terra_Pipenv run bash -c 'cd ${TERRA_TERRA_DIR};
+                                flake8 \
                                 "${TERRA_TERRA_DIR}/terra"'
       ;;
 

README.md

@@ -17,3 +17,13 @@ source 'setup.env'
 just terra sync
 just terra run
 ```
+
+## Running an app in celery
+
+1. `just terra up` - To start redis queue (only once)
+2. `just run celery` - To start a celery worker (run on each worker node)
+3. `just run dsm ...` - To start processing job
+
+When done
+4. `just shutdown celery` - To shutdown _all_ celery workers on _all_ nodes
+5. `just terra down` - To shutdown redis.

docker-compose-main.yml

@@ -6,17 +6,17 @@ services:
       dockerfile: docker/terra.Dockerfile
     # prevent different users from clobbering each others images
     image: ${TERRA_DOCKER_REPO}:terra_${TERRA_USERNAME}
-    environment:
+    environment: &terra_environment
       # Variables for docker_entrypoint.bsh
-      - DOCKER_UID=${TERRA_UID}
-      - DOCKER_GIDS=${TERRA_GIDS}
-      - DOCKER_GROUP_NAMES=${TERRA_GROUP_NAMES}
-      - DOCKER_USERNAME=user
-      - DISPLAY
-      - JUSTFILE=${TERRA_TERRA_DIR_DOCKER}/docker/terra.Justfile
-      - JUST_SETTINGS=${TERRA_TERRA_DIR_DOCKER}/terra.env
-      - PYTHONPATH=${TERRA_PYTHONPATH-}
-      - TZ
+      DOCKER_UID: ${TERRA_UID}
+      DOCKER_GIDS: ${TERRA_GIDS}
+      DOCKER_GROUP_NAMES: ${TERRA_GROUP_NAMES}
+      DOCKER_USERNAME: user
+      JUSTFILE: ${TERRA_TERRA_DIR_DOCKER}/docker/terra.Justfile
+      JUST_SETTINGS: ${TERRA_TERRA_DIR_DOCKER}/terra.env
+      PYTHONPATH: ${TERRA_PYTHONPATH-}
+      DISPLAY:
+      TZ:
     cap_add:
       - SYS_PTRACE # Useful for gdb
     volumes:
@@ -38,6 +38,12 @@ services:
         source: terra-venv
         target: /venv
 
+  terra-demo:
+    <<: *terra
+    environment:
+      <<: *terra_environment
+      TERRA_SETTINGS_FILE:
+
   redis-commander:
     image: rediscommander/redis-commander
     ports:
@@ -46,14 +52,14 @@ services:
       - source: redis_secret
         target: ${TERRA_REDIS_SECRET_DOCKER}
       - source: redis_commander_secret
-        target: ${TERRA_REDIS_COMMANDER_SECRET_FILE}
+        target: ${TERRA_REDIS_COMMANDER_SECRET}
     command: |
       sh -c '
         echo -n '"'"'{
           "connections":[
             {
               "password": "'"'"' > /redis-commander/config/local-production.json
-        cat /run/secrets/redis_password | sed '"'"'s|\\|\\\\|g;s|"|\\"|g'"'"' >> /redis-commander/config/local-production.json
+        cat /run/secrets/${TERRA_REDIS_SECRET_DOCKER} | sed '"'"'s|\\|\\\\|g;s|"|\\"|g'"'"' >> /redis-commander/config/local-production.json
         echo -n '"'"'",
               "host": "${TERRA_REDIS_HOSTNAME_DOCKER}",
               "label": "terra",
@@ -68,7 +74,7 @@ services:
             "httpAuth": {
               "username": "admin",
               "passwordHash": "'"'"'>> /redis-commander/config/local-production.json
-          cat "/run/secrets/${TERRA_REDIS_COMMANDER_SECRET_FILE}" | sed '"'"'s|\\|\\\\|g;s|"|\\"|g'"'"' >> /redis-commander/config/local-production.json
+          cat "/run/secrets/${TERRA_REDIS_COMMANDER_SECRET}" | sed '"'"'s|\\|\\\\|g;s|"|\\"|g'"'"' >> /redis-commander/config/local-production.json
           echo '"'"'"
             }
           }

docs/conf.py

@@ -51,6 +51,7 @@
 intersphinx_mapping = {
     'python': ('https://docs.python.org/3.6', None),
     'vsi_common': ('https://visionsystemsinc.github.io/vsi_common/', None),
+    'celery': ('https://docs.celeryproject.org/en/stable/', None)
 }
 
 # Add any paths that contain templates here, relative to this directory.
@@ -103,6 +104,8 @@
     ('py:class', 'json.encoder.JSONEncoder'),
     ('py:class', 'concurrent.futures._base.Executor'),
     ('py:class', 'concurrent.futures._base.Future'),
+    ('py:class', 'concurrent.futures.process.ProcessPoolExecutor'),
+    ('py:class', 'concurrent.futures.thread.ThreadPoolExecutor'),
     ('py:class', 'argparse._AppendAction'),
     ('py:data',  'logging.DEBUG'),
     ('py:data',  'logging.WARNING'),

docs/terra/settings.rst

@@ -1,6 +1,19 @@
 
 .. _settings:
 
+Terra Settings
+--------------
+
+.. option:: terra.zone
+
+    Terra can be running in one of three areas of execution, or "zones": the master controller (``controller``), a service runner (``runner``), or a task (``task``). The different zones could all be running on the main host, or other containers or computers, depending on the compute and executor.
+
+    The master controller includes: the CLI, workflow, stage and service definitions layers.
+
+    This variable is automatically updated, and should only be read.
+
+    Default: ``controller``
+
 Workflow Settings
 -----------------
 

setup.py

@@ -11,6 +11,8 @@
       extra_requires=extra_requires,
       install_requires=[
         "jstyleson",
-        "envcontext"
+        "envcontext",
+        # I use signal and task from celery, no matter what
+        "celery"
       ]
 )

terra.env

@@ -72,16 +72,20 @@ fi
 
 if [[ ! -f /.dockerenv && ! -s "${TERRA_REDIS_SECRET_FILE}" ]]; then
   source "${VSI_COMMON_DIR}/linux/random.bsh"
-  # No quotes allowed
-  urandom_password 20 '\x21\x23-\x26\x28-\x7E' > "${TERRA_REDIS_SECRET_FILE}"
+  # Allow printable ascii characters excpet quotes, ';' (for an unknown redis/celery parsing reason), ':' or '@' (for redis url reasons)
+  urandom_password 20 '\x21\x23-\x26\x28-\x39\x3c-\x3f\x41-\x7E' > "${TERRA_REDIS_SECRET_FILE}"
 fi
 
 #**
 # .. envvar:: TERRA_CELERY_MAIN_NAME
 #
-# Name of the main module if running as __main__. This is used as the prefix for auto-generated task names.
+# (Optional) Name of the main module if running as __main__. This is used as the prefix for auto-generated task names that are defined in the same module as ``__main__`` (Usually caused by ``python -m``). At first, python will try ``sys.modules['__main__'].__spec__.name``, before using this value, when that fails.
+#
+# .. envvar:: TERRA_KEEP_TEMP_DIR
+#
+# Optional environment variable, that when set to ``1`` will keep the temporary config files generated for containers. For debug use.
 #**
-: ${TERRA_CELERY_MAIN_NAME=terra}
+
 #**
 # .. envvar:: TERRA_CELERY_CONF
 #

terra/compute/base.py

@@ -1,8 +1,17 @@
 import os
-
+import time
+import atexit
+from logging import StreamHandler
+from logging.handlers import SocketHandler
+import threading
+import warnings
+
+from terra import settings
 import terra.compute.utils
 from terra.executor import Executor
-from terra.logger import getLogger
+from terra.logger import (
+  getLogger, LogRecordSocketReceiver, SkipStdErrAddFilter
+)
 logger = getLogger(__name__)
 
 
@@ -63,9 +72,6 @@ def add_volume(self, local, remote, flags=None, prefix=None,
     self._validate_volume(local, remote, local_must_exist=local_must_exist)
     self.volumes.append((local, remote))
 
-  def get_volume_map(self, config, service_info):
-    return []
-
   def pre_run(self):
     '''
     A function that runs before the run service
@@ -75,7 +81,13 @@ def pre_run(self):
     :class:`terra.compute.base.BaseService` is mainly responsible for handling
     Executors that need a separate volume translation
     '''
-    self.executor_configuration_map = Executor.configuration_map(self)
+
+    # The executor volume map is calculated on the host side, where all the
+    # information is available. For example if using docker and celery, then
+    # docker config need to be run to get the container volumes, and that has
+    # to be run on the host machine. So this is calculated here.
+    settings.executor.volume_map = Executor.configuration_map(self)
+    logger.debug4("Executor Volume map: %s", settings.executor.volume_map)
 
   def post_run(self):
     pass
@@ -156,6 +168,9 @@ def defaultCommand(self, service_class, *args, **kwargs):
     # bind function and return it
     return defaultCommand.__get__(self, type(self))
 
+  def get_volume_map(self, config, service_info):
+    return []
+
   def run_service(self, *args, **kwargs):
     '''
     Place holder for code to run an instance in the compute. Runs
@@ -178,5 +193,89 @@ def configuration_map_service(self, service_info):
 
     return service_info.volumes
 
+  @staticmethod
+  def configure_logger(sender, **kwargs):
+    if settings.terra.zone == 'controller':
+      # Setup log file for use in configure
+      sender._log_file = os.path.join(
+          settings.processing_dir,
+          terra.logger._SetupTerraLogger.default_log_prefix)
+      os.makedirs(settings.processing_dir, exist_ok=True)
+      sender._log_file = open(sender._log_file, 'a')
+      sender.main_log_handler = StreamHandler(stream=sender._log_file)
+      sender.root_logger.addHandler(sender.main_log_handler)
+
+      # setup the TCP socket listener
+      sender.tcp_logging_server = LogRecordSocketReceiver(
+          settings.logging.server.hostname, settings.logging.server.port)
+      listener_thread = threading.Thread(
+          target=sender.tcp_logging_server.serve_until_stopped)
+      listener_thread.setDaemon(True)
+      listener_thread.start()
+
+      # Wait up to a second, to make sure the thread started
+      for _ in range(1000):
+        if sender.tcp_logging_server.ready:
+          break
+        time.sleep(0.001)
+      else:  # pragma: no cover
+        warnings.warn("TCP Logging server thread did not startup. "
+                      "This is probably not a problem, unless logging isn't "
+                      "working.", RuntimeWarning)
+
+      # Auto cleanup
+      @atexit.register
+      def cleanup_thread():
+        sender.tcp_logging_server.abort = 1
+        listener_thread.join(timeout=5)
+        if listener_thread.is_alive():  # pragma: no cover
+          warnings.warn("TCP Logger Server Thread did not shut down "
+                        "gracefully. Attempting to exit anyways.",
+                        RuntimeWarning)
+    elif settings.terra.zone == 'runner':
+      sender.main_log_handler = SocketHandler(
+          settings.logging.server.hostname, settings.logging.server.port)
+      # By default, all runners have access to the master controllers stderr,
+      # so there is no need for the master controller to echo out the log
+      # messages a second time.
+      sender.main_log_handler.addFilter(SkipStdErrAddFilter())
+      sender.root_logger.addHandler(sender.main_log_handler)
+
+  @staticmethod
+  def reconfigure_logger(sender, **kwargs):
+    # sender is logger in this case
+    #
+    # The default logging handler is a StreamHandler. This will reconfigure its
+    # output stream
+
+    if settings.terra.zone == 'controller':
+      log_file = os.path.join(
+          settings.processing_dir,
+          terra.logger._SetupTerraLogger.default_log_prefix)
+
+      # Check to see if _log_file is unset. If it is, this is due to _log_file
+      # being called without configure being called. While it is not important
+      # this work, it's more likely for unit testsing
+      # if not os.path.samefile(log_file, sender._log_file.name):
+      if getattr(sender, '_log_file', None) is not None and \
+         log_file != sender._log_file.name:
+        os.makedirs(settings.processing_dir, exist_ok=True)
+        sender._log_file.close()
+        sender._log_file = open(log_file, 'a')
+    elif settings.terra.zone == 'runner':
+      # Only if it's changed
+      if settings.logging.server.hostname != sender.main_log_handler.host or \
+         settings.logging.server.port != sender.main_log_handler.port:
+        # Reconnect Socket Handler
+        sender.main_log_handler.close()
+        try:
+          sender.root_logger.removeHandler(sender.main_log_handler)
+        except ValueError:  # pragma: no cover
+          pass
+
+        sender.main_log_handler = SocketHandler(
+            settings.logging.server.hostname, settings.logging.server.port)
+        sender.root_logger.addHandler(sender.main_log_handler)
+
 
 services = {}