<gh_stars>1-10
from js9 import j


def get_container(service, password):
    from zeroos.orchestrator.sal.Container import Container
    return Container.from_ays(service.parent, password, logger=service.logger)


def is_port_listening(container, port, timeout=30, listen=True):
    import time
    start = time.time()
    while start + timeout > time.time():
        if port not in container.node.freeports(port, nrports=3):
            return True
        if not listen:
            return False
        time.sleep(0.2)
    return False


def is_job_running(container, cmd='/bin/tlogserver'):
    try:
        for job in container.client.job.list():
            arguments = job['cmd']['arguments']
            if 'name' in arguments and arguments['name'] == cmd:
                return job
        return False
    except Exception as err:
        if str(err).find("invalid container id"):
            return False
        raise


def save_config(job, vdisks=None):
    import yaml
    from zeroos.orchestrator.sal.ETCD import EtcdCluster
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)

    service = job.service
    config = {"servers": [service.model.data.bind]}
    yamlconfig = yaml.safe_dump(config, default_flow_style=False)

    etcd_cluster = service.aysrepo.servicesFind(role='etcd_cluster')[0]
    etcd = EtcdCluster.from_ays(etcd_cluster, job.context['token'])

    etcd.put(key="%s:cluster:conf:tlog" % service.name, value=yamlconfig)

    for vdisk in vdisks:
        vdiskstore = vdisk.parent
        config = {
            "storageClusterID": vdiskstore.model.data.blockCluster,
            "tlogServerClusterID": service.name,
            "slaveStorageClusterID": vdiskstore.model.data.slaveCluster or "",
        }
        job.logger.debug("tlogserver %s save config for vdisk %s", service, vdisk)
        job.logger.debug(config)
        yamlconfig = yaml.safe_dump(config, default_flow_style=False)
        etcd.put(key="%s:vdisk:conf:storage:nbd" % vdisk.name, value=yamlconfig)


def install(job):
    from zeroos.orchestrator.sal.ETCD import EtcdCluster
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)

    service = job.service

    etcd_cluster = service.aysrepo.servicesFind(role='etcd_cluster')[0]
    etcd_cluster = EtcdCluster.from_ays(etcd_cluster, job.context['token'])

    vm = service.consumers['vm'][0]
    vdisks = vm.producers.get('vdisk', [])
    container = get_container(service, job.context['token'])
    config = {
        'storageClusters': set(),
        'data-shards': 0,
        'parity-shards': 0,
    }

    backup = False
    for vdiskservice in vdisks:
        vdiskstore = vdiskservice.parent
        objectcluster = vdiskstore.model.data.objectCluster
        if objectcluster and objectcluster not in config['storageClusters']:
            data_shards, parity_shards = get_storagecluster_config(job, objectcluster)
            config['storageClusters'].add(objectcluster)
            config['data-shards'] += data_shards
            config['parity-shards'] += parity_shards
            if vdiskstore.model.data.slaveCluster:
                backup = True

    if not config['storageClusters']:
        return

    save_config(job, vdisks)
    data_shards = config.pop('data-shards')
    parity_shards = config.pop('parity-shards')

    # check if we consume another tlog on which we need to sync at startup
    tlogWaitAddr = None
    if 'tlogserver' in service.producers:
        waitTlogServer_service = service.producers['tlogserver'][0]
        tlogWaitAddr = waitTlogServer_service.model.data.waitListenBind

    bind = service.model.data.bind
    waitListenBind = service.model.data.waitListenBind

    if not is_port_listening(container, int(bind.split(':')[1]), listen=False):
        cmd = '/bin/tlogserver \
                -id {id} \
                -flush-size 128 \
                -address {bind} \
                -wait-listen-addr {waitListenBind} \
                -config "{dialstrings}" \
                '.format(id=vm.name,
                         bind=bind,
                         waitListenBind=waitListenBind,
                         dialstrings=etcd_cluster.dialstrings)
        if backup:
            cmd += ' -with-slave-sync'
        if tlogWaitAddr:
            cmd += ' -wait-connect-addr {}'.format(tlogWaitAddr)
        if service.model.data.acceptAddress:
            cmd += ' -accept-address {}'.format(service.model.data.acceptAddress)

        job.logger.info("Starting tlog server: %s" % cmd)
        container.client.system(cmd, id="{}.{}".format(service.model.role, service.name))
        if not is_port_listening(container, int(bind.split(":")[1])):
            raise j.exceptions.RuntimeError('Failed to start tlogserver {}'.format(service.name))
    service.model.data.status = 'running'
    service.saveAll()

    tcpsrv = service.producers['tcp'][0]
    if tcpsrv.model.data.status == "dropped":
        tcpsrv.executeAction('install', context=job.context)


def start(job):
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)
    service = job.service
    service.executeAction('install', context=job.context)


def get_storagecluster_config(job, storagecluster):
    objectcluster_service = job.service.aysrepo.serviceGet(role='storagecluster.object',
                                                           instance=storagecluster)
    return objectcluster_service.model.data.dataShards, objectcluster_service.model.data.dataShards


def stop(job):
    import time
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)

    service = job.service
    if service.model.data.status != 'running':
        return

    service.model.data.status = 'halting'
    service.saveAll()
    container = get_container(service, job.context['token'])
    bind = service.model.data.bind
    port = int(bind.split(':')[1])
    tlogjob = is_job_running(container)
    if tlogjob:
        job.logger.info("killing job {}".format(tlogjob['cmd']['arguments']['name']))
        container.client.job.kill(tlogjob['cmd']['id'])

        job.logger.info("wait for tlogserver to stop")
        for i in range(60):
            time.sleep(1)
            if not is_port_listening(container, port):
                break
            raise j.exceptions.RuntimeError("Failed to stop Tlog server")

    # after stop, in case this service was consume by another tlog server for synchronisation
    # need to clean the consumer relation cause the sync is done just before stop.
    # the relation doesn't need to exists anymore.
    for consumer in service.consumers.get('tlogserver', []):
        service.model.consumerRemove(consumer)

    service.model.data.status = 'halted'
    service.saveAll()

    tcpsrv = service.producers['tcp'][0]
    if tcpsrv.model.data.status == "opened":
        tcpsrv.executeAction('drop', context=job.context)


def monitor(job):
    from zeroos.orchestrator.configuration import get_jwt_token

    service = job.service
    if not service.model.actionsState['install'] == 'ok':
        return

    if str(service.model.data.status) != 'running':
        return

    bind = service.model.data.bind
    port = int(bind.split(':')[1])
    container = get_container(service, get_jwt_token(job.service.aysrepo))
    if is_port_listening(container, port):
        return

    service.executeAction('start', context={"token": get_jwt_token(job.service.aysrepo)})


def watchdog_handler(job):
    import asyncio
    loop = j.atyourservice.server.loop
    service = job.service
    if str(service.model.data.status) != 'running':
        return
    eof = job.model.args['eof']
    if eof:
        asyncio.ensure_future(service.asyncExecuteAction('start', context=job.context), loop=loop)

from js9 import j


def get_container(service, password):
    from zeroos.orchestrator.sal.Container import Container
    return Container.from_ays(service.parent, password, logger=service.logger)


def is_port_listening(container, port, timeout=30, listen=True):
    import time
    start = time.time()
    while start + timeout > time.time():
        if port not in container.node.freeports(port, nrports=3):
            return True
        if not listen:
            return False
        time.sleep(0.2)
    return False


def is_job_running(container, cmd='/bin/tlogserver'):
    try:
        for job in container.client.job.list():
            arguments = job['cmd']['arguments']
            if 'name' in arguments and arguments['name'] == cmd:
                return job
        return False
    except Exception as err:
        if str(err).find("invalid container id"):
            return False
        raise


def save_config(job, vdisks=None):
    import yaml
    from zeroos.orchestrator.sal.ETCD import EtcdCluster
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)

    service = job.service
    config = {"servers": [service.model.data.bind]}
    yamlconfig = yaml.safe_dump(config, default_flow_style=False)

    etcd_cluster = service.aysrepo.servicesFind(role='etcd_cluster')[0]
    etcd = EtcdCluster.from_ays(etcd_cluster, job.context['token'])

    etcd.put(key="%s:cluster:conf:tlog" % service.name, value=yamlconfig)

    for vdisk in vdisks:
        vdiskstore = vdisk.parent
        config = {
            "storageClusterID": vdiskstore.model.data.blockCluster,
            "tlogServerClusterID": service.name,
            "slaveStorageClusterID": vdiskstore.model.data.slaveCluster or "",
        }
        job.logger.debug("tlogserver %s save config for vdisk %s", service, vdisk)
        job.logger.debug(config)
        yamlconfig = yaml.safe_dump(config, default_flow_style=False)
        etcd.put(key="%s:vdisk:conf:storage:nbd" % vdisk.name, value=yamlconfig)


def install(job):
    from zeroos.orchestrator.sal.ETCD import EtcdCluster
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)

    service = job.service

    etcd_cluster = service.aysrepo.servicesFind(role='etcd_cluster')[0]
    etcd_cluster = EtcdCluster.from_ays(etcd_cluster, job.context['token'])

    vm = service.consumers['vm'][0]
    vdisks = vm.producers.get('vdisk', [])
    container = get_container(service, job.context['token'])
    config = {
        'storageClusters': set(),
        'data-shards': 0,
        'parity-shards': 0,
    }

    backup = False
    for vdiskservice in vdisks:
        vdiskstore = vdiskservice.parent
        objectcluster = vdiskstore.model.data.objectCluster
        if objectcluster and objectcluster not in config['storageClusters']:
            data_shards, parity_shards = get_storagecluster_config(job, objectcluster)
            config['storageClusters'].add(objectcluster)
            config['data-shards'] += data_shards
            config['parity-shards'] += parity_shards
            if vdiskstore.model.data.slaveCluster:
                backup = True

    if not config['storageClusters']:
        return

    save_config(job, vdisks)
    data_shards = config.pop('data-shards')
    parity_shards = config.pop('parity-shards')

    # check if we consume another tlog on which we need to sync at startup
    tlogWaitAddr = None
    if 'tlogserver' in service.producers:
        waitTlogServer_service = service.producers['tlogserver'][0]
        tlogWaitAddr = waitTlogServer_service.model.data.waitListenBind

    bind = service.model.data.bind
    waitListenBind = service.model.data.waitListenBind

    if not is_port_listening(container, int(bind.split(':')[1]), listen=False):
        cmd = '/bin/tlogserver \
                -id {id} \
                -flush-size 128 \
                -address {bind} \
                -wait-listen-addr {waitListenBind} \
                -config "{dialstrings}" \
                '.format(id=vm.name,
                         bind=bind,
                         waitListenBind=waitListenBind,
                         dialstrings=etcd_cluster.dialstrings)
        if backup:
            cmd += ' -with-slave-sync'
        if tlogWaitAddr:
            cmd += ' -wait-connect-addr {}'.format(tlogWaitAddr)
        if service.model.data.acceptAddress:
            cmd += ' -accept-address {}'.format(service.model.data.acceptAddress)

        job.logger.info("Starting tlog server: %s" % cmd)
        container.client.system(cmd, id="{}.{}".format(service.model.role, service.name))
        if not is_port_listening(container, int(bind.split(":")[1])):
            raise j.exceptions.RuntimeError('Failed to start tlogserver {}'.format(service.name))
    service.model.data.status = 'running'
    service.saveAll()

    tcpsrv = service.producers['tcp'][0]
    if tcpsrv.model.data.status == "dropped":
        tcpsrv.executeAction('install', context=job.context)


def start(job):
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)
    service = job.service
    service.executeAction('install', context=job.context)


def get_storagecluster_config(job, storagecluster):
    objectcluster_service = job.service.aysrepo.serviceGet(role='storagecluster.object',
                                                           instance=storagecluster)
    return objectcluster_service.model.data.dataShards, objectcluster_service.model.data.dataShards


def stop(job):
    import time
    from zeroos.orchestrator.configuration import get_jwt_token

    job.context['token'] = get_jwt_token(job.service.aysrepo)

    service = job.service
    if service.model.data.status != 'running':
        return

    service.model.data.status = 'halting'
    service.saveAll()
    container = get_container(service, job.context['token'])
    bind = service.model.data.bind
    port = int(bind.split(':')[1])
    tlogjob = is_job_running(container)
    if tlogjob:
        job.logger.info("killing job {}".format(tlogjob['cmd']['arguments']['name']))
        container.client.job.kill(tlogjob['cmd']['id'])

        job.logger.info("wait for tlogserver to stop")
        for i in range(60):
            time.sleep(1)
            if not is_port_listening(container, port):
                break
            raise j.exceptions.RuntimeError("Failed to stop Tlog server")

    # after stop, in case this service was consume by another tlog server for synchronisation
    # need to clean the consumer relation cause the sync is done just before stop.
    # the relation doesn't need to exists anymore.
    for consumer in service.consumers.get('tlogserver', []):
        service.model.consumerRemove(consumer)

    service.model.data.status = 'halted'
    service.saveAll()

    tcpsrv = service.producers['tcp'][0]
    if tcpsrv.model.data.status == "opened":
        tcpsrv.executeAction('drop', context=job.context)


def monitor(job):
    from zeroos.orchestrator.configuration import get_jwt_token

    service = job.service
    if not service.model.actionsState['install'] == 'ok':
        return

    if str(service.model.data.status) != 'running':
        return

    bind = service.model.data.bind
    port = int(bind.split(':')[1])
    container = get_container(service, get_jwt_token(job.service.aysrepo))
    if is_port_listening(container, port):
        return

    service.executeAction('start', context={"token": get_jwt_token(job.service.aysrepo)})


def watchdog_handler(job):
    import asyncio
    loop = j.atyourservice.server.loop
    service = job.service
    if str(service.model.data.status) != 'running':
        return
    eof = job.model.args['eof']
    if eof:
        asyncio.ensure_future(service.asyncExecuteAction('start', context=job.context), loop=loop)
# check if we consume another tlog on which we need to sync at startup # after stop, in case this service was consume by another tlog server for synchronisation # need to clean the consumer relation cause the sync is done just before stop. # the relation doesn't need to exists anymore.