<commit_before>"""
Copyright (c) 2010-2015, Contrail consortium.
All rights reserved.

Redistribution and use in source and binary forms,
with or without modification, are permitted provided
that the following conditions are met:

 1. Redistributions of source code must retain the
    above copyright notice, this list of conditions
    and the following disclaimer.
 2. Redistributions in binary form must reproduce
    the above copyright notice, this list of
    conditions and the following disclaimer in the
    documentation and/or other materials provided
    with the distribution.
 3. Neither the name of the Contrail consortium nor the
    names of its contributors may be used to endorse
    or promote products derived from this software
    without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
"""

from threading import Thread
# import memcache
from shutil import rmtree
import pickle
import zipfile
import tarfile
import tempfile
import stat
import os.path
import time

from conpaas.core.expose import expose
from conpaas.core.manager import BaseManager, ManagerException

from conpaas.core import git
from conpaas.core.https.server import HttpJsonResponse, HttpErrorResponse,\
    HttpFileDownloadResponse, FileUploadField
from conpaas.services.generic.misc import archive_open, archive_get_members,\
    archive_close, archive_get_type, archive_extract_file

from conpaas.core.log import create_logger
from conpaas.services.generic.agent import client
from conpaas.services.generic.manager.config import CodeVersion,\
    ServiceConfiguration, VolumeInfo

class GenericManager(BaseManager):
    """Manager class with the following exposed methods:

    startup() -- POST
    shutdown() -- POST
    add_nodes(count) -- POST
    remove_nodes(count) -- POST
    list_nodes() -- GET
    get_service_info() -- GET
    get_node_info(serviceNodeId) -- GET
    list_code_versions() -- GET
    list_authorized_keys() -- GET
    upload_authorized_key(key) -- UPLOAD
    git_push_hook() -- POST
    upload_code_version(code, description) -- UPLOAD
    download_code_version(codeVersionId) -- GET
    enable_code(codeVersionId) -- POST
    delete_code_version(codeVersionId) -- POST
    list_volumes() -- GET
    generic_create_volume(volumeName, volumeSize, agentId) -- POST
    generic_delete_volume(volumeName) -- POST
    execute_script(command, parameters) -- POST
    get_script_status() -- GET
    get_agent_log(filename) -- GET
    """

    # String template for error messages returned when performing actions in
    # the wrong state
    WRONG_STATE_MSG = "ERROR: cannot perform %(action)s in state %(curstate)s"

    # String template for error messages returned when a required argument is
    # missing
    REQUIRED_ARG_MSG = "ERROR: %(arg)s is a required argument"

    # String template for debugging messages logged on nodes creation
    ACTION_REQUESTING_NODES = "requesting %(count)s nodes in %(action)s"

    # String used as an error message when 'interrupt' is called when no
    # scripts are currently running
    NO_SCRIPTS_ARE_RUNNING_MSG = "ERROR: No scripts are currently running inside "\
                                "agents. Nothing to interrupt."

    # String used as an error message when scripts are running inside agents
    SCRIPTS_ARE_RUNNING_MSG = "ERROR: Scripts are still running inside at "\
                                "least one agent. Please wait for them to "\
                                "finish execution or call 'interrupt' first."



    def __init__(self, config_parser, **kwargs):
        """Initialize a Generic Manager.

        'config_parser' represents the manager config file.
        **kwargs holds anything that can't be sent in config_parser."""
        BaseManager.__init__(self, config_parser)

        self.code_repo = config_parser.get('manager', 'CODE_REPO')

        if kwargs['reset_config']:
            self._create_initial_configuration()

        # self.nodes = []
        self.agents_info = []
        self.master_ip = None

    def _prepare_default_config_script(self, script_name):
        fileno, path = tempfile.mkstemp()
        fd = os.fdopen(fileno, 'w')
        fd.write('''#!/bin/bash
date >> /root/generic.out
echo "Executing script ${0##*/}" >> /root/generic.out
echo "Parameters ($#): $@" >> /root/generic.out
echo "My IP is $MY_IP" >> /root/generic.out
echo "My role is $MY_ROLE" >> /root/generic.out
echo "My master IP is $MASTER_IP" >> /root/generic.out
echo "Information about other agents is stored at /var/cache/cpsagent/agents.json" >> /root/generic.out
cat /var/cache/cpsagent/agents.json >> /root/generic.out
echo "" >> /root/generic.out
echo "" >> /root/generic.out
''')
        fd.close()
        os.chmod(path, stat.S_IRWXU | stat.S_IROTH | stat.S_IXOTH)
        return path

    def get_service_type(self):
        return 'generic'

    def _create_initial_configuration(self):
        self.logger.info("Creating initial configuration")

        config = ServiceConfiguration()

        if len(config.codeVersions) > 0:
            return

        if not os.path.exists(self.code_repo):
            os.makedirs(self.code_repo)

        tfile = tarfile.TarFile(name=os.path.join(self.code_repo, 'code-default'),
                mode='w')

        scripts = ['init.sh', 'notify.sh', 'run.sh', 'interrupt.sh', 'cleanup.sh']
        for script in scripts:
            path = self._prepare_default_config_script(script)
            tfile.add(path, script)
            os.remove(path)

        tfile.close()
        config.codeVersions['code-default'] = CodeVersion('code-default',
                'code-default.tar', 'tar', description='Initial version')
        config.currentCodeVersion = 'code-default'
        self._configuration_set(config)

   

    def on_start(self, nodes):
        """Start up the service. The first node will be the master node."""

        nr_instances = 1


        vals = { 'action': '_do_startup', 'count': nr_instances }
        self.logger.debug(self.ACTION_REQUESTING_NODES % vals)
        try:            
            config = self._configuration_get()

            roles = {'master':'1'}

            agents_info = self._update_agents_info(nodes, roles)

            self._init_agents(config, nodes, agents_info)
            self._update_code(config, nodes)

            # Extend the nodes list with the newly created one
            
            self.agents_info += agents_info
            self.master_ip = nodes[0].ip
            return True
        except Exception, err:
            self.logger.exception('_do_startup: Failed to create agents: %s' % err)
            return False
            

    def _update_agents_info(self, nodes, roles):
        id_ip = []
        for node in nodes:
            id_ip.append( { 'id': node.id, 'ip': node.ip })

        id_ip_role = []
        for role in roles:
            for _ in range(int(roles[role])):
                if len(id_ip):
                    node_ip_id = id_ip.pop(0)
                    node_ip_id.update({'role':role})
                    id_ip_role.append(node_ip_id)

        return id_ip_role

    def _init_agents(self, config, nodes, agents_info):
        self.logger.info("Initializing agents %s" %
                [ node.id for node in nodes ])

        for serviceNode in nodes:
            try:
                client.init_agent(serviceNode.ip, self.AGENT_PORT, agents_info)
            except client.AgentException:
                self.logger.exception('Failed initialize agent at node %s'
                        % str(serviceNode))
                self.state_set(self.S_ERROR, msg='Failed to initialize agent at node %s'
                        % str(serviceNode))
                raise
    
    def on_stop(self):
        """Delete all nodes and switch to status STOPPED"""
        

        self.logger.info("Removing nodes: %s" %
                [ node.id for node in self.nodes ])
        # self.controller.delete_nodes(self.nodes)
        del_nodes = self.nodes[:]
        self.agents_info = []
        self.master_ip = None
        return del_nodes

    def __check_count_in_args(self, kwargs):
        """Return 'count' if all is good. HttpErrorResponse otherwise."""
        count = None

        if 'count' in kwargs:
            count = kwargs.pop('count')
        # The frontend sends count under 'node'.
        elif 'node' in kwargs:
            count = kwargs.pop('node')
        else:
            return HttpErrorResponse(self.REQUIRED_ARG_MSG % { 'arg': 'count' })

        if not isinstance(count, int):
            return HttpErrorResponse(
                "ERROR: Expected an integer value for 'count'")

        return count

    def on_add_nodes(self, node_instances):
        # (genc): i have to figure out how to deal with the roles
        start_role = 'node'
        roles = {start_role: str(len(node_instances))}
        # Startup agents
        agents_info = self._update_agents_info(node_instances, roles)

        nodes_before = filter(lambda n: n not in node_instances, self.nodes)
        self.agents_info += agents_info

        config = self._configuration_get()
        self._init_agents(config, node_instances, self.agents_info)
        self._update_code(config, node_instances)

        self._do_execute_script('notify', nodes_before)
        return True


    def on_remove_nodes(self, node_roles):
        count = sum(node_roles.values())
        del_nodes = []
        cp_nodes = self.nodes[:]
        for _ in range(0, count):
            node = cp_nodes.pop()
            del_nodes += [ node ]
            self.agents_info.pop()
            self.logger.info("Removing node with IP %s" % node.ip)
        if not cp_nodes:
            self.master_ip = None
            self.state_set(self.S_STOPPED)
        else:
            self._do_execute_script('notify', self.cp_nodes)
            self.state_set(self.S_RUNNING)

        return del_nodes

    def __is_master(self, node):
        """Return True if the given node is the Generic master"""
        return node.ip == self.master_ip

    @expose('GET')
    def list_nodes(self, kwargs):
        """Return a list of running nodes"""
        state = self.state_get()
        if state != self.S_RUNNING:<commit_msg>Generic Service: fixed bug when removing a node which is not the last<commit_after>"""
Copyright (c) 2010-2015, Contrail consortium.
All rights reserved.

Redistribution and use in source and binary forms,
with or without modification, are permitted provided
that the following conditions are met:

 1. Redistributions of source code must retain the
    above copyright notice, this list of conditions
    and the following disclaimer.
 2. Redistributions in binary form must reproduce
    the above copyright notice, this list of
    conditions and the following disclaimer in the
    documentation and/or other materials provided
    with the distribution.
 3. Neither the name of the Contrail consortium nor the
    names of its contributors may be used to endorse
    or promote products derived from this software
    without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
"""

from threading import Thread
# import memcache
from shutil import rmtree
import pickle
import zipfile
import tarfile
import tempfile
import stat
import os.path
import time

from conpaas.core.expose import expose
from conpaas.core.manager import BaseManager, ManagerException

from conpaas.core import git
from conpaas.core.https.server import HttpJsonResponse, HttpErrorResponse,\
    HttpFileDownloadResponse, FileUploadField
from conpaas.services.generic.misc import archive_open, archive_get_members,\
    archive_close, archive_get_type, archive_extract_file

from conpaas.core.log import create_logger
from conpaas.services.generic.agent import client
from conpaas.services.generic.manager.config import CodeVersion,\
    ServiceConfiguration, VolumeInfo

class GenericManager(BaseManager):
    """Manager class with the following exposed methods:

    startup() -- POST
    shutdown() -- POST
    add_nodes(count) -- POST
    remove_nodes(count) -- POST
    list_nodes() -- GET
    get_service_info() -- GET
    get_node_info(serviceNodeId) -- GET
    list_code_versions() -- GET
    list_authorized_keys() -- GET
    upload_authorized_key(key) -- UPLOAD
    git_push_hook() -- POST
    upload_code_version(code, description) -- UPLOAD
    download_code_version(codeVersionId) -- GET
    enable_code(codeVersionId) -- POST
    delete_code_version(codeVersionId) -- POST
    list_volumes() -- GET
    generic_create_volume(volumeName, volumeSize, agentId) -- POST
    generic_delete_volume(volumeName) -- POST
    execute_script(command, parameters) -- POST
    get_script_status() -- GET
    get_agent_log(filename) -- GET
    """

    # String template for error messages returned when performing actions in
    # the wrong state
    WRONG_STATE_MSG = "ERROR: cannot perform %(action)s in state %(curstate)s"

    # String template for error messages returned when a required argument is
    # missing
    REQUIRED_ARG_MSG = "ERROR: %(arg)s is a required argument"

    # String template for debugging messages logged on nodes creation
    ACTION_REQUESTING_NODES = "requesting %(count)s nodes in %(action)s"

    # String used as an error message when 'interrupt' is called when no
    # scripts are currently running
    NO_SCRIPTS_ARE_RUNNING_MSG = "ERROR: No scripts are currently running inside "\
                                "agents. Nothing to interrupt."

    # String used as an error message when scripts are running inside agents
    SCRIPTS_ARE_RUNNING_MSG = "ERROR: Scripts are still running inside at "\
                                "least one agent. Please wait for them to "\
                                "finish execution or call 'interrupt' first."



    def __init__(self, config_parser, **kwargs):
        """Initialize a Generic Manager.

        'config_parser' represents the manager config file.
        **kwargs holds anything that can't be sent in config_parser."""
        BaseManager.__init__(self, config_parser)

        self.code_repo = config_parser.get('manager', 'CODE_REPO')

        if kwargs['reset_config']:
            self._create_initial_configuration()

        # self.nodes = []
        self.agents_info = []
        self.master_ip = None

    def _prepare_default_config_script(self, script_name):
        fileno, path = tempfile.mkstemp()
        fd = os.fdopen(fileno, 'w')
        fd.write('''#!/bin/bash
date >> /root/generic.out
echo "Executing script ${0##*/}" >> /root/generic.out
echo "Parameters ($#): $@" >> /root/generic.out
echo "My IP is $MY_IP" >> /root/generic.out
echo "My role is $MY_ROLE" >> /root/generic.out
echo "My master IP is $MASTER_IP" >> /root/generic.out
echo "Information about other agents is stored at /var/cache/cpsagent/agents.json" >> /root/generic.out
cat /var/cache/cpsagent/agents.json >> /root/generic.out
echo "" >> /root/generic.out
echo "" >> /root/generic.out
''')
        fd.close()
        os.chmod(path, stat.S_IRWXU | stat.S_IROTH | stat.S_IXOTH)
        return path

    def get_service_type(self):
        return 'generic'

    def _create_initial_configuration(self):
        self.logger.info("Creating initial configuration")

        config = ServiceConfiguration()

        if len(config.codeVersions) > 0:
            return

        if not os.path.exists(self.code_repo):
            os.makedirs(self.code_repo)

        tfile = tarfile.TarFile(name=os.path.join(self.code_repo, 'code-default'),
                mode='w')

        scripts = ['init.sh', 'notify.sh', 'run.sh', 'interrupt.sh', 'cleanup.sh']
        for script in scripts:
            path = self._prepare_default_config_script(script)
            tfile.add(path, script)
            os.remove(path)

        tfile.close()
        config.codeVersions['code-default'] = CodeVersion('code-default',
                'code-default.tar', 'tar', description='Initial version')
        config.currentCodeVersion = 'code-default'
        self._configuration_set(config)

   

    def on_start(self, nodes):
        """Start up the service. The first node will be the master node."""

        nr_instances = 1


        vals = { 'action': '_do_startup', 'count': nr_instances }
        self.logger.debug(self.ACTION_REQUESTING_NODES % vals)
        try:            
            config = self._configuration_get()

            roles = {'master':'1'}

            agents_info = self._update_agents_info(nodes, roles)

            self._init_agents(config, nodes, agents_info)
            self._update_code(config, nodes)

            # Extend the nodes list with the newly created one
            
            self.agents_info += agents_info
            self.master_ip = nodes[0].ip
            return True
        except Exception, err:
            self.logger.exception('_do_startup: Failed to create agents: %s' % err)
            return False
            

    def _update_agents_info(self, nodes, roles):
        id_ip = []
        for node in nodes:
            id_ip.append( { 'id': node.id, 'ip': node.ip })

        id_ip_role = []
        for role in roles:
            for _ in range(int(roles[role])):
                if len(id_ip):
                    node_ip_id = id_ip.pop(0)
                    node_ip_id.update({'role':role})
                    id_ip_role.append(node_ip_id)

        return id_ip_role

    def _init_agents(self, config, nodes, agents_info):
        self.logger.info("Initializing agents %s" %
                [ node.id for node in nodes ])

        for serviceNode in nodes:
            try:
                client.init_agent(serviceNode.ip, self.AGENT_PORT, agents_info)
            except client.AgentException:
                self.logger.exception('Failed initialize agent at node %s'
                        % str(serviceNode))
                self.state_set(self.S_ERROR, msg='Failed to initialize agent at node %s'
                        % str(serviceNode))
                raise
    
    def on_stop(self):
        """Delete all nodes and switch to status STOPPED"""
        

        self.logger.info("Removing nodes: %s" %
                [ node.id for node in self.nodes ])
        # self.controller.delete_nodes(self.nodes)
        del_nodes = self.nodes[:]
        self.agents_info = []
        self.master_ip = None
        return del_nodes

    def __check_count_in_args(self, kwargs):
        """Return 'count' if all is good. HttpErrorResponse otherwise."""
        count = None

        if 'count' in kwargs:
            count = kwargs.pop('count')
        # The frontend sends count under 'node'.
        elif 'node' in kwargs:
            count = kwargs.pop('node')
        else:
            return HttpErrorResponse(self.REQUIRED_ARG_MSG % { 'arg': 'count' })

        if not isinstance(count, int):
            return HttpErrorResponse(
                "ERROR: Expected an integer value for 'count'")

        return count

    def on_add_nodes(self, node_instances):
        # (genc): i have to figure out how to deal with the roles
        start_role = 'node'
        roles = {start_role: str(len(node_instances))}
        # Startup agents
        agents_info = self._update_agents_info(node_instances, roles)

        nodes_before = filter(lambda n: n not in node_instances, self.nodes)
        self.agents_info += agents_info

        config = self._configuration_get()
        self._init_agents(config, node_instances, self.agents_info)
        self._update_code(config, node_instances)

        self._do_execute_script('notify', nodes_before)
        return True


    def on_remove_nodes(self, node_roles):
        count = sum(node_roles.values())
        del_nodes = []
        cp_nodes = self.nodes[:]
        for _ in range(0, count):
            node = cp_nodes.pop()
            del_nodes += [ node ]
            self.agents_info.pop()
            self.logger.info("Removing node with IP %s" % node.ip)
        if not cp_nodes:
            self.master_ip = None
            self.state_set(self.S_STOPPED)
        else:
            self._do_execute_script('notify', cp_nodes)
            self.state_set(self.S_RUNNING)

        return del_nodes

    def __is_master(self, node):
        """Return True if the given node is the Generic master"""
        return node.ip == self.master_ip

    @expose('GET')
    def list_nodes(self, kwargs):
        """Return a list of running nodes"""
        state = self.state_get()
        if state != self.S_RUNNING:
{'fraction_non_alphanumeric': 0.08958210176890546, 'fraction_numerical': 0.0016370333318175617, 'mean_word_length': 2.5712893796687237, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7589345', 'n_tokens_mistral': 6344, 'n_tokens_neox': 6051, 'n_words': 1877}