Skip to content

Instantly share code, notes, and snippets.

@esolitos
Last active March 6, 2019 07:34
Show Gist options
  • Save esolitos/1f5ff5be1de61fe4eaae354cf284867c to your computer and use it in GitHub Desktop.
Save esolitos/1f5ff5be1de61fe4eaae354cf284867c to your computer and use it in GitHub Desktop.
Upgrade all non Prod instances in Wodby for a given server UUID (Or _all_ if set to `None`)

Notes

Currently seems to work only on Python 2.7 due to swagger-codegen using the reserved keyword async.

It also needs a patch on six.py (info here). One must move the second parameter "unicode_escape" from the unicoide() call in six.py:L653 (see on github).

import time
from os import environ as env
from sys import stdout
from wodby import Configuration as WodbyConfiguration, InstanceApi, ApplicationApi, TaskApi, Instance, Task, \
InstanceType, ResponseTask
from wodby import ApiClient as WodbyApiClient
from wodby.rest import ApiException
def wodby_client():
if 'WODBY_API_KEY' not in env:
raise RuntimeError("Missing WODBY_API_KEY in environment.")
conf = WodbyConfiguration()
conf.api_key['X-API-KEY'] = env['WODBY_API_KEY']
return WodbyApiClient(configuration=conf)
def main():
client = wodby_client()
app_api = ApplicationApi(client)
instance_api = InstanceApi(client)
task_api = TaskApi(client)
server_uuid = 'your-server-uuid-here'
server_instances = instance_api.get_instances(server_id=server_uuid)
apps = {}
inst = None # type: Instance
for inst in server_instances:
if inst.type is None:
print("Missing UUID")
continue
if inst.type is InstanceType.PROD:
print(u"Prod instance, skipping: {}".format(inst.id))
continue
app_id = inst.app_id
if app_id not in apps:
apps[app_id] = app_api.get_app(app_id)
task = None
try:
task = instance_api.upgrade_instance(inst.id) # type: ResponseTask
print(u'Upgrading {} - {}'.format(apps[app_id].title, inst.title))
except ApiException as e:
if e.status == 400:
from json import loads
data = loads(e.body)
if data['error']['attributes']['errors'][0] == u'There is no new version available':
print(u'OK: {} - {}'.format(apps[app_id].title, inst.title))
# Check if an error occurred
if task is None:
continue
task = task.task # type: Task
task_id = task.id
interval = 10.0
waited = 0
while task.status != "done":
task = task_api.get_task(task_id) # type: Task
if task.status == "failed":
print("Failed upgrade task.")
break
elif task.status == "canceled":
print("Canceled: Upgrade task.")
elif waited > 300:
print("Expired 5 minutes TTL. Last task status was: {} (ID: {})".format(task.status, task.id))
else:
# Try to wait a few seconds and check again.
stdout.write('.')
stdout.flush()
time.sleep(interval)
waited += interval
# Slowly increase wait time by 20% on each wait.
interval += (interval/5)
stdout.write(': Finished in circa {} seconds!\n'.format(waited))
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment