Revision 22f54174
| b/snf-cyclades-app/synnefo/logic/backend.py | ||
|---|---|---|
| 89 | 89 |
# failed server |
| 90 | 90 |
serial = vm.serial |
| 91 | 91 |
if job_status == rapi.JOB_STATUS_SUCCESS: |
| 92 |
quotas.accept_serial(serial)
|
|
| 92 |
quotas.accept_resource_serial(vm)
|
|
| 93 | 93 |
elif job_status in [rapi.JOB_STATUS_ERROR, rapi.JOB_STATUS_CANCELED]: |
| 94 | 94 |
log.debug("Job %s failed. Rejecting related serial %s", job_id,
|
| 95 | 95 |
serial) |
| 96 |
quotas.reject_serial(serial) |
|
| 97 |
vm.serial = None |
|
| 96 |
quotas.reject_resource_serial(vm) |
|
| 98 | 97 |
elif job_status == rapi.JOB_STATUS_SUCCESS: |
| 99 | 98 |
commission_info = quotas.get_commission_info(resource=vm, |
| 100 | 99 |
action=action, |
| ... | ... | |
| 103 | 102 |
# Commission for this change has not been issued, or the issued |
| 104 | 103 |
# commission was unaware of the current change. Reject all previous |
| 105 | 104 |
# commissions and create a new one in forced mode! |
| 106 |
log.debug("Expected job was %s. Processing job %s.",
|
|
| 107 |
vm.task_job_id, job_id) |
|
| 105 |
log.debug("Expected job was %s. Processing job %s. "
|
|
| 106 |
"Attached serial %s", |
|
| 107 |
vm.task_job_id, job_id, vm.serial) |
|
| 108 | 108 |
reason = ("client: dispatcher, resource: %s, ganeti_job: %s"
|
| 109 | 109 |
% (vm, job_id)) |
| 110 |
quotas.handle_resource_commission(vm, action, |
|
| 111 |
action_fields=job_fields, |
|
| 112 |
commission_name=reason, |
|
| 113 |
force=True, |
|
| 114 |
auto_accept=True) |
|
| 115 |
log.debug("Issued new commission: %s", vm.serial)
|
|
| 116 |
# NOTE: Since we rejected the serial that was associated with the |
|
| 117 |
# 'vm.task_job_id' job, we must also clear the 'vm.serial' field. |
|
| 118 |
# If not, there will be no new commission for the 'vm.task_job_id' |
|
| 119 |
# job! |
|
| 120 |
vm.serial = None |
|
| 121 |
|
|
| 110 |
serial = quotas.handle_resource_commission( |
|
| 111 |
vm, action, |
|
| 112 |
action_fields=job_fields, |
|
| 113 |
commission_name=reason, |
|
| 114 |
force=True, |
|
| 115 |
auto_accept=True) |
|
| 116 |
log.debug("Issued new commission: %s", serial)
|
|
| 122 | 117 |
return vm |
| 123 | 118 |
|
| 124 | 119 |
|
| b/snf-cyclades-app/synnefo/logic/servers.py | ||
|---|---|---|
| 138 | 138 |
log.debug("Rejecting commission: '%s', could not perform"
|
| 139 | 139 |
" action '%s': %s" % (vm.serial, action, e)) |
| 140 | 140 |
transaction.rollback() |
| 141 |
quotas.reject_serial(vm.serial)
|
|
| 141 |
quotas.reject_resource_serial(vm)
|
|
| 142 | 142 |
transaction.commit() |
| 143 | 143 |
raise |
| 144 | 144 |
|
| ... | ... | |
| 147 | 147 |
# commission because the VM has been stored in DB. Also, if |
| 148 | 148 |
# communication with Ganeti fails, the job will never reach |
| 149 | 149 |
# Ganeti, and the commission will never be resolved. |
| 150 |
quotas.accept_serial(vm.serial)
|
|
| 150 |
quotas.accept_resource_serial(vm)
|
|
| 151 | 151 |
|
| 152 | 152 |
log.info("user: %s, vm: %s, action: %s, job_id: %s, serial: %s",
|
| 153 | 153 |
user_id, vm.id, action, job_id, vm.serial) |
| b/snf-cyclades-app/synnefo/logic/tests/servers.py | ||
|---|---|---|
| 197 | 197 |
vm.task_job_id = None |
| 198 | 198 |
vm.save() |
| 199 | 199 |
with mocked_quotaholder(): |
| 200 |
quotas.accept_serial(vm.serial)
|
|
| 200 |
quotas.accept_resource_serial(vm)
|
|
| 201 | 201 |
mrapi().RebootInstance.return_value = 1 |
| 202 | 202 |
with mocked_quotaholder(): |
| 203 | 203 |
servers.reboot(vm, "HARD") |
| b/snf-cyclades-app/synnefo/quotas/__init__.py | ||
|---|---|---|
| 43 | 43 |
log = logging.getLogger(__name__) |
| 44 | 44 |
|
| 45 | 45 |
|
| 46 |
QUOTABLE_RESOURCES = [VirtualMachine, Network, IPAddress] |
|
| 47 |
|
|
| 48 |
|
|
| 46 | 49 |
DEFAULT_SOURCE = 'system' |
| 47 | 50 |
RESOURCES = [ |
| 48 | 51 |
"cyclades.vm", |
| ... | ... | |
| 115 | 118 |
force=force, |
| 116 | 119 |
auto_accept=auto_accept) |
| 117 | 120 |
|
| 118 |
if serial: |
|
| 119 |
serial_info = {"serial": serial}
|
|
| 120 |
if auto_accept: |
|
| 121 |
serial_info["pending"] = False |
|
| 122 |
serial_info["accept"] = True |
|
| 123 |
serial_info["resolved"] = True |
|
| 124 |
return QuotaHolderSerial.objects.create(**serial_info) |
|
| 125 |
else: |
|
| 121 |
if not serial: |
|
| 126 | 122 |
raise Exception("No serial")
|
| 127 | 123 |
|
| 124 |
serial = QuotaHolderSerial.objects.create(serial=serial) |
|
| 125 |
if auto_accept: |
|
| 126 |
serial.pending = False |
|
| 127 |
serial.accept = True |
|
| 128 |
serial.resolved = True |
|
| 129 |
serial.save() |
|
| 130 |
else: |
|
| 131 |
resource.serial = serial |
|
| 132 |
resource.save() |
|
| 133 |
return serial |
|
| 134 |
|
|
| 128 | 135 |
|
| 129 |
def accept_serial(serial, strict=True): |
|
| 130 |
assert serial.pending or serial.accept |
|
| 131 |
response = resolve_commissions(accept=[serial.serial], strict=strict) |
|
| 136 |
def mark_accept_serial(serial): |
|
| 132 | 137 |
serial.pending = False |
| 133 | 138 |
serial.accept = True |
| 134 |
serial.resolved = True |
|
| 135 | 139 |
serial.save() |
| 136 |
return response |
|
| 137 | 140 |
|
| 138 | 141 |
|
| 139 |
def reject_serial(serial, strict=True): |
|
| 140 |
assert serial.pending or not serial.accept |
|
| 141 |
response = resolve_commissions(reject=[serial.serial], strict=strict) |
|
| 142 |
serial.pending = False |
|
| 143 |
serial.accept = False |
|
| 144 |
serial.resolved = True |
|
| 145 |
serial.save() |
|
| 146 |
return response |
|
| 142 |
def accept_resource_serial(resource, strict=True): |
|
| 143 |
serial = resource.serial |
|
| 144 |
assert serial.pending or serial.accept, "%s can't be accepted" % serial |
|
| 145 |
response = _resolve_commissions(accept=[serial.serial], strict=strict) |
|
| 146 |
if serial.serial in response.get("accepted", []):
|
|
| 147 |
resource.serial = None |
|
| 148 |
resource.save() |
|
| 147 | 149 |
|
| 148 | 150 |
|
| 149 |
def resolve_commissions(accept=None, reject=None, strict=True): |
|
| 151 |
def reject_resource_serial(resource, strict=True): |
|
| 152 |
serial = resource.serial |
|
| 153 |
assert serial.pending or not serial.accept, "%s can't be rejected" % serial |
|
| 154 |
response = _resolve_commissions(reject=[serial.serial], strict=strict) |
|
| 155 |
if serial.serial in response.get("rejected", []):
|
|
| 156 |
resource.serial = None |
|
| 157 |
resource.save() |
|
| 158 |
|
|
| 159 |
|
|
| 160 |
def _resolve_commissions(accept=None, reject=None, strict=True): |
|
| 150 | 161 |
if accept is None: |
| 151 | 162 |
accept = [] |
| 152 | 163 |
if reject is None: |
| ... | ... | |
| 156 | 167 |
with AstakosClientExceptionHandler(): |
| 157 | 168 |
response = qh.resolve_commissions(accept, reject) |
| 158 | 169 |
|
| 159 |
# Update correspodning entries in DB |
|
| 160 |
QuotaHolderSerial.objects.filter(serial__in=accept).update(accept=True, |
|
| 161 |
pending=False, |
|
| 162 |
resolved=True) |
|
| 163 |
QuotaHolderSerial.objects.filter(serial__in=reject).update(accept=False, |
|
| 164 |
pending=False, |
|
| 165 |
resolved=True) |
|
| 170 |
accepted = response.get("accepted", [])
|
|
| 171 |
rejected = response.get("rejected", [])
|
|
| 172 |
|
|
| 173 |
if accepted: |
|
| 174 |
QuotaHolderSerial.objects.filter(serial__in=accepted).update( |
|
| 175 |
accept=True, pending=False, resolved=True) |
|
| 176 |
if rejected: |
|
| 177 |
QuotaHolderSerial.objects.filter(serial__in=rejected).update( |
|
| 178 |
accept=False, pending=False, resolved=True) |
|
| 166 | 179 |
|
| 167 | 180 |
if strict: |
| 168 | 181 |
failed = response["failed"] |
| ... | ... | |
| 173 | 186 |
return response |
| 174 | 187 |
|
| 175 | 188 |
|
| 189 |
def reconcile_resolve_commissions(accept=None, reject=None, strict=True): |
|
| 190 |
response = _resolve_commissions(accept=accept, |
|
| 191 |
reject=reject, |
|
| 192 |
strict=strict) |
|
| 193 |
affected = response.get("accepted", []) + response.get("rejected", [])
|
|
| 194 |
for resource in QUOTABLE_RESOURCES: |
|
| 195 |
resource.objects.filter(serial__in=affected).update(serial=None) |
|
| 196 |
|
|
| 197 |
|
|
| 176 | 198 |
def resolve_pending_commissions(): |
| 177 | 199 |
"""Resolve quotaholder pending commissions. |
| 178 | 200 |
|
| ... | ... | |
| 249 | 271 |
""" |
| 250 | 272 |
commission_reason = ("client: api, resource: %s, action: %s"
|
| 251 | 273 |
% (resource, action)) |
| 252 |
serial = handle_resource_commission(resource=resource, action=action,
|
|
| 253 |
action_fields=action_fields,
|
|
| 254 |
commission_name=commission_reason)
|
|
| 274 |
handle_resource_commission(resource=resource, action=action, |
|
| 275 |
action_fields=action_fields, |
|
| 276 |
commission_name=commission_reason) |
|
| 255 | 277 |
|
| 256 | 278 |
# Mark the serial as one to accept and associate it with the resource |
| 257 |
serial.pending = False |
|
| 258 |
serial.accept = True |
|
| 259 |
serial.save() |
|
| 279 |
mark_accept_serial(resource.serial) |
|
| 260 | 280 |
transaction.commit() |
| 261 | 281 |
|
| 262 | 282 |
try: |
| 263 | 283 |
# Accept the commission to quotaholder |
| 264 |
accept_serial(serial)
|
|
| 284 |
accept_resource_serial(resource)
|
|
| 265 | 285 |
except: |
| 266 | 286 |
# Do not crash if we can not accept commission to Quotaholder. Quotas |
| 267 | 287 |
# have already been reserved and the resource already exists in DB. |
| 268 | 288 |
# Just log the error |
| 269 |
log.exception("Failed to accept commission: %s", serial)
|
|
| 270 |
|
|
| 271 |
return serial |
|
| 289 |
log.exception("Failed to accept commission: %s", resource.serial)
|
|
| 272 | 290 |
|
| 273 | 291 |
|
| 274 | 292 |
def get_commission_info(resource, action, action_fields=None): |
| ... | ... | |
| 348 | 366 |
# The one who succeeds will be finally accepted, and all other will be |
| 349 | 367 |
# rejected |
| 350 | 368 |
force = force or (action == "DESTROY") |
| 351 |
resolve_commission(resource.serial, force=force)
|
|
| 369 |
resolve_resource_commission(resource, force=force)
|
|
| 352 | 370 |
|
| 353 | 371 |
serial = issue_commission(resource, action, name=commission_name, |
| 354 | 372 |
force=force, auto_accept=auto_accept, |
| 355 | 373 |
action_fields=action_fields) |
| 356 |
resource.serial = serial |
|
| 357 |
resource.save() |
|
| 358 | 374 |
return serial |
| 359 | 375 |
|
| 360 | 376 |
|
| ... | ... | |
| 362 | 378 |
pass |
| 363 | 379 |
|
| 364 | 380 |
|
| 365 |
def resolve_commission(serial, force=False): |
|
| 381 |
def resolve_resource_commission(resource, force=False): |
|
| 382 |
serial = resource.serial |
|
| 366 | 383 |
if serial is None or serial.resolved: |
| 367 | 384 |
return |
| 368 | 385 |
if serial.pending and not force: |
| ... | ... | |
| 370 | 387 |
raise ResolveError(m) |
| 371 | 388 |
log.warning("Resolving pending commission: %s", serial)
|
| 372 | 389 |
if not serial.pending and serial.accept: |
| 373 |
accept_serial(serial)
|
|
| 390 |
accept_resource_serial(resource)
|
|
| 374 | 391 |
else: |
| 375 |
reject_serial(serial) |
|
| 392 |
reject_resource_serial(resource) |
|
| b/snf-cyclades-app/synnefo/quotas/management/commands/reconcile-commissions-cyclades.py | ||
|---|---|---|
| 63 | 63 |
|
| 64 | 64 |
if fix and (accepted or rejected): |
| 65 | 65 |
self.stdout.write("Fixing pending commissions..\n")
|
| 66 |
quotas.resolve_commissions(accept=accepted, reject=rejected, |
|
| 67 |
strict=False) |
|
| 66 |
quotas.reconcile_resolve_commissions(accept=accepted, |
|
| 67 |
reject=rejected, |
|
| 68 |
strict=False) |
|
| 68 | 69 |
|
| 69 | 70 |
|
| 70 | 71 |
def list_to_string(l): |
| b/snf-django-lib/snf_django/utils/testing.py | ||
|---|---|---|
| 204 | 204 |
return (len(astakos.return_value.issue_one_commission.mock_calls) + |
| 205 | 205 |
serial) |
| 206 | 206 |
astakos.return_value.issue_one_commission.side_effect = foo |
| 207 |
astakos.return_value.resolve_commissions.return_value = {"failed": []}
|
|
| 207 |
def resolve_mock(*args, **kwargs): |
|
| 208 |
return {"failed": [],
|
|
| 209 |
"accepted": args[0], |
|
| 210 |
"rejected": args[1], |
|
| 211 |
} |
|
| 212 |
astakos.return_value.resolve_commissions.side_effect = resolve_mock |
|
| 208 | 213 |
yield astakos.return_value |
| 209 | 214 |
|
| 210 | 215 |
|
Also available in: Unified diff