# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2014 IBM Corporation
# Copyright 2015-2017 Lenovo
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

# This is the main application.
# It should check for existing UDP socket to negotiate socket listen takeover
# It will have three paths into it:
#   -Unix domain socket
#   -TLS socket
#   -WSGI
# Additionally, it will be able to receive particular UDP packets to facilitate
# Things like heartbeating and discovery
# It also will optionally snoop SLP DA requests


#import logging
#logging.basicConfig(filename='/tmp/asyn.log', level=logging.DEBUG)
import atexit
import confluent.auth as auth
import confluent.config.conf as conf
import confluent.config.configmanager as configmanager
import confluent.debugger as debugger
try:
    import anydbm as dbm
except ModuleNotFoundError:
    import dbm
import confluent.consoleserver as consoleserver
import confluent.core as confluentcore
import confluent.httpapi as httpapi
import confluent.log as log
import confluent.collective.manager as collective
import linecache
try:
    import confluent.sockapi as sockapi
except ImportError:
    #On platforms without pwd, give up on the sockapi in general and be http
    #only for now
    pass
import confluent.discovery.core as disco
havefcntl = True
try:
    import fcntl
except ImportError:
    havefcntl = False
#import multiprocessing
import asyncio
import sys
import os
import glob
import signal
import socket
import subprocess
import time
import traceback
import tempfile
import uuid


def format_stack(task):
    task.print_stack()
    extracted_list = []
    checked = set()
    for f in task.get_stack():
        lineno = f.f_lineno
        co = f.f_code
        filename = co.co_filename
        name = co.co_name
        if filename not in checked:
            checked.add(filename)
            linecache.checkcache(filename)
        line = linecache.getline(filename, lineno, f.f_globals)
        extracted_list.append((filename, lineno, name, line))

    exc = task._exception
    if not extracted_list:
        yield f'No stack for {task!r}'
    elif exc is not None:
        yield f'Traceback for {task!r} (most recent call last):'
    else:
        yield f'Stack for {task!r} (most recent call last):'

    for x in traceback.format_list(extracted_list):
        yield x
    if exc is not None:
        for line in traceback.format_exception_only(exc.__class__, exc):
            yield line


def _daemonize():
    if 'fork' not in os.__dict__:
        return
    thispid = os.fork()
    if thispid > 0:
        os.waitpid(thispid, 0)
        os._exit(0)
    os.setsid()
    thispid = os.fork()
    if thispid > 0:
        print('confluent server starting as pid {0}'.format(thispid))
        os._exit(0)
    os.closerange(0, 2)
    os.open(os.devnull, os.O_RDWR)
    os.dup2(0, 1)
    os.dup2(0, 2)
    log.daemonized = True


def _redirectoutput():
    os.umask(63)
    configmanager.set_global('logdirectory', _get_logdirectory())
    sys.stdout = log.Logger('stdout', buffered=False)
    sys.stderr = log.Logger('stderr', buffered=False)


def _updatepidfile():
    pidfile = open('/var/run/confluent/pid', 'w+')
    fcntl.flock(pidfile, fcntl.LOCK_EX)
    pidfile.write(str(os.getpid()))
    fcntl.flock(pidfile, fcntl.LOCK_UN)
    pidfile.close()


def is_running():
    # Utility function for utilities to check if confluent is running
    try:
        pidfile = open('/var/run/confluent/pid', 'r+')
        fcntl.flock(pidfile, fcntl.LOCK_SH)
        pid = pidfile.read()
        if pid != '':
            try:
                os.kill(int(pid), 0)
                return pid
            except OSError:
                # There is no process running by that pid, must be stale
                pass
        fcntl.flock(pidfile, fcntl.LOCK_UN)
        pidfile.close()
    except IOError:
        pass
    return None


def _checkpidfile():
    try:
        pidfile = open('/var/run/confluent/pid', 'r+')
        fcntl.flock(pidfile, fcntl.LOCK_EX)
        pid = pidfile.read()
        if pid != '':
            try:
                os.kill(int(pid), 0)
                print ('/var/run/confluent/pid exists and indicates %s is still '
                       'running' % pid)
                sys.exit(1)
            except OSError:
                # There is no process running by that pid, must be stale
                pass
        pidfile.seek(0)
        pidfile.write(str(os.getpid()))
        fcntl.flock(pidfile, fcntl.LOCK_UN)
        pidfile.close()
    except IOError:
        try:
            pidfile = open('/var/run/confluent/pid', 'w')
        except IOError as e:
            if e.errno != 2:
                raise
            os.makedirs('/var/run/confluent')
            pidfile = open('/var/run/confluent/pid', 'w')
        fcntl.flock(pidfile, fcntl.LOCK_EX)
        pidfile.write(str(os.getpid()))
        fcntl.flock(pidfile, fcntl.LOCK_UN)
        pidfile.close()
        pidfile = open('/var/run/confluent/pid', 'r')
        fcntl.flock(pidfile, fcntl.LOCK_SH)
        pid = pidfile.read()
        if pid != str(os.getpid()):
            print ('/var/run/confluent/pid exists and indicates %s is still '
                   'running' % pid)
            sys.exit(1)
        fcntl.flock(pidfile, fcntl.LOCK_UN)
        pidfile.close()


_stopevent = None


def request_stop():
    """Ask the service to shut down.

    Called from the event loop, either by a signal handler or by a client
    asking for a shutdown, so it only has to wake the main coroutine.
    """
    if _stopevent is not None:
        _stopevent.set()


def _finish_shutdown():
    """Say the service is going and get the configuration on disk."""
    log.log({'info': 'Confluent management service shutting down'}, flush=True)
    configmanager.ConfigManager.wait_for_sync()


def terminate(signalname, frame):
    # Only for platforms where the event loop cannot deliver signals for us.
    # Raising SystemExit from a handler while the loop runs escapes run_forever
    # and leaves asyncio unable to close the loop, which is why the loop is
    # asked to stop instead wherever that is possible.
    sys.exit(0)

def dumptrace(signalname, frame):
    ht = open('/var/log/confluent/hangtraces', 'a')
    ht.write('Dumping active trace on ' + time.strftime('%X %x\n'))
    ht.write(''.join(traceback.format_stack(frame)))
    for atask in asyncio.all_tasks():
        ht.write('Async trace: ({0})\n'.format(id(atask)))
        ht.write(''.join([x for x in format_stack(atask)]))
    ht.close()

def doexit():
    if not havefcntl:
        return
    try:
        os.remove('/var/run/confluent/dbg.sock')
    except OSError:
        pass
    try:
        with open('/var/run/confluent/pid') as pidfile:
            pid = pidfile.read()
        if pid == str(os.getpid()):
            os.remove('/var/run/confluent/pid')
    except OSError:
        # Someone else already tidied it up, which is not worth a traceback out
        # of an atexit callback
        pass


def _initsecurity(config):
    if config.has_option('security', 'externalcfgkey'):
        keyfile = config.get('security', 'externalcfgkey')
        with open(keyfile, 'r') as keyhandle:
            key = keyhandle.read()
        configmanager.init_masterkey(key)
        # We don't want to os._exit() until sync finishes from
        # init above
        configmanager.ConfigManager.wait_for_sync()


def setlimits():
    try:
        import resource
        currlimit = resource.getrlimit(resource.RLIMIT_NOFILE)
        if currlimit[0] < currlimit[1]:
            resource.setrlimit(
                resource.RLIMIT_NOFILE, (currlimit[1], currlimit[1]))
    except Exception:
        pass

def assure_ownership(path):
    try:
        if os.getuid() != os.stat(path).st_uid:
            if os.getuid() == 0:
                sys.stderr.write('Attempting to run as root, when non-root usage is detected\n')
            else:
                sys.stderr.write('{} is not owned by confluent user, change ownership\n'.format(path))
            sys.exit(1)
    except OSError as e:
        if e.errno == 13:
            if os.getuid() == 0:
                sys.stderr.write('Attempting to run as root, when non-root usage is detected\n')
            else:
                sys.stderr.write('{} is not owned by confluent user, change ownership\n'.format(path))
            sys.exit(1)

def sanity_check():
    assure_ownership('/etc/confluent')
    assure_ownership('/etc/confluent/cfg')
    for filename in glob.glob('/etc/confluent/cfg/*'):
        assure_ownership(filename)
    assure_ownership('/etc/confluent/privkey.pem')
    assure_ownership('/etc/confluent/srvcert.pem')


def migrate_db():
    tdir = tempfile.mkdtemp()
    subprocess.check_call(['python3', '-c', 'pass'])
    subprocess.check_call(['python2', '/opt/confluent/bin/confluentdbutil', 'dump', '-u', tdir])
    subprocess.check_call(['python3', '/opt/confluent/bin/confluentdbutil', 'restore', '-u', tdir])
    subprocess.check_call(['rm', '-rf', tdir])
    configmanager.init()


def run(args):
    asyncio.run(asyncrun(args))

async def asyncrun(args):
    setlimits()
    try:
        configmanager.ConfigManager(None)
    except dbm.error:
        migrate_db()
    try:
        signal.signal(signal.SIGUSR1, dumptrace)
    except AttributeError:
        pass   # silly windows
    if havefcntl:
        _checkpidfile()
    conf.init_config()
    try:
        config = conf.get_config()
        _initsecurity(config)
    except:
        sys.stderr.write("Error unlocking credential store\n")
        doexit()
        sys.exit(1)
    sanity_check()
    try:
        confluentcore.load_plugins()
    except:
        doexit()
        raise
    try:
        log.log({'info': 'Confluent management service starting'}, flush=True)
    except (OSError, IOError) as e:
        print(repr(e))
        sys.exit(1)
    if '-f' not in args:
        sys.stderr.write("-f is now required")
        # the fork wreaks havoc with asyncio thread executor
        # If someone comes along with a non-systemd demand, will just have to have a daemonize wrapper
        sys.exit(1)
        #_daemonize()
    if '-o' not in args:
        _redirectoutput()
    if havefcntl:
        _updatepidfile()
    asyncio.get_running_loop().set_debug(True)
    global _stopevent
    _stopevent = asyncio.Event()
    for stopsig in (signal.SIGINT, signal.SIGTERM):
        try:
            asyncio.get_running_loop().add_signal_handler(stopsig, request_stop)
        except NotImplementedError:  # silly windows
            signal.signal(stopsig, terminate)
    configmanager.set_shutdown_hook(request_stop)
    atexit.register(doexit)
    confluentuuid = configmanager.get_global('confluent_uuid')
    if not confluentuuid:
        confluentuuid = str(uuid.uuid4())
        configmanager.set_global('confluent_uuid', confluentuuid)
    if not configmanager._masterkey:
        configmanager.init_masterkey()
    debugger.start_dbgif()
    auth.check_for_yaml()
    collective.startup()
    await consoleserver.initialize()
    http_bind_host, http_bind_port, http_bind_group, http_bind_perms = _get_connector_config('http')
    sock_bind_host, sock_bind_port, sock_bind_group, sock_bind_perms = _get_connector_config('socket')
    try:
        sockservice = sockapi.SockApi(sock_bind_host, sock_bind_port, sock_bind_group, sock_bind_perms)
        asyncio.get_running_loop().create_task(sockservice.start())
    except NameError:
        pass
    webservice = httpapi.HttpApi(http_bind_host, http_bind_port, http_bind_group, http_bind_perms)
    webservice.start()
    notifysock = os.environ.get('NOTIFY_SOCKET', None)
    watchdogsecs = 200
    if notifysock:
        watchdogsecs = int(os.environ.get('WATCHDOG_USEC', '0')) / 1000000
        if not watchdogsecs:
            watchdogsecs = 200
        watchdogsecs = watchdogsecs / 2
        if notifysock.startswith('@'):
            notifysock = '\0' + notifysock[1:]
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        sock.connect(notifysock)
        sock.send(b'READY=1')
        firstdeadline = time.time() + watchdogsecs
    while not _stopevent.is_set() and len(
            list(configmanager.list_collective())) >= 2:
        # If in a collective, stall automatic startup activity
        # until we establish quorum
        try:
            configmanager.check_quorum()
            break
        except Exception:
            # Waiting on the stop event rather than sleeping through it.  A
            # member that cannot reach quorum stalls here for as long as that
            # lasts, and asking it to stop has to be answered in that state
            # rather than after quorum returns.
            try:
                await asyncio.wait_for(_stopevent.wait(), 0.5)
            except asyncio.TimeoutError:
                if notifysock and time.time() >= firstdeadline:
                    sock.send(b'WATCHDOG=1')
                    firstdeadline = time.time() + watchdogsecs
    if _stopevent.is_set():
        # Asked to stop before startup got as far as serving anything, so
        # there is nothing to unwind beyond what stopping normally does
        return _finish_shutdown()
    disco.start_detection()
    await asyncio.sleep(1)
    await consoleserver.start_console_sessions()

    while not _stopevent.is_set():
        try:
            await asyncio.wait_for(_stopevent.wait(), watchdogsecs)
        except asyncio.TimeoutError:
            pass
        else:
            break
        if notifysock:
            sock.send(b'WATCHDOG=1')
    _finish_shutdown()

def _get_connector_config(session):
    host = conf.get_option(session, 'bindhost')
    port = conf.get_int_option(session, 'bindport')
    group = conf.get_option(session, 'bindgroup')
    perms = conf.get_option(session, 'bindperms')
    if perms:
        if perms.startswith('0'):
            perms = int(perms, 8)
        else:
            # Parse rw-rw-rw- format (user, group, other)
            perms_value = 0
            perm_map = {'r': 4, 'w': 2, 'x': 1}
            for i, section in enumerate([perms[0:3], perms[3:6], perms[6:9]]):
                for char in section:
                    if char in perm_map:
                        perms_value += perm_map[char] * (8 ** (2 - i))
            perms = perms_value
    else:
        perms = None
    return (host, port, group, perms)

def _get_logdirectory():
    return conf.get_option('globals', 'logdirectory')