Concurrent delete fix

- Fix concurrent acquire lock issue
- Add cluster name for events (during delete)

deployer/tasks/deployment.py

@@ -232,12 +232,13 @@ def delete(name, version=None):
         versions are undeployed.
     :return:
     """
-    search_params = {
+    search_params = create_search_parameters({
         'deployment': {
             'name': name,
-            'version': version or 'all'
+            'version': version or 'all',
+            'id': ''
         }
-    }
+    })
     return _using_lock.si(
         search_params, name,
         do_task=(
@@ -247,7 +248,7 @@ def delete(name, version=None):
             _fleet_undeploy.si(name, version) |
             _wait_for_undeploy.si(name, version, search_params=search_params)
         )
-    )()
+    ).apply_async()
 
 
 @app.task
@@ -280,7 +281,7 @@ def _using_lock(self, search_params, name, do_task, cleanup_tasks=None,
     try:
         lock = LockService().apply_lock(name)
     except ResourceLockedException as lock_error:
-        self.retry(exc=lock_error)
+        raise self.retry(exc=lock_error)
 
     _release_lock_s = _release_lock.si(lock)
     cleanup_tasks = cleanup_tasks or []