Merge branch 'dev'

Author: joestubbs

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.pythonPath": "/usr/bin/python3"
+}

CHANGELOG.md

@@ -2,6 +2,22 @@
 All notable changes to this project will be documented in this file.
 
 
+## 1.7.0 - 2020-11-10
+### Added
+- Added support for a cron scheduling feature for actors. The cron schedule feature allows users to 
+instruct Abaco to automatically execute actors based on a schedule provided by the user. More
+information available from the docs (https://tacc-cloud.readthedocs.io/projects/abaco/en/latest/technical/messages.html#cron-schedule).
+- Added support for configuring Abaco with a DockerHub credential to be used when pulling images
+from DockerHub. In particular, Abaco can be configured with the credentials of a licensed account
+with increased pull quota to avoid "toomanyrequests: You have reached your pull rate limit" errors from the Docker daemon. 
+
+### Changed
+- No change.
+
+### Removed
+- No change.
+
+
 ## 1.6.0 - 2020-04-30
 ### Added
 - Added the `GET /actors/search/{search_type}?{search_terms}` endpoint for mongo database full-text search and matching

Makefile

@@ -45,6 +45,10 @@ deploy:
 build-core:
 	@docker build -t abaco/core:$$TAG ./
 
+# Builds prometheus locally
+build-prom:
+	@docker build -t abaco/prom:$$TAG prometheus/.
+
 
 # Builds nginx
 build-nginx:

abaco.conf

@@ -109,6 +109,21 @@ show_traceback: false
 # Here we set the to 12 hours.
 log_ex: 43200
 
+# Max amount of time that any log can be set.
+log_ex_limit: 86400
+
+# Amount of time, in seconds, to which sd2e's logs will be set.
+sd2e_log_ex: 10800
+
+# Max amount of time, in seconds, to which sd2e's logs can be set.
+sd2e_log_ex_limit: 21600
+
+# Max amount of time, in seconds, to which this DEV's logs can be set.
+DEV_log_ex_limit: 30000
+
+# Amount of time, in seconds, to which this DEV's logs will be set.
+DEV_log_ex: 15000
+
 # Max length (in bytes) to store an actor execution's log. If a log exceeds this length, the log will be truncated.
 # Note: max_log_length must not exceed the maximum document length for the log store.
 # here we default it to 1 MB

abaco.log

@@ -0,0 +1 @@
+

actors/config.py

@@ -11,7 +11,7 @@
 
 
 def read_config():
-    parser = ConfigParser()
+    parser = ConfigParser(interpolation=None)
     places = ['/service.conf',
               '/etc/service.conf']
     place = places[0]

actors/controllers.py

@@ -12,6 +12,7 @@
 from flask_restful import Resource, Api, inputs
 from werkzeug.exceptions import BadRequest
 from agaveflask.utils import RequestParser, ok
+from parse import parse
 
 from auth import check_permissions, get_tas_data, tenant_can_use_tas, get_uid_gid_homedir, get_token_default
 from channels import ActorMsgChannel, CommandChannel, ExecutionResultsChannel, WorkerChannel
@@ -65,6 +66,68 @@ def get(self, search_type):
         return ok(result=result, msg="Search completed successfully.")
 
 
+class CronResource(Resource):
+    def get(self):
+        logger.debug("HERE I AM IN GET /cron")
+        actor_ids = [actor['db_id'] for actor in actors_store.items()]
+        logger.debug(f"actor ids are {actor_ids}")
+        # Loop through all actor ids to check for cron schedules
+        for actor_id in actor_ids:
+            # Create actor based on the actor_id
+            actor = actors_store[actor_id]
+            logger.debug(f"cron_on equals {actor.get('cron_on')} for actor {actor_id}")
+            try:
+                # Check if next execution == UTC current time
+                if self.cron_execution_datetime(actor):
+                    # Check if cron switch is on
+                    if actor.get('cron_on'):
+                        d = {}
+                        logger.debug("the current time is the same as the next cron scheduled, adding execution")
+                        # Execute actor
+                        before_exc_time = timeit.default_timer()
+                        exc = Execution.add_execution(actor_id, {'cpu': 0,
+                                                'io': 0,
+                                                'runtime': 0,
+                                                'status': codes.SUBMITTED,
+                                                'executor': 'cron'})
+                        logger.debug("execution has been added, now making message")
+                        # Create & add message to the queue 
+                        d['Time_msg_queued'] = before_exc_time
+                        d['_abaco_execution_id'] = exc
+                        d['_abaco_Content_Type'] = 'str'
+                        ch = ActorMsgChannel(actor_id=actor_id)
+                        ch.put_msg(message="This is your cron execution", d=d)
+                        ch.close()
+                        logger.debug("Message added to actor inbox. id: {}.".format(actor_id))
+                        # Update the actor's next execution
+                        actors_store[actor_id, 'cron_next_ex'] = Actor.set_next_ex(actor, actor_id)
+                    else:
+                        logger.debug("Actor's cron is not activated, but next execution will be incremented")
+                        actors_store[actor_id, 'cron_next_ex'] = Actor.set_next_ex(actor, actor_id)
+                else:
+                    logger.debug("now is not the time")
+            except:
+                logger.debug("Actor has no cron setup")
+
+    def cron_execution_datetime(self, actor):
+        logger.debug("inside cron_execution_datetime method")
+        now = get_current_utc_time()
+        now = datetime.datetime(now.year, now.month, now.day, now.hour)
+        logger.debug(f"the current utc time is {now}")
+        # Get cron execution datetime
+        cron = actor['cron_next_ex']
+        logger.debug(f"cron_next_ex is {cron}")
+        # Parse the next execution into a list of the form: [year,month,day,hour]
+        cron_datetime = parse("{}-{}-{} {}", cron)
+        logger.debug(f"cron datetime is {cron_datetime}")
+        # Create a datetime out of cron_datetime
+        cron_execution = datetime.datetime(int(cron_datetime[0]), int(cron_datetime[1]), int(cron_datetime[2]), int(cron_datetime[3]))
+        logger.debug(f"cron execution is {cron_execution}")
+        # Return true/false comparing now with the next cron execution
+        logger.debug(f"does cron == now? {cron_execution == now}")
+        return cron_execution == now
+
+
 class MetricsResource(Resource):
     def get(self):
         enable_autoscaling = Config.get('workers', 'autoscaling')
@@ -769,6 +832,36 @@ def post(self):
         else:
             token = get_token_default()
         args['token'] = token
+         # adding check for 'log_ex'
+        if 'logEx' in args and args.get('logEx') is not None:
+            log_ex = int(args.get('logEx'))
+            args['log_ex'] = log_ex
+        # cron attribute
+        cron = None
+        if Config.get('web', 'case') == 'camel':
+            logger.debug("Case is camel")
+            if 'cronSchedule' in args and args.get('cronSchedule') is not None:
+                cron = args.get('cronSchedule')
+        else:
+            if 'cron_schedule' in args and args.get('cron_schedule') is not None:
+                logger.debug("Case is snake")
+                cron = args.get('cron_schedule')
+        if cron is not None:
+            logger.debug("Cron has been posted")
+            # set_cron checks for the 'now' alias 
+            # It also checks that the cron schedule is greater than or equal to the current UTC time
+            r = Actor.set_cron(cron)
+            logger.debug(f"r is {r}")
+            if r.fixed[2] in ['hours', 'hour', 'days', 'day', 'weeks', 'week', 'months', 'month']: 
+                args['cron_schedule'] = cron
+                logger.debug(f"setting cron_next_ex to {r.fixed[0]}")
+                args['cron_next_ex'] = r.fixed[0]
+                args['cron_on'] = True
+            else:
+                raise BadRequest(f'{r.fixed[2]} is an invalid unit of time')
+                args['cron_on'] = False
+        else:
+            logger.debug("Cron schedule was not sent in")
         if Config.get('web', 'case') == 'camel':
             max_workers = args.get('maxWorkers')
             args['max_workers'] = max_workers
@@ -891,6 +984,36 @@ def put(self, actor_id):
         args = self.validate_put(actor)
         logger.debug("PUT args validated successfully.")
         args['tenant'] = g.tenant
+        cron = None
+        if 'logEx' in args and args.get('logEx') is not None:
+            log_ex = int(args.get('logEx'))
+            logger.debug(f"log_ex in args; using: {log_ex}")
+            args['log_ex'] = log_ex
+        # Check for both camel and snake case
+        if Config.get('web', 'case') == 'camel':
+            if 'cronSchedule' in args and args.get('cronSchedule') is not None:
+                cron = args.get('cronSchedule')
+            if 'cronOn' in args and args.get('cronOn') is not None:
+                actor['cron_on'] = args.get('cronOn')
+        else:
+            if 'cron_schedule' in args and args.get('cron_schedule') is not None:
+                cron = args.get('cron_schedule')
+            if 'cron_on' in args and args.get('cron_on') is not None:
+                actor['cron_on'] = args.get('cron_on')
+        if cron is not None:
+            # set_cron checks for the 'now' alias 
+            # It also checks that the cron schedule is greater than or equal to the current UTC time
+            # Check for proper unit of time
+            r = Actor.set_cron(cron)
+            if r.fixed[2] in ['hours', 'hour', 'days', 'day', 'weeks', 'week', 'months', 'month']: 
+                args['cron_schedule'] = cron
+                logger.debug(f"setting cron_next_ex to {r.fixed[0]}")
+                args['cron_next_ex'] = r.fixed[0]
+            else:
+                raise BadRequest(f'{r.fixed[2]} is an invalid unit of time')
+                args['cron_on'] = False
+        else:
+            logger.debug("No cron schedule has been sent")
         if args['queue']:
             queues_list = Config.get('spawner', 'host_queues').replace(' ', '')
             valid_queues = queues_list.split(',')
@@ -970,6 +1093,7 @@ def validate_put(self, actor):
             actor.pop('max_workers')
             actor.pop('mem_limit')
             actor.pop('max_cpus')
+            actor.pop('log_ex')
 
         # this update overrides all required and optional attributes
         try:

actors/docker_utils.py

@@ -5,6 +5,7 @@
 import time
 import timeit
 import datetime
+import random
 
 import docker
 from requests.packages.urllib3.exceptions import ReadTimeoutError
@@ -17,7 +18,7 @@
 from config import Config
 from codes import BUSY, READY, RUNNING
 import globals
-from models import Execution, get_current_utc_time, display_time
+from models import Actor, Execution, get_current_utc_time, display_time
 from stores import workers_store
 
 
@@ -52,13 +53,66 @@ class DockerStartContainerError(DockerError):
 class DockerStopContainerError(DockerError):
     pass
 
+def get_docker_credentials():
+    """
+    Get the docker credentials from the config.
+    """
+    # we try to get as many credentials as have been
+    creds = []
+    cnt = 1
+    while True:
+        try:
+            username = Config.get('docker', f'dockerhub_username_{cnt}')
+            password = Config.get('docker', f'dockerhub_password_{cnt}')
+        except:
+            break
+        if not username or not password:
+            break
+        creds.append({'username': username, 'password': password})
+        cnt = cnt + 1
+    return creds
+
+
+dockerhub_creds = get_docker_credentials()
+
+
+def get_random_dockerhub_cred():
+    """
+    Chose a dockerhub credential at random
+    """
+    if len(dockerhub_creds) == 0:
+        return None, None
+    creds = random.choice(dockerhub_creds)
+    try:
+        username = creds['username']
+        password = creds['password']
+    except Exception as e:
+        logger.debug("Got exception trying to get dockerhub credentials")
+        return None, None
+    return username, password
+
+
+def cli_login(cli, username, password):
+    """
+    Try to login a dockerhub cli with a username and password
+    """
+    try:
+        cli.login(username=username, password=password)
+    except Exception as e:
+        logger.error(f"Could not login using dockerhub creds; username: {username}."
+                     f"Exception: {e}")
+
+
 def rm_container(cid):
     """
     Remove a container.
     :param cid:
     :return:
     """
     cli = docker.APIClient(base_url=dd, version="auto")
+    username, password = get_random_dockerhub_cred()
+    if username and password:
+        cli_login(cli, username, password)
     try:
         rsp = cli.remove_container(cid, force=True)
     except Exception as e:
@@ -74,6 +128,9 @@ def pull_image(image):
     """
     logger.debug("top of pull_image()")
     cli = docker.APIClient(base_url=dd, version="auto")
+    username, password = get_random_dockerhub_cred()
+    if username and password:
+        cli_login(cli, username, password)
     try:
         rsp = cli.pull(repository=image)
     except Exception as e:
@@ -607,6 +664,7 @@ def execute_actor(actor_id,
     # a counter of the number of iterations through the main "running" loop;
     # this counter is used to determine when less frequent actions, such as log aggregation, need to run.
     loop_idx = 0
+    log_ex = Actor.get_actor_log_ttl(actor_id)
     while running and not globals.force_quit:
         loop_idx += 1
         logger.debug("top of while running loop; loop_idx: {}".format(loop_idx))
@@ -667,7 +725,7 @@ def execute_actor(actor_id,
         # grab the logs every 5th iteration --
         if loop_idx % 5 == 0:
             logs = cli.logs(container.get('Id'))
-            Execution.set_logs(execution_id, logs, actor_id, tenant, worker_id)
+            Execution.set_logs(execution_id, logs, actor_id, tenant, worker_id, log_ex)
             logs = None
 
         # checking the container status to see if it is still running ----

actors/metrics_api.py

@@ -4,7 +4,7 @@
 
 from agaveflask.utils import AgaveApi, handle_error
 
-from controllers import MetricsResource
+from controllers import MetricsResource, CronResource
 
 from errors import errors
 
@@ -28,6 +28,7 @@
 
 # Resources
 api.add_resource(MetricsResource, '/metrics')
+api.add_resource(CronResource, '/cron')
 
 if __name__ == '__main__':
     app.run(host='0.0.0.0', debug=True)