Przeglądaj źródła

Task server Docker changes and backend tasks changes (#3688)

Change-Id: I1104a6e71530da51f4ea80f814c3bca19b59a5d6

Improving the URL routing and Task logs logic

Change-Id: I8d5bab4183950658412336ce05e1adceb6d73b1c

removing dead celery code

Change-Id: Ib1b1cfbd7dba1981658a5e645a499286d63f52a7

Co-authored-by: Athithyaa Selvam <aselvam@cloudera.com>
Athithyaa Selvam 1 rok temu
rodzic
commit
03fcee03d4

+ 183 - 0
apps/filebrowser/src/filebrowser/tasks.py

@@ -0,0 +1,183 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+import subprocess
+import os
+
+import shutil
+import psutil
+from datetime import datetime, timedelta
+
+from celery.utils.log import get_task_logger
+from celery import states
+from django.db import transaction
+from django.http import HttpRequest
+from django.utils import timezone
+
+from desktop.auth.backend import rewrite_user
+from desktop.celery import app
+from desktop.conf import TASK_SERVER
+from desktop.lib import fsmanager
+from useradmin.models import User
+from filebrowser.views import UPLOAD_CLASSES
+
+LOG_TASK = get_task_logger(__name__)
+LOG = logging.getLogger()
+STATE_MAP = {
+  'SUBMITTED': 'waiting',
+  states.RECEIVED: 'waiting',
+  states.PENDING: 'waiting',
+  states.STARTED: 'running',
+  states.RETRY: 'running',
+  'PROGRESS': 'running',
+  'AVAILABLE': 'available',
+  states.SUCCESS: 'available',
+  states.FAILURE: 'failure',
+  states.REVOKED: 'canceled',
+  states.REJECTED: 'rejected',
+  states.IGNORED: 'ignored'
+}
+
+@app.task
+def error_handler(request, exc, traceback):
+  print('Task {0} raised exception: {1!r}\n{2!r}'.format(
+    request.id, exc, traceback))
+
+@app.task()
+def upload_file_task(**kwargs):
+  task_id = kwargs.get("qquuid")
+  user_id = kwargs["user_id"]
+  scheme = kwargs["scheme"]
+  postdict = kwargs.get("postdict", None)
+  request = _get_request(postdict=postdict, user_id=user_id, scheme=scheme)
+  kwargs["username"] = request.user.username
+  kwargs["task_name"] = "fileupload"
+  kwargs["state"] = "STARTED"
+  now = timezone.now()
+  kwargs["task_start"] = now.strftime("%Y-%m-%dT%H:%M:%S")
+  upload_file_task.update_state(task_id=task_id, state='STARTED', meta=kwargs)
+  try:
+    upload_class = UPLOAD_CLASSES.get(kwargs["scheme"])
+    _fs = upload_class(request, args=[], **kwargs)
+    kwargs["state"] = "RUNNING"
+    upload_file_task.update_state(task_id=task_id, state='RUNNING', meta=kwargs)
+    _fs.upload_chunks()
+  except Exception as err:
+    kwargs["state"] = "FAILURE"
+    upload_file_task.update_state(task_id=task_id, state='FAILURE', meta=kwargs)
+    raise Exception(f"Upload failed %s" % err)
+
+  kwargs["state"] = "SUCCESS"
+  kwargs["started"] = now.strftime("%Y-%m-%dT%H:%M:%S")
+  kwargs["task_end"] = now.strftime("%Y-%m-%dT%H:%M:%S")
+  upload_file_task.update_state(task_id=task_id, state='SUCCESS', meta=kwargs)
+  return None
+
+def _get_request(postdict=None, user_id=None, scheme=None):
+  request = HttpRequest()
+  request.POST = postdict
+  request.fs_ref = "default"
+  request.fs = fsmanager.get_filesystem(request.fs_ref)
+  request.jt = None
+
+  user = User.objects.get(id=user_id)
+  user = rewrite_user(user)
+  request.user = user
+
+  return request
+
+@app.task()
+def document_cleanup_task(**kwargs):
+  keep_days = kwargs.get('keep_days')
+  task_id = kwargs.get("qquuid")
+  now = timezone.now()
+
+  kwargs["username"] = kwargs["username"]
+  kwargs["task_name"] = "document_cleanup"
+  kwargs["state"] = "STARTED"
+  kwargs["parameters"] = keep_days
+  kwargs["task_id"] = task_id
+  kwargs["progress"] = "0%"
+  kwargs["task_start"] = now.strftime("%Y-%m-%dT%H:%M:%S")
+
+  try:
+    INSTALL_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
+    document_cleanup_task.update_state(task_id=task_id, state='RUNNING', meta=kwargs)
+    subprocess.check_call([INSTALL_DIR + '/build/env/bin/hue', 'desktop_document_cleanup', f'--keep-days={keep_days}'])
+    kwargs["state"] = "SUCCESS"
+    LOG.info(f"Document_cleanup_task completed successfully.")
+  except Exception as err:
+    document_cleanup_task.update_state(task_id=task_id, state='FAILURE', meta=kwargs)
+    raise Exception(f"Upload failed %s" % err)
+
+  kwargs["state"] = "SUCCESS"
+  kwargs["progress"] = "100%"
+  kwargs["task_end"] = now.strftime("%Y-%m-%dT%H:%M:%S")
+  document_cleanup_task.update_state(task_id=task_id, state='SUCCESS', meta=kwargs)
+
+  return None
+
+@app.task()
+def check_disk_usage_and_clean_task(**kwargs):
+  task_id = kwargs.get("qquuid")
+  default_cleanup_threshold = 90
+  cleanup_threshold = kwargs.get('cleanup_threshold', default_cleanup_threshold)
+  username = kwargs.get("username", "celery_scheduler")
+  now = timezone.now()
+  kwargs = {
+    "username": username,
+    "task_name": "tmp_cleanup",
+    "task_id": task_id,
+    "parameters": cleanup_threshold,
+    "progress": "0%",
+    "task_start": now.strftime("%Y-%m-%dT%H:%M:%S")
+  }
+
+  disk_usage = psutil.disk_usage('/')
+  if disk_usage.percent >= int(cleanup_threshold):
+    LOG.info(f"Disk usage is above {cleanup_threshold}%, cleaning up /tmp directory...")
+    tmp_dir = '/tmp'
+    for filename in os.listdir(tmp_dir):
+      file_path = os.path.join(tmp_dir, filename)
+      try:
+        file_modified_time = datetime.fromtimestamp(os.path.getmtime(file_path))
+        # Make file_modified_time timezone-aware
+        file_modified_time = timezone.make_aware(file_modified_time, timezone.get_default_timezone())
+        if file_modified_time < now - timedelta(minutes=15):
+          if os.path.isfile(file_path) or os.path.islink(file_path):
+            os.unlink(file_path)
+          elif os.path.isdir(file_path):
+            shutil.rmtree(file_path)
+          LOG.info(f"Deleted {file_path}")
+      except PermissionError:
+        LOG.warning(f"Permission denied: unable to delete {file_path}")
+      except Exception as err:
+        check_disk_usage_and_clean_task.update_state(task_id=task_id, state='FAILURE', meta=kwargs)
+        raise Exception(f"Failed to delete {file_path}. Reason: {err}")
+
+    kwargs["progress"] = "100%"
+    check_disk_usage_and_clean_task.update_state(task_id=task_id, state='SUCCESS', meta=kwargs)
+    LOG.info("/tmp directory cleaned.")
+
+  else:
+    kwargs["progress"] = "100%"
+    check_disk_usage_and_clean_task.update_state(task_id=task_id, state='SUCCESS', meta=kwargs)
+    LOG.info(f"Disk usage is {disk_usage.percent}%, no need to clean up.")
+
+  # Get available disk space after cleanup
+  free_space = psutil.disk_usage('/tmp').free
+  return {'free_space': free_space}

+ 4 - 2
desktop/core/base_requirements.txt

@@ -12,8 +12,9 @@ cx-Oracle==8.3.0
 django-auth-ldap==4.3.0
 Django==3.2.20
 daphne==3.0.2
-django-celery-beat==2.2.1
-django-celery-results==2.4.0
+django-redis==5.4.0
+django-celery-beat==2.5.0
+django-celery-results==2.5.1
 django-cors-headers==3.7.0
 django-crequest==2018.5.11
 django-debug-panel==0.8.3
@@ -73,6 +74,7 @@ git+https://github.com/gethue/django-babel.git
 werkzeug==2.3.6  # Should remove it here and from devtools.mk
 django-utils-six==2.0
 six==1.16.0
+psutil==5.8.0
 -e file:///${ROOT}/desktop/core/ext-py3/requests-2.27.1
 -e file:///${ROOT}/desktop/core/ext-py3/boto-2.49.0
 -e file:///${ROOT}/desktop/core/ext-py3/django-axes-5.13.0

+ 147 - 1
desktop/core/src/desktop/api2.py

@@ -26,6 +26,17 @@ import sys
 import tempfile
 import zipfile
 
+from celery.app.control import Control
+from desktop.conf import TASK_SERVER
+if hasattr(TASK_SERVER, 'get') and TASK_SERVER.ENABLED.get():
+  from desktop.celery import app as celery_app
+import psutil
+import datetime
+import redis
+import re
+from django.http import HttpResponse, JsonResponse
+
+
 from collections import defaultdict
 from datetime import datetime
 
@@ -47,7 +58,7 @@ from beeswax.models import Namespace
 from desktop import appmanager
 from desktop.auth.backend import is_admin
 from desktop.conf import ENABLE_CONNECTORS, ENABLE_GIST_PREVIEW, CUSTOM, get_clusters, IS_K8S_ONLY, ENABLE_SHARING
-from desktop.conf import ENABLE_NEW_STORAGE_BROWSER, ENABLE_CHUNKED_FILE_UPLOADER
+from desktop.conf import ENABLE_NEW_STORAGE_BROWSER, ENABLE_CHUNKED_FILE_UPLOADER, TASK_SERVER
 from desktop.lib.conf import BoundContainer, GLOBAL_CONFIG, is_anonymous
 from desktop.lib.django_util import JsonResponse, login_notrequired, render
 from desktop.lib.exceptions_renderable import PopupException
@@ -57,6 +68,9 @@ from desktop.lib.paths import get_desktop_root
 from desktop.models import Document2, Document, Directory, FilesystemException, uuid_default, \
   UserPreferences, get_user_preferences, set_user_preferences, get_cluster_config, __paginate, _get_gist_document
 from desktop.views import get_banner_message, serve_403_error
+from desktop.log import DEFAULT_LOG_DIR
+from filebrowser.tasks import check_disk_usage_and_clean_task
+from filebrowser.tasks import document_cleanup_task
 
 from hadoop.cluster import is_yarn
 
@@ -101,6 +115,7 @@ def get_config(request):
   config['hue_config']['is_yarn_enabled'] = is_yarn()
   config['hue_config']['enable_new_storage_browser'] = ENABLE_NEW_STORAGE_BROWSER.get()
   config['hue_config']['enable_chunked_file_uploader'] = ENABLE_CHUNKED_FILE_UPLOADER.get()
+  config['hue_config']['enable_task_server'] = TASK_SERVER.ENABLED.get()
   config['clusters'] = list(get_clusters(request.user).values())
   config['documents'] = {
     'types': list(Document2.objects.documents(user=request.user).order_by().values_list('type', flat=True).distinct())
@@ -688,6 +703,137 @@ def share_document(request):
     'document': doc.to_dict()
   })
 
+@api_error_handler
+@require_POST
+def handle_submit(request):
+  # Extract the task name and params from the request
+  try:
+    data = json.loads(request.body)
+    task_name = data.get('taskName')
+    task_params = data.get('taskParams')
+  except json.JSONDecodeError as e:
+    return JsonResponse({'error': str(e)}, status=500)
+
+  if task_name == 'document cleanup':
+    keep_days = task_params.get('keep-days')
+    task_kwargs = {
+      'keep_days': keep_days,
+      'user_id': request.user.id,
+      'username': request.user.username,
+    }
+    # Enqueue the document cleanup task with keyword arguments
+    task = document_cleanup_task.apply_async(kwargs=task_kwargs)
+
+    # Return a response indicating the task has been scheduled
+    return JsonResponse({
+      'taskName': task_name,
+      'taskParams': task_params,
+      'status': 'Scheduled',
+      'task_id': task.id,  # The task ID generated by Celery
+    })
+
+  elif task_name == 'tmp clean up':
+    cleanup_threshold = task_params.get('threshold for clean up')
+    disk_check_interval = task_params.get('disk check interval')
+    task_kwargs = {
+      'username': request.user.username,
+      'cleanup_threshold': cleanup_threshold,
+      'disk_check_interval': disk_check_interval
+    }
+    task = check_disk_usage_and_clean_task.apply_async(kwargs=task_kwargs)
+
+    return JsonResponse({
+      'taskName': task_name,
+      'status': 'Scheduled',
+      'task_id': task.id,  # The task ID generated by Celery
+    })
+
+  return JsonResponse({
+    'status': 0
+  })
+
+@api_error_handler
+def get_taskserver_tasks(request):
+  """Retirve the tasks from the database"""
+  redis_client = redis.Redis(host='localhost', port=6379, db=0)
+  tasks = []
+
+  # Use scan_iter to efficiently iterate over keys matching the first pattern
+  for key in redis_client.scan_iter('celery-task-meta-*'):
+    task = json.loads(redis_client.get(key))
+    tasks.append(task)
+
+  # Use scan_iter to efficiently iterate over keys matching the second pattern
+  for key in redis_client.scan_iter('task:*'):
+    task = json.loads(redis_client.get(key))
+    tasks.append(task)
+
+  return JsonResponse(tasks, safe=False)
+
+@api_error_handler
+def check_upload_status(request, task_id):
+  redis_client = redis.Redis(host='localhost', port=6379, db=0)
+  task_key = f'celery-task-meta-{task_id}'
+  task_data = redis_client.get(task_key)
+
+  if task_data is None:
+    return JsonResponse({'error': 'Task not found'}, status=404)
+
+  task = json.loads(task_data)
+  is_finalized = task.get('status') == 'SUCCESS'
+  is_running = task.get('status') == 'RUNNING'
+  is_failure = task.get('status') == 'FAILURE'
+  is_revoked = task.get('status') == 'REVOKED'
+
+  return JsonResponse({'isFinalized': is_finalized, 'isRunning': is_running, 'isFailure': is_failure, 'isRevoked': is_revoked})
+
+@api_error_handler
+def kill_task(request, task_id):
+  # Check the current status of the task
+  status_response = check_upload_status(request, task_id)
+  status_data = json.loads(status_response.content)
+
+  if status_data.get('isFinalized') or status_data.get('isRevoked') or status_data.get('isFailure'):
+    return JsonResponse({'status': 'info', 'message': f'Task {task_id} has already been completed or revoked.'})
+
+  try:
+    control = Control(app=celery_app)
+    control.revoke(task_id, terminate=True)
+    return JsonResponse({'status': 'success', 'message': f'Task {task_id} has been terminated.'})
+  except Exception as e:
+    return JsonResponse({'status': 'error', 'message': f'Failed to terminate task {task_id}: {str(e)}'})
+
+
+def get_available_space(request):
+  free_space = psutil.disk_usage('/tmp').free
+  return JsonResponse({'free_space': free_space})
+
+def get_task_logs(request, task_id):
+  log_dir = os.getenv("DESKTOP_LOG_DIR", DEFAULT_LOG_DIR)
+  log_file = "%s/rungunicornserver.log" % (log_dir)
+  task_log = []
+  escaped_task_id = re.escape(task_id)
+
+  # Using a simpler and more explicit regex to debug
+  task_end_pattern = re.compile(rf"\[{escaped_task_id}\].*succeeded")
+  task_start_pattern = re.compile(rf"\[{escaped_task_id}\].*received")
+  try:
+    with open(log_file, 'r') as file:
+      recording = False
+      for line in file:
+        if task_start_pattern.search(line):
+          recording = True  # Start recording log lines
+        if recording:
+          task_log.append(line)
+        if task_end_pattern.search(line) and recording:
+          break  # Stop recording after finding the end of the task
+
+  except FileNotFoundError:
+    return HttpResponse(f'Log file not found at {log_file}', status=404)
+  except Exception as e:
+    return HttpResponse(str(e), status=500)
+
+  return HttpResponse(''.join(task_log), content_type='text/plain')
 
 @api_error_handler
 @require_POST

+ 38 - 21
desktop/core/src/desktop/celery.py

@@ -26,35 +26,52 @@ from celery.schedules import crontab
 
 from desktop.conf import TASK_SERVER
 from desktop.settings import TIME_ZONE, INSTALLED_APPS
-
+if hasattr(TASK_SERVER, 'get') and TASK_SERVER.ENABLED.get():
+  from desktop.settings import CELERY_RESULT_BACKEND, CELERY_BROKER_URL
+from django.utils import timezone
 
 # Set the default Django settings module for the 'celery' program.
 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'desktop.settings')
 
-app = Celery('desktop')
-
-# Using a string here means the worker doesn't have to serialize
-# the configuration object to child processes.
-# - namespace='CELERY' means all celery-related configuration keys
-#   should have a `CELERY_` prefix.
-app.config_from_object('django.conf:settings', namespace='CELERY')
-app.conf.timezone = TIME_ZONE
+class HueCelery(Celery):
+  def gen_task_name(self, name, module):
+    if module.endswith('.tasks'):
+      module = module[:-6]
+    return super().gen_task_name(name, module)
 
-# Load task modules from all registered Django app configs.
-app.autodiscover_tasks()
+  # Method to configure the beat_schedule
+  def setup_beat_schedule(self):
+    now = timezone.now()
 
+    self.conf.beat_schedule = {
+      'check_disk_usage_and_clean_task': {
+      'task': 'filebrowser.check_disk_usage_and_clean_task',
+      'schedule': 1000.0,  # Run every 1000 seconds
+      'args': (),
+      'kwargs': {'cleanup_threshold': 90},  # Provide task arguments if needed
+      },
+    }
 
-@app.task(bind=True)
-def debug_task(self):
-  print('Request: {0!r}'.format(self.request))
-  return 'Hello'
+if hasattr(TASK_SERVER, 'get') and TASK_SERVER.ENABLED.get():
+  app = HueCelery('desktop', backend=CELERY_RESULT_BACKEND, broker=CELERY_BROKER_URL)
+  app.conf.broker_transport_options = {'visibility_timeout': 3600}  # 1 hour.
+  app.conf.result_key_prefix = 'desktop_'
 
+  # Call the setup_beat_schedule method
+  app.setup_beat_schedule()
 
-if 'django_celery_beat' in INSTALLED_APPS and False: # Config not available yet
-  app.conf.beat_schedule = {}
+  # Using a string here means the worker doesn't have to serialize
+  # the configuration object to child processes.
+  # - namespace='CELERY' means all celery-related configuration keys
+  #   should have a `CELERY_` prefix.
+  app.config_from_object('django.conf:settings', namespace='CELERY')
+  app.conf.timezone = TIME_ZONE
 
-  if TASK_SERVER.BEAT_SCHEDULES_FILE.get():
-    schedules = imp.load_source('schedules', TASK_SERVER.BEAT_SCHEDULES_FILE.get())
+  # Load task modules from all registered Django app configs.
+  app.autodiscover_tasks()
 
-    for schedule in schedules.periodic_tasks:
-      app.conf.beat_schedule.update(schedule)
+else:
+  app = Celery('desktop')
+  app.config_from_object('django.conf:settings', namespace='CELERY')
+  app.conf.timezone = TIME_ZONE
+  app.autodiscover_tasks()

+ 27 - 10
desktop/core/src/desktop/management/commands/runcelery.py

@@ -21,6 +21,7 @@ import sys
 
 from desktop import conf
 from desktop.lib.daemon_utils import drop_privileges_if_necessary
+from desktop.log import DEFAULT_LOG_DIR
 
 from django.core.management.base import BaseCommand
 from django.utils import autoreload
@@ -62,6 +63,14 @@ class Command(BaseCommand):
         type=str,
         default='DEBUG'
     )
+    parser.add_argument('--beat')
+    parser.add_argument(
+        '--schedule_file',
+        type=str,
+        required=True,
+        default='celerybeat-schedule',
+        help='Path to the celerybeat-schedule file'
+    )
 
   def handle(self, *args, **options):
     runcelery(*args, **options)
@@ -71,23 +80,31 @@ class Command(BaseCommand):
 
 def runcelery(*args, **options):
   # Native does not load Hue's config
-  # CeleryCommand().handle_argv(['worker', '--app=desktop.celery', '--concurrency=1', '--loglevel=DEBUG'])
+  log_dir = os.getenv("DESKTOP_LOG_DIR", DEFAULT_LOG_DIR)
+  log_file = "%s/rungunicornserver.log" % (log_dir)
+  concurrency = int(conf.GUNICORN_NUMBER_OF_WORKERS.get()/4) or options['concurrency']
+  schedule_file = options['schedule_file']
   opts = [
-      'runcelery',
-      'worker',
-      '--app=' + options['app'],
-      '--concurrency=' + str(options['concurrency']),
-      '--loglevel=' + options['loglevel']
+    'celery',
+    '--app=' + options['app'],
+    'worker',
+    '--loglevel=' + options['loglevel'],
+    '--concurrency=' + str(concurrency),
+    '--beat',
+    '-s', schedule_file,
+    '--logfile=' + log_file
   ]
   drop_privileges_if_necessary(CELERY_OPTIONS)
 
-  if conf.DEV.get():
-    autoreload.main(celery_main, (opts,))
-  else:
-    celery_main(opts)
+  # Set command-line arguments for Celery
+  sys.argv = opts
+
+  # Call the Celery main function
+  celery_main()
 
   LOG.info("Stopping command '%s'" % ' '.join(opts))
   sys.exit(-1)
 
+
 if __name__ == '__main__':
   runcelery(sys.argv[1:])

+ 8 - 0
desktop/core/src/desktop/urls.py

@@ -93,6 +93,7 @@ else:
 
 dynamic_patterns += [
   re_path(r'^logs$', desktop_views.log_view, name="desktop.views.log_view"),
+  re_path(r'^task_server$', desktop_views.task_server_view, name='desktop.views.task_server_view'),
   re_path(r'^desktop/log_analytics$', desktop_views.log_analytics),
   re_path(r'^desktop/log_js_error$', desktop_views.log_js_error),
   re_path(r'^desktop/dump_config$', desktop_views.dump_config, name="desktop.views.dump_config"),
@@ -167,6 +168,13 @@ dynamic_patterns += [
   re_path(r'^desktop/api2/doc/restore/?$', desktop_api2.restore_document),
   re_path(r'^desktop/api2/doc/share/link/?$', desktop_api2.share_document_link),
   re_path(r'^desktop/api2/doc/share/?$', desktop_api2.share_document),
+  re_path(r'^desktop/api2/taskserver/handle_submit/?$', desktop_api2.handle_submit, name="desktop_api2.handle_submit"),
+  re_path(r'^desktop/api2/taskserver/get_taskserver_tasks/?$', desktop_api2.get_taskserver_tasks, name="desktop_api2.get_taskserver_tasks"),
+  re_path(r'^desktop/api2/taskserver/get_task_logs/(?P<task_id>[^/]+)/?$', desktop_api2.get_task_logs, name="desktop_api2.get_task_logs"),
+  re_path(r'^desktop/api2/taskserver/check_upload_status/(?P<task_id>[^/]+)/?$', desktop_api2.check_upload_status,
+          name="desktop_api2.check_upload_status"),
+  re_path(r'^desktop/api2/taskserver/kill_task/(?P<task_id>[^/]+)/?$', desktop_api2.kill_task, name="desktop_api2.kill_task"),
+  re_path(r'^desktop/api2/taskserver/get_available_space/?$', desktop_api2.get_available_space, name="desktop_api2.get_available_space"),
 
   re_path(r'^desktop/api2/get_config/?$', desktop_api2.get_config),
   re_path(r'^desktop/api2/get_hue_config/?$', desktop_api2.get_hue_config),

+ 13 - 0
desktop/core/src/desktop/views.py

@@ -67,6 +67,7 @@ from desktop.log import set_all_debug as _set_all_debug, reset_all_debug as _res
 from desktop.models import Settings, hue_version, _get_apps, UserPreferences
 from libsaml.conf import REQUIRED_GROUPS, REQUIRED_GROUPS_ATTRIBUTE
 from useradmin.models import get_profile
+from useradmin.models import User
 
 if sys.version_info[0] > 2:
   from io import StringIO as string_io
@@ -289,6 +290,18 @@ def log_view(request):
     )
   )
 
+def task_server_view(request):
+  """
+  This view renders the Task Server page with basic functionality.
+  """
+  # You can add more logic here if needed
+  LOG.debug("task_server_view called with request: %s", request)
+  return render('taskserver_list_tasks.mako', request, {
+    'message': _(''),
+    'is_embeddable': request.GET.get('is_embeddable', False),
+    'users': User.objects.all(),
+    'users_json': json.dumps(list(User.objects.values_list('id', flat=True)))
+  })
 
 @hue_admin_required
 @access_log_level(logging.WARN)

+ 13 - 0
tools/container/base/hue/Dockerfile

@@ -39,6 +39,19 @@ RUN set -eux; \
         xmlsec1-openssl \
         zlib-devel
 
+# Install redis
+RUN set -eux; \
+    dnf install -y gcc make && \
+    curl -O http://download.redis.io/redis-stable.tar.gz && \
+    tar xzf redis-stable.tar.gz && \
+    cd redis-stable && \
+    make && \
+    make install && \
+    cd .. && \
+    rm -rf redis-stable redis-stable.tar.gz && \
+    dnf clean all && \
+    rm -rf /var/cache/yum
+
 RUN set -eux; \
       /usr/bin/pip3.8 install supervisor \
       && curl -s https://files.pythonhosted.org/packages/45/78/4621eb7085162bc4d2252ad92af1cc5ccacbd417a50e2ee74426331aad18/psycopg2_binary-2.9.3-cp38-cp38-musllinux_1_1_x86_64.whl -o /tmp/psycopg2_binary-2.9.3-cp38-cp38-musllinux_1_1_x86_64.whl \

+ 13 - 0
tools/container/compile/hue/Dockerfile

@@ -45,6 +45,19 @@ RUN set -eux; \
       xmlsec1-openssl \
       zlib-devel
 
+# Install redis
+RUN set -eux; \
+    dnf install -y gcc make && \
+    curl -O http://download.redis.io/redis-stable.tar.gz && \
+    tar xzf redis-stable.tar.gz && \
+    cd redis-stable && \
+    make && \
+    make install && \
+    cd .. && \
+    rm -rf redis-stable redis-stable.tar.gz && \
+    dnf clean all && \
+    rm -rf /var/cache/yum
+
 RUN set -eux; \
       curl -sL https://rpm.nodesource.com/setup_14.x | bash - \
         && yum install -y nodejs \

+ 3 - 1
tools/container/hue/Dockerfile

@@ -56,12 +56,14 @@ ADD supervisor-files/etc/supervisord.conf /etc/supervisord.conf
 ADD supervisor-files/hue_server.conf /etc/supervisor.d/
 ADD supervisor-files/hue_ktrenewer.conf /etc/supervisor.d/
 ADD supervisor-files/hue_loglistener.conf /etc/supervisor.d/
+ADD supervisor-files/hue_celery.conf /etc/supervisor.d/
+ADD supervisor-files/hue_redis.conf /etc/supervisor.d/
 RUN chown -R ${HUEUSER}:${HUEUSER} /etc/supervisord.conf && chown -R ${HUEUSER}:${HUEUSER} /etc/supervisor.d
 
 # Remove chardet package
 RUN rm -rf ${HUE_HOME}/build/env/lib/python3.8/site-packages/pip/_vendor/chardet
 
-EXPOSE 8888 9111
+EXPOSE 8888 9111 6379
 
 # Switch to non-root default user
 USER hive

+ 25 - 0
tools/container/hue/hue.sh

@@ -111,6 +111,19 @@ function set_samlcert() {
   export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/lib64
 }
 
+function start_celery() {
+  echo "Starting Celery worker..."
+  # The schedule of periodic tasks performed by celery-beat is stored in the celerybeat-schedule file.
+  touch $HUE_HOME/celerybeat-schedule
+  chmod 644 $HUE_HOME/celerybeat-schedule
+  $HUE_BIN/hue runcelery worker --app desktop.celery --loglevel=DEBUG --schedule_file $HUE_HOME/celerybeat-schedule
+}
+
+function start_redis() {
+  echo "Starting Redis server..."
+  redis-server --port 6379 --daemonize no
+}
+
 # If database connectivity is not set then fail
 ret=$(db_connectivity_check)
 if [[ $ret == "fail" ]];  then
@@ -134,6 +147,18 @@ elif [[ $1 == rungunicornserver ]]; then
   fix_column_issue "axes_accesslog" "trusted"
   fix_column_issue "axes_accessattempt" "trusted"
   $HUE_BIN/hue rungunicornserver
+elif [[ $1 == start_celery ]]; then
+  if awk '/^\[\[task_server\]\]/{flag=1;next}/^\[/{flag=0}flag && /enabled=True/{print "Task server enabled"; exit}' $HUE_CONF_DIR/zhue_safety_valve.ini || \
+     awk '/^\[\[task_server\]\]/{flag=1;next}/^\[/{flag=0}flag && /enabled=True/{print "Task server enabled"; exit}' $HUE_CONF_DIR/zhue.ini || \
+     awk '/^\[\[task_server\]\]/{flag=1;next}/^\[/{flag=0}flag && /enabled=True/{print "Task server enabled"; exit}' $HUE_CONF_DIR/hue.ini; then
+    start_celery
+  fi
+elif [[ $1 == start_redis ]]; then
+  if awk '/^\[\[task_server\]\]/{flag=1;next}/^\[/{flag=0}flag && /enabled=True/{print "Task server enabled"; exit}' $HUE_CONF_DIR/zhue_safety_valve.ini || \
+     awk '/^\[\[task_server\]\]/{flag=1;next}/^\[/{flag=0}flag && /enabled=True/{print "Task server enabled"; exit}' $HUE_CONF_DIR/zhue.ini || \
+     awk '/^\[\[task_server\]\]/{flag=1;next}/^\[/{flag=0}flag && /enabled=True/{print "Task server enabled"; exit}' $HUE_CONF_DIR/hue.ini; then
+    start_redis
+  fi
 fi
 
 exit 0

+ 18 - 0
tools/container/hue/supervisor-files/hue_celery_template

@@ -0,0 +1,18 @@
+[program:hue_celery]
+directory=${HUE_HOME}
+command=${HUE_HOME}/hue.sh start_celery
+autostart=true
+startretries=3
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
+stderr_logfile=/dev/stderr
+stderr_logfile_maxbytes=0
+environment=DESKTOP_LOG_DIR="/var/log/${HUEUSER}",PYTHON_EGG_CACHE="${HUE_CONF_DIR}/.python-eggs",HUE_PROCESS_NAME="hue_celery"
+user=${HUEUSER}
+group=${HUEUSER}
+exitcodes=0,2
+autorestart=true
+startsecs=10
+stopwaitsecs=30
+killasgroup=true
+# rlimits

+ 18 - 0
tools/container/hue/supervisor-files/hue_redis_template

@@ -0,0 +1,18 @@
+[program:hue_redis]
+directory=${HUE_HOME}
+command=${HUE_HOME}/hue.sh start_redis
+autostart=true
+startretries=3
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
+stderr_logfile=/dev/stderr
+stderr_logfile_maxbytes=0
+environment=DESKTOP_LOG_DIR="/var/log/${HUEUSER}",PYTHON_EGG_CACHE="${HUE_CONF_DIR}/.python-eggs",HUE_PROCESS_NAME="hue_redis"
+user=${HUEUSER}
+group=${HUEUSER}
+exitcodes=0,2
+autorestart=true
+startsecs=10
+stopwaitsecs=30
+killasgroup=true
+# rlimits