#!/usr/bin/env python
# -----------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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 script starts a simulated DUCC cluster.  All processes are real, so all
# codepaths are executed.  The simulation comes from running multiple DUCC agents
# on a single node, allowing large-scale testing on small-scale clusters.  All 
# the system-test jobs sleep rather than compute and as such are very small and
# inexpensive, which allows large-scale testing to be performed on minimal 
# hardware.
#

import os
import sys
import time
import getopt

from threading import *
import Queue

#designed to run only here, relative to ducc_runtime
DUCC_HOME = os.path.abspath(__file__ + '/../../..')
sys.path.append(DUCC_HOME + '/admin')

from ducc_util  import DuccUtil
from properties import Properties
from ducc       import Ducc
from ducc_util  import ThreadPool

# multi-threaded start can overwhelm ssh if going to the same target host.  We inject
# a short sleep between starts to make it better.  This is how long to sleep in secs.
# Note: when 0.1 the ducc.properties merge step run by each agent caused problems/hangs
global SLEEP_TIME
SLEEP_TIME = 0.5

class StartSim(DuccUtil):

    def __init__(self):
        DuccUtil.__init__(self, True)

    def start_broker(self):
        
        if ( not self.automanage_broker ):
            print "Broker is not automanaged, returning."
            return

        broker_host = self.localhost
        print 'broker host', broker_host
        lines = self.ssh(broker_host, True, "'", self.DUCC_HOME + '/admin/ducc.py', '-c', 'broker', '--simtest', "'")

        for i in range(0, 9):
            if ( self.is_amq_active() ):
                return
            print 'Waiting for broker', str(i)
            time.sleep(1)

    def start_database(self):
        
        if ( not self.automanage_database ):
            print "Database is not automanaged, returning."
            return
		
        try:
        	self.db_start()                
        except Exception (e):
            # print e
            print sys.exc_info()[0], "Can't start the database."
            sys.exit(1)
		
    def run_local_agent(self, pnode, ip, memory ):

        if ( not self.verify_jvm() ):
            return        
        self.verify_duccling()
        self.verify_limits()

        memory = int(memory) * 1024 * 1024    # to GB from KB
        CMDPARMS = []
        CMDPARMS.append(self.java())
        CMDPARMS.append('-Dducc.deploy.components=agent')
        CMDPARMS.append('-Dos.page.size=' + self.os_pagesize)
        CMDPARMS.append('-Dducc.deploy.configuration=' + self.DUCC_HOME + "/resources/ducc.properties")
        CMDPARMS.append('-Djava.library.path=' + self.DUCC_HOME)
        CMDPARMS.append('-Xmx100M')
        CMDPARMS.append('-XX:+HeapDumpOnOutOfMemoryError')
        
        ducc_heap_dump_path = self.ducc_properties.get('ducc.heap.dump.path')
        if ducc_heap_dump_path != None:
            CMDPARMS.append('-XX:HeapDumpPath='+ducc_heap_dump_path)
            
        CMDPARMS.append('-Dducc.agent.node.metrics.fake.memory.size=' + str(memory))
        # Put virtual IP on command line for duplicate daemon detector
        CMDPARMS.append('-Dducc.agent.virtual=' + ip)
        CMDPARMS.append('org.apache.uima.ducc.common.main.DuccService')
        
        print "Start agent with pnode", pnode, "IP", ip, "memory", memory
        os.environ['DUCC_NODENAME'] = pnode
        os.environ['DUCC_IP'] = ip

        self.nohup(CMDPARMS)
        #self.spawn(' '.join(CMDPARMS))

    #
    # Start admin components rm pm sm ws db or, on local node using Ducc.py
    #        
    def startComponent(self, args):

        msgs = []
        com, or_parms = args

        if ( com in ('ws', 'viz') ):
            node = self.webserver_node
        else:
            node = self.localhost

        if ( com == 'or' ):
            lines = self.ssh(node, True, "'", 
                             self.DUCC_HOME + '/admin/ducc.py', '-c', 'or', '-b', 
                             '--or_parms', or_parms, "'")
        else:
            lines = self.ssh(node, True, "'", 
                             self.DUCC_HOME + '/admin/ducc.py', '-c', com, '-b', "'")

        msgs.append(('Start', com, 'on', node))
        while 1:
            line = lines.readline().strip()
            if ( not line ):
                break
            #print '[] ' + line
            if ( line.startswith('PID') ):
                toks = line.split(' ')    # get the PID
                self.pidlock.acquire();
                self.pids.put(com, self.localhost + ' ' + toks[1] + ' ' + self.localhost)
                self.pidlock.release();
                lines.close()
                msgs.append(('    PID', toks[1]))
                break

        return msgs

    #
    # Read the special nodelist and start "special" agents
    #
    # Nodelist has records compatible with java properties like this:
    #
    #   index nodename memory-in-gb
    #
    # We generate fake IP addresses from 192.168.4.X where x is the index.
    # We generate fake node names from nodename-X where x is the index.
    # We pass the memory-in-gb to the agent as the fake memory it reports instead of real memory.
    #
    # We use Jerry's memory override for each agent to get it to falsely report the memory
    # instead of reading from the real machine.
    #

    def startOneAgent(self, args):
        response = []
        node, cmd, mem, ip, pnode, index = args
        response.append(('Starting agent on', node, 'instance', index, 'as pseudo-node', pnode, 'IP', ip, 'memory', mem))
        lines = self.ssh(node, True, "'", cmd, '--agent', '--memory', mem, '--addr', ip, '--pseudoname', pnode, "'")
        while 1:
            line = lines.readline().strip()
            if ( not line ):
                break
            #print '[]', line
            if ( line.startswith('PID')):
                toks = line.split(' ')    # get the PID
                lines.close()
                response.append(('  ... Started,  PID', toks[1]))

                # Gottoa run on old python that doesn't know 'with'
                self.pidlock.acquire()
                self.pids.put(index, node + ' ' + toks[1] + ' ' + pnode)
                self.pidlock.release()

                break
        return response
  
    def startAgents(self, nodelist, instances):        
        do_all = True
        if ( len(instances) > 0 ):
            do_all = False
            print 'Starting instances:', instances

        print "Starting from nodes in", nodelist
        props = Properties()
        props.load(nodelist)

        # first parse the node config into a list of tuples describing the simulated nodes
        # each tuple is like this: (index, nodename, mem)
        # where 'index' is a unique number (must be a number)
        #       'nodename' is the name of a real node, which will contain one or more agents
        #                  to simulate multiple nodes
        #       'mem' is the simulated memory the agent will report
        nodes = props.get('nodes').split()
        mems  = props.get('memory').split()

        allnodes = []
        ndx = 1
        for node in nodes:
            # allow the index to be reset by-node so you can vary the number of
            # pseudo nodes on a real node without changing the indexes of others
            # if desired, e.g. to manage a custom ducc.nodes
            key = 'index.' + node
            if ( props.has_key(key) ):
                ndx = int(props.get(key))

            # generate node names and associate simulated memory
            for mem in mems:
                count = props.get(node + '.' + mem)
                if ( count != None ):
                    for i in range(0, int(count)):
                        allnodes.append( (str(ndx), node, mem) )
                        ndx = ndx + 1
                            

        here = os.getcwd()
        cmd = os.path.abspath(sys.argv[0])   
        for (index, node, mem) in allnodes:
            if ( not do_all ):
                if ( not instances.has_key(index) ):
                    continue

            ip = '192.168.4.' + index
            pnode = node + '-' + index
            self.threadpool.invoke(self.startOneAgent, node, cmd, mem, ip, pnode, index)
            time.sleep(SLEEP_TIME)

    def usage(self, msg):
        if (msg != None):
            if ( msg[0] != None ):
                msg = ' '.join(msg)
            print msg
              
        print "Usage:"
        print "   start_sim [options]"
        print "   If no options are given this help screen is shown."
        print ""
        print "Options:"
        print "   -n --nodelist nodelist"
        print "        The nodelist describing agents that is to be used. Not valid with --agent, --memory, --agent, or --pseudoname."
        print "        A nodelist provides the parameters for starting agents. Lines are of the form:"
        print "           index nodename mem-in-GB"
        print "        'index' is any unique number."
        print "        'nodename' is the physical node you want the agent placed on"
        print "        'mem-in-GB' is the amount of simulated memory the agent will report"
        print ""
        print "        Example:"
        print "          1  node290 31"
        print "          2  node290 31"
        print "          9  node291 31"
        print "          10 node291 31"
        print "          17 node292 47"
        print "          18 node292 47"
        print "          25 node293 47"
        print "          26 node293 47"
        print ""
        print "   -c --components component"
        print "        Start the indicated component, must be one of", ' '.join(self.default_components), "or 'all'"
        print ""
        print "   -i --instance instanceid"
        print "        Start only this instance of an agent from the nodelist."
        print ""
        print "   --agent"
        print "        Start an agent only on localhost. All of --memory, --addr, and --pseudoname are required, -n is disallowed."
        print ""
        print "   --memory mem"
        print "        Use this memory override.  Valid only with --agent."
        print ""
        print "   --addr"
        print "        Use this IP override. Valid only with --agent."
        print ""
        print "   --pseudoname pseudoname"
        print "        Use this as the hostname for the agent. Valid only with -a."
        print ""
        print "    --nothreading"
        print "        Disable multithreaded operation if it would otherwise be used"
        print ""
        print "   -v, --version"
        print "        Print the current DUCC version"
        sys.exit(1)


    def invalid(self, *msg):
        if ( msg != None ):
            if ( msg[0] != None ):
                msg = ' '.join(msg)
        print msg

        print "For usage run"
        print "    start_sim -h"
        print 'or'
        print '    start_sim --help'
        sys.exit(1)

    def main(self, argv):
        print "Running as", os.getpid()

        if ( len(argv) == 0 ):
            self.usage(None)
    
        os.environ['DUCC_SIM'] = 'True'
        
        node_config = None
        components = {}
        instances = {}
        run_agent = False
        memory = None
        pseudoname = None
        IP = None

        or_parms = self.ducc_properties.get('ducc.orchestrator.start.type')

        try:
            opts, args = getopt.getopt(argv, 'c:i:n:vh?', ['component=', 'help', 'agent', 'memory=', 
                                                          'instance=', 'addr=', 'pseudoname=', 'node-config=', 
                                                          'version', 'warm', 'cold',
                                                           'nothreading'])
        except:
            self.invalid('Invalid arguments', ' '.join(argv))
                  
        for ( o, a ) in opts:
            if o in ( '-n', '--node-config' ): 
                node_config = a
            elif o in ( '--agent' ):
                run_agent = True
            elif o in ( '-c', '--component' ):
                if ( a == 'all' ):
                    for cmp in self.default_components:
                        components[cmp] = cmp
                else:
                    components[a] = a
            elif o in ( '--memory' ):
                memory = a
            elif o in ( '--addr' ):
                IP = a
            elif o in ( '-i', '--instance' ):
                instances[a] = a
            elif o in ( '--pseudoname' ):
                pseudoname = a
            elif o in ( '--nothreading' ):
                self.disable_threading()
            elif o in ( '-v', '--version' ):
                print self.version()
                os.exit(0)
            elif o in ( '--warm', '--cold' ):
                or_parms = o[2:]
            elif o in ( '-h', '--help' ):
                self.usage(None)
            elif ( o == '-?'):
                self.usage(None)
            else:
                self.invalid('bad args: ', ' '.join(argv))

        self.pids = Properties()
        self.ducc = Ducc()

        if ( not self.verify_jvm() ):
            return  

        self.set_duccling_version()
        self.verify_duccling()

        if ( os.path.exists('sim.pids') ):
            self.pids.load('sim.pids')
        if ( run_agent ):
            #
            # checks that aren't valid if we want run_agent
            #
            if ( node_config != None ):
                self.invalid("Nodelist is not compatible with agent")
                
            if ( (IP == None ) or ( memory == None) or ( pseudoname == None )):
                    self.invalid("Missing IP, memory, or pseudoname")

            self. run_local_agent(pseudoname, IP, memory)
            sys.exit(0)
        else:         
            self.pidlock = Lock()
            self.threadpool = ThreadPool(50)

            if ( (IP != None) or (memory != None) or ( pseudoname != None )) :
                self.invalid("Running with a nodelist is not compatible with running a single agent.");

            try:
                specials = ['broker', 'db']
                sc = set(components)
                sb = set(specials)
                specials_only = False
                if ( sc.issubset(sb) ):
                    read_pids = True

                print '-------- start broker'
                if ( components.get('broker') != None ):
                    self.start_broker()
    
                print '-------- start database'

                if ( components.get('db') != None ):
                	self.start_database()
                        
                print '-------- specials_only', specials_only
                if ( specials_only ):
                    return

                print '-------- start agents'
                if ( node_config != None ):
                    self.startAgents(node_config, instances)
    
                for (com, com) in components.items():
                    if ( not com in specials ):    # specials start with different rules
                        self.threadpool.invoke(self.startComponent, com, or_parms)
                        time.sleep(SLEEP_TIME)

            except:
                pass

            self.threadpool.quit()
            self.pids.write('sim.pids')            

if __name__ == "__main__":
    starter = StartSim()
    starter.main(sys.argv[1:])