safe_celery_get

12840687 · IK · a0d175b0 · 12840687 · 12840687 · 12840687
Commit 12840687 authored Oct 23, 2026 by IK
Show whitespace changes
Inline Side-by-side

Showing with 112 additions and 18 deletions

circle/common/operations.py
+7 -3

circle/common/safe_celery_get.py
+69 -0

circle/dashboard/migrations/0010_auto_20261023_0834.py
+20 -0

circle/vm/operations.py
+16 -15

No files found.
--- a/circle/common/operations.py
+++ b/circle/common/operations.py
@@ -16,7 +16,8 @@
 # with CIRCLE.  If not, see <http://www.gnu.org/licenses/>.
 from inspect import getargspec
-from logging import getLogger
+#from logging import getLogger
+from celery.utils.log import get_task_logger
 from django.core.exceptions import PermissionDenied, ImproperlyConfigured
 from django.utils.translation import ugettext_noop
@@ -24,7 +25,8 @@ from django.utils.translation import ugettext_noop
 from .models import (activity_context, has_suffix, humanize_exception,
                     HumanReadableObject)
-logger = getLogger(__name__)
+logger = get_task_logger(__name__)
+#logger = getLogger(__name__)
 class SubOperationMixin(object):
@@ -305,7 +307,9 @@ class Operation(object):
                    kwargs)
        activity, allargs, auxargs = self.__prelude(kwargs)
        allargs['activity'] = activity
-        return self._exec_op(allargs, auxargs)
+	sync_ret = self._exec_op(allargs, auxargs)
+        logger.info("sync_ret %r", sync_ret)
+	return sync_ret
    def check_precond(self):
        pass

--- a/circle/common/safe_celery_get.py
+++ b/circle/common/safe_celery_get.py
+# safe_celery_get.py
+"""
+Drop-in replacement for Celery's .get() that works with mixed Celery 2.7 / 3 / 4 workers.
+Handles all result formats: raw Exception, Celery meta-dict, JSON, pickle, pickle_v2.
+"""
+class RemoteTaskError(Exception):
+    """Unified remote task failure exception (Celery 3/4 compatible)."""
+    def __init__(self, exc_type, message, traceback=None):
+        self.exc_type = exc_type
+        self.remote_message = message
+        self.remote_traceback = traceback
+        # Python 2.7: no f-strings
+        text = "[{0}] {1}".format(exc_type, message)
+        # Python 2.7-compatible super()
+        super(RemoteTaskError, self).__init__(text)
+def safe_celery_get(async_result, timeout=None):
+    """
+    Celery 3 and 4 compatible .get() replacement.
+    - Always calls get(propagate=False) so Celery never raises dict.
+    - Interprets Celery meta info manually.
+    - Reconstructs clean Python Exception for caller.
+    :param async_result: Celery AsyncResult object
+    :param timeout: timeout for result.wait/get
+    :return: task result or raises RemoteTaskError / Exception
+    """
+    # Prevent Celery from raising exceptions on our behalf
+    _ = async_result.get(timeout=timeout, propagate=False)
+    # Load full metadata
+    meta = async_result.backend.get_task_meta(async_result.id)
+    status = meta.get("status")
+    result = meta.get("result")
+    # SUCCESS return payload
+    if status == "SUCCESS":
+        return result
+    # FAILURE handling
+    # Case 1: If worker sent actual Exception (possible on Celery 3 + pickle)
+    if isinstance(result, BaseException):
+        raise result
+    # Case 2: Celery meta-dict with exc_type + exc_message (Celery 3 and 4)
+    if isinstance(result, dict) and "exc_type" in result and "exc_message" in result:
+        exc_type = result.get("exc_type") or "RemoteError"
+        exc_message = result.get("exc_message")
+        # e.g. ('msg',) or ['msg']
+        if isinstance(exc_message, (list, tuple)):
+            if exc_message:
+                exc_message = exc_message[0]
+            else:
+                exc_message = "<empty remote message>"
+        traceback = meta.get("traceback")
+        raise RemoteTaskError(exc_type, exc_message, traceback)
+    # Case 3: Some unknown format
+    raise Exception("Remote task failed with unknown result format: {0!r}".format(result))
--- a/circle/dashboard/migrations/0010_auto_20261023_0834.py
+++ b/circle/dashboard/migrations/0010_auto_20261023_0834.py
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.6 on 2026-10-23 06:34
+from __future__ import unicode_literals
+from django.db import migrations, models
+class Migration(migrations.Migration):
+    dependencies = [
+        ('dashboard', '0009_auto_20210201_1730'),
+    ]
+    operations = [
+        migrations.AlterField(
+            model_name='profile',
+            name='preferred_language',
+            field=models.CharField(choices=[(b'en', 'English'), (b'hu', 'Hungarian')], default=b'hu', max_length=32, verbose_name='preferred language'),
+        ),
+    ]
--- a/circle/vm/operations.py
+++ b/circle/vm/operations.py
@@ -29,6 +29,7 @@ import os
 import time
 from celery.contrib.abortable import AbortableAsyncResult
 from celery.exceptions import TimeLimitExceeded, TimeoutError
+from celery.utils.log import get_task_logger
 from django.conf import settings
 from django.core.exceptions import PermissionDenied, SuspiciousOperation
 from django.core.urlresolvers import reverse
@@ -57,7 +58,9 @@ from .tasks.local_tasks import (
    abortable_async_instance_operation, abortable_async_node_operation,
 )
-logger = getLogger(__name__)
+from common.safe_celery_get import safe_celery_get
+logger = get_task_logger(__name__)
 class RemoteOperationMixin(object):
@@ -66,9 +69,10 @@ class RemoteOperationMixin(object):
    def _operation(self, **kwargs):
        args = self._get_remote_args(**kwargs)
        logger.info("RemoteOpMixin %s %s %r", self.id, self._get_remote_queue(), args);
-        return self.task.apply_async(
+        ar = self.task.apply_async(
            args=args, queue=self._get_remote_queue()
-        ).get(timeout=self.remote_timeout)
+        )
+        return safe_celery_get(ar, timeout=self.remote_timeout)
    def check_precond(self):
        super(RemoteOperationMixin, self).check_precond()
@@ -425,6 +429,7 @@ class DeployOperation(InstanceOperation):
        self.instance.allocate_vnc_port()
        if node is not None:
            self.instance.node = node
+	    self.instance.capability_group = node.capability
            self.instance.save()
        else:
            self.instance.allocate_node()
@@ -597,7 +602,7 @@ class MigrateOperation(RemoteInstanceOperation):
            with activity.sub_activity('scheduling',
                                       readable_name=ugettext_noop(
                                           "schedule")) as sa:
-                to_node = self.instance.select_node()
+                to_node = self.instance.select_node(excl=self.instance.node)
                sa.result = to_node
        try:
@@ -784,6 +789,7 @@ class SaveAsTemplateOperation(InstanceOperation):
        try:
            self.instance._cleanup(parent_activity=activity, user=user)
        except:
+	    logger.error("Cleanup error")
            pass
        if with_shutdown:
@@ -989,8 +995,9 @@ class WakeUpOperation(InstanceOperation):
    def _operation(self, activity):
        # Schedule vm
        self.instance.allocate_vnc_port()
+	if self.instance.capability_group is None:
+	    self.instance.capability_group = "old"	# we suppose that this was suspened on old node
        self.instance.allocate_node()
        # Resume vm
        self.instance._wake_up_vm(parent_activity=activity)
@@ -1072,7 +1079,7 @@ class RenewOperation(InstanceOperation):
        return create_readable(ugettext_noop(
            "Renewed to suspend at %(suspend)s and destroy at %(delete)s."),
            suspend=self.instance.time_of_suspend,
-            delete=self.instance.time_of_suspend)
+            delete=self.instance.time_of_delete)
 @register_operation
@@ -1588,6 +1595,7 @@ class AgentStartedOperation(InstanceOperation):
            self.instance._set_time(parent_activity=activity)
            self.instance._set_hostname(parent_activity=activity)
+	logger.info("Agent_Started in instance: %s sys: %s ver: %s", self.instance, agent_system, old_version);
        new_version = settings.GET_AGENT_VERSION_BY_SYSTEM(agent_system)[0]
 #        if agent_system and new_version and (old_version is None or ("NOAGENTUPDATE" not in old_version and new_version != old_version)) :
        if agent_system and new_version and old_version and "NOAGENTUPDATE" not in old_version and new_version != old_version :
@@ -1707,19 +1715,12 @@ class UpdateAgentOperation(RemoteAgentOperation):
        _vers, _dir = settings.GET_AGENT_VERSION_BY_SYSTEM(agent_system)
        queue = self._get_remote_queue()
        instance = self.instance
-        if agent_system == "Windows":
+        if agent_system != "Linux":
            executable = sorted(os.listdir(_dir))[0]
            data = self.create_windows_tar(agent_system)
-        elif agent_system == "Linux":
-            executable = ""
-            data = self.create_linux_tar(agent_system)
        else:
-            # Legacy update method
            executable = ""
-            return agent_tasks.update_legacy.apply_async(
+            data = self.create_linux_tar(agent_system)
-                queue=queue,
-                args=(instance.vm_name, self.create_linux_tar())
-            ).get(timeout=60)
        checksum = md5(data).hexdigest()
        chunk_size = 1024 * 1024