# 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.
import gzip
import logging
import tarfile
import uuid
import xml.etree.cElementTree as elementTree
from xml.etree.cElementTree import Element
from io import StringIO
from io import BytesIO
from textwrap import dedent

import docker
import os
import yaml
from copy import copy

import time
from collections import OrderedDict

class Cluster(object):
    """
    Base Cluster class. This is intended to be a generic interface
    to different types of clusters. Clusters could be Kubernetes clusters,
    Docker swarms, or cloud compute/container services.
    """

    def deploy_flow(self, flow, name=None, vols=None):
        """
        Deploys a flow to the cluster.
        """

    def __enter__(self):
        """
        Allocate ephemeral cluster resources.
        """
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Clean up ephemeral cluster resources.
        """


class SingleNodeDockerCluster(Cluster):
    """
    A "cluster" which consists of a single docker node. Useful for
    testing or use-cases which do not span multiple compute nodes.
    """

    def __init__(self):
        self.minifi_version = os.environ['MINIFI_VERSION']
        self.nifi_version = '1.7.0'
        self.minifi_root = '/opt/minifi/nifi-minifi-cpp-' + self.minifi_version
        self.nifi_root = '/opt/nifi/nifi-' + self.nifi_version
        self.network = None
        self.containers = OrderedDict()
        self.images = []
        self.tmp_files = []

        # Get docker client
        self.client = docker.from_env()

    def deploy_flow(self,
                    flow,
                    name=None,
                    vols=None,
                    engine='minifi-cpp'):
        """
        Compiles the flow to a valid config file and overlays it into a new image.
        """

        if vols is None:
            vols = {}

        logging.info('Deploying %s flow...%s', engine,name)

        if name is None:
            name = engine + '-' + str(uuid.uuid4())
            logging.info('Flow name was not provided; using generated name \'%s\'', name)

        # Create network if necessary
        if self.network is None:
            net_name = 'nifi-' + str(uuid.uuid4())
            logging.info('Creating network: %s', net_name)
            self.network = self.client.networks.create(net_name)

        if engine == 'nifi':
            self.deploy_nifi_flow(flow, name, vols)
        elif engine == 'minifi-cpp':
            self.deploy_minifi_cpp_flow(flow, name, vols)
        elif engine == 'kafka-broker':
            self.deploy_kafka_broker(name)
        elif engine == 'http-proxy':
            self.deploy_http_proxy()
        else:
            raise Exception('invalid flow engine: \'%s\'' % engine)

    def deploy_minifi_cpp_flow(self, flow, name, vols):

        # Build configured image
        dockerfile = dedent("""FROM {base_image}
                USER root
                ADD config.yml {minifi_root}/conf/config.yml
                RUN chown minificpp:minificpp {minifi_root}/conf/config.yml
                USER minificpp
                """.format(name=name,hostname=name,
                           base_image='apacheminificpp:minimal-' + self.minifi_version,
                           minifi_root=self.minifi_root))

        test_flow_yaml = minifi_flow_yaml(flow)
        logging.info('Using generated flow config yml:\n%s', test_flow_yaml)

        conf_file_buffer = BytesIO()

        try:
            conf_file_buffer.write(test_flow_yaml.encode('utf-8'))
            conf_file_len = conf_file_buffer.tell()
            conf_file_buffer.seek(0)

            context_files = [
                {
                    'name': 'config.yml',
                    'size': conf_file_len,
                    'file_obj': conf_file_buffer
                }
            ]

            configured_image = self.build_image(dockerfile, context_files)

        finally:
            conf_file_buffer.close()

        logging.info('Creating and running docker container for flow...')

        container = self.client.containers.run(
                configured_image[0],
                detach=True,
                name=name,
                network=self.network.name,
                volumes=vols)

        logging.info('Started container \'%s\'', container.name)

        self.containers[container.name] = container

    def deploy_nifi_flow(self, flow, name, vols):
        dockerfile = dedent("""FROM {base_image}
                USER root
                ADD flow.xml.gz {nifi_root}/conf/flow.xml.gz
                RUN chown nifi:nifi {nifi_root}/conf/flow.xml.gz
                RUN sed -i -e 's/^\(nifi.remote.input.host\)=.*/\\1={name}/' {nifi_root}/conf/nifi.properties
                RUN sed -i -e 's/^\(nifi.remote.input.socket.port\)=.*/\\1=5000/' {nifi_root}/conf/nifi.properties
                USER nifi
                """.format(name=name,
                           base_image='apache/nifi:' + self.nifi_version,
                           nifi_root=self.nifi_root))

        test_flow_xml = nifi_flow_xml(flow, self.nifi_version)
        logging.info('Using generated flow config xml:\n%s', test_flow_xml)

        conf_file_buffer = BytesIO()

        try:
            with gzip.GzipFile(mode='wb', fileobj=conf_file_buffer) as conf_gz_file_buffer:
                conf_gz_file_buffer.write(test_flow_xml.encode())
            conf_file_len = conf_file_buffer.tell()
            conf_file_buffer.seek(0)

            context_files = [
                {
                    'name': 'flow.xml.gz',
                    'size': conf_file_len,
                    'file_obj': conf_file_buffer
                }
            ]

            configured_image = self.build_image(dockerfile, context_files)

        finally:
            conf_file_buffer.close()

        logging.info('Creating and running docker container for flow...')

        container = self.client.containers.run(
                configured_image[0],
                detach=True,
                name=name,
                hostname=name,
                network=self.network.name,
                volumes=vols)

        logging.info('Started container \'%s\'', container.name)

        self.containers[container.name] = container

    def deploy_kafka_broker(self, name):
        logging.info('Creating and running docker containers for kafka broker...')
        zookeeper = self.client.containers.run(
                    self.client.images.pull("wurstmeister/zookeeper:latest"),
                    detach=True,
                    name='zookeeper',
                    network=self.network.name,
                    ports={'2181/tcp': 2181},
                    )
        self.containers[zookeeper.name] = zookeeper

        test_dir = os.environ['PYTHONPATH'].split(':')[-1] # Based on DockerVerify.sh
        broker_image = self.build_image_by_path(test_dir + "/resources/kafka_broker", 'minifi-kafka')
        broker = self.client.containers.run(
                    broker_image[0],
                    detach=True,
                    name='kafka-broker',
                    network=self.network.name,
                    ports={'9092/tcp': 9092},
                    environment=["KAFKA_LISTENERS=PLAINTEXT://kafka-broker:9092,SSL://kafka-broker:9093", "KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181"],
                    )
        self.containers[broker.name] = broker

        dockerfile = dedent("""FROM {base_image}
                USER root
                CMD $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server kafka-broker:9092 --topic test > heaven_signal.txt
                """.format(base_image='wurstmeister/kafka:2.12-2.5.0'))
        configured_image = self.build_image(dockerfile, [])
        consumer = self.client.containers.run(
                    configured_image[0],
                    detach=True,
                    name='kafka-consumer',
                    network=self.network.name,
                    )
        self.containers[consumer.name] = consumer

    def deploy_http_proxy(self):
        logging.info('Creating and running http-proxy docker container...')
        dockerfile = dedent("""FROM {base_image}
                RUN apt update && apt install -y apache2-utils
                RUN htpasswd -b -c /etc/squid/.squid_users {proxy_username} {proxy_password}
                RUN echo 'auth_param basic program /usr/lib/squid3/basic_ncsa_auth /etc/squid/.squid_users'  > /etc/squid/squid.conf && \
                    echo 'auth_param basic realm proxy' >> /etc/squid/squid.conf && \
                    echo 'acl authenticated proxy_auth REQUIRED' >> /etc/squid/squid.conf && \
                    echo 'http_access allow authenticated' >> /etc/squid/squid.conf && \
                    echo 'http_port {proxy_port}' >> /etc/squid/squid.conf
                ENTRYPOINT ["/sbin/entrypoint.sh"]
                """.format(base_image='sameersbn/squid:3.5.27-2', proxy_username='admin', proxy_password='test101', proxy_port='3128'))
        configured_image = self.build_image(dockerfile, [])
        consumer = self.client.containers.run(
                    configured_image[0],
                    detach=True,
                    name='http-proxy',
                    network=self.network.name,
                    ports={'3128/tcp': 3128},
                    )
        self.containers[consumer.name] = consumer

    def build_image(self, dockerfile, context_files):
        conf_dockerfile_buffer = BytesIO()
        docker_context_buffer = BytesIO()

        try:
            # Overlay conf onto base nifi image
            conf_dockerfile_buffer.write(dockerfile.encode())
            conf_dockerfile_buffer.seek(0)

            with tarfile.open(mode='w', fileobj=docker_context_buffer) as docker_context:
                dockerfile_info = tarfile.TarInfo('Dockerfile')
                dockerfile_info.size = conf_dockerfile_buffer.getbuffer().nbytes
                docker_context.addfile(dockerfile_info,
                                       fileobj=conf_dockerfile_buffer)

                for context_file in context_files:
                    file_info = tarfile.TarInfo(context_file['name'])
                    file_info.size = context_file['size']
                    docker_context.addfile(file_info,
                                           fileobj=context_file['file_obj'])
            docker_context_buffer.seek(0)

            logging.info('Creating configured image...')
            configured_image = self.client.images.build(fileobj=docker_context_buffer,
                                                        custom_context=True,
                                                        rm=True,
                                                        forcerm=True)
            logging.info('Created image with id: %s', configured_image[0].id)
            self.images.append(configured_image)

        finally:
            conf_dockerfile_buffer.close()
            docker_context_buffer.close()

        return configured_image

    def build_image_by_path(self, dir, name=None):
        try:
            logging.info('Creating configured image...')
            configured_image = self.client.images.build(path=dir,
                                                        tag=name,
                                                        rm=True,
                                                        forcerm=True)
            logging.info('Created image with id: %s', configured_image[0].id)
            self.images.append(configured_image)
            return configured_image
        except Exception as e:
            logging.info(e)
            raise

    def __enter__(self):
        """
        Allocate ephemeral cluster resources.
        """
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Clean up ephemeral cluster resources
        """

        # Clean up containers
        for container in self.containers.values():
            logging.info('Cleaning up container: %s', container.name)
            container.remove(v=True, force=True)

        # Clean up images
        for image in reversed(self.images):
            logging.info('Cleaning up image: %s', image[0].id)
            self.client.images.remove(image[0].id, force=True)

        # Clean up network
        if self.network is not None:
            logging.info('Cleaning up network network: %s', self.network.name)
            self.network.remove()

        # Clean up tmp files
        for tmp_file in self.tmp_files:
            os.remove(tmp_file)


class Connectable(object):
    def __init__(self,
                 name=None,
                 auto_terminate=None):

        self.uuid = uuid.uuid4()

        if name is None:
            self.name = str(self.uuid)
        else:
            self.name = name

        if auto_terminate is None:
            self.auto_terminate = []
        else:
            self.auto_terminate = auto_terminate

        self.connections = {}
        self.out_proc = self

        self.drop_empty_flowfiles = False

    def connect(self, connections):
        for rel in connections:

            # Ensure that rel is not auto-terminated
            if rel in self.auto_terminate:
                del self.auto_terminate[self.auto_terminate.index(rel)]

            # Add to set of output connections for this rel
            if rel not in self.connections:
                self.connections[rel] = []
            self.connections[rel].append(connections[rel])

        return self

    def __rshift__(self, other):
        """
        Right shift operator to support flow DSL, for example:

            GetFile('/input') >> LogAttribute() >> PutFile('/output')

        """

        connected = copy(self)
        connected.connections = copy(self.connections)

        if self.out_proc is self:
            connected.out_proc = connected
        else:
            connected.out_proc = copy(connected.out_proc)

        if isinstance(other, tuple):
            if isinstance(other[0], tuple):
                for rel_tuple in other:
                    rel = {rel_tuple[0]: rel_tuple[1]}
                    connected.out_proc.connect(rel)
            else:
                rel = {other[0]: other[1]}
                connected.out_proc.connect(rel)
        else:
            connected.out_proc.connect({'success': other})
            connected.out_proc = other

        return connected

    def __invert__(self):
        """
        Invert operation to set empty file filtering on incoming connections
        GetFile('/input') >> ~LogAttribute()
        """
        self.drop_empty_flowfiles = True

        return self


class Processor(Connectable):
    def __init__(self,
                 clazz,
                 properties=None,
                 schedule=None,
                 name=None,
                 controller_services=None,
                 auto_terminate=None):

        super(Processor, self).__init__(name=name,
                                        auto_terminate=auto_terminate)

        if controller_services is None:
            controller_services = []

        if schedule is None:
            schedule = {}

        if properties is None:
            properties = {}

        if name is None:
            pass

        self.clazz = clazz
        self.properties = properties
        self.controller_services = controller_services

        self.schedule = {
            'scheduling strategy': 'TIMER_DRIVEN',
            'scheduling period': '1 sec',
            'penalization period': '30 sec',
            'yield period': '1 sec',
            'run duration nanos': 0
        }
        self.schedule.update(schedule)

    def nifi_property_key(self, key):
        """
        Returns the Apache NiFi-equivalent property key for the given key. This is often, but not always, the same as
        the internal key.
        """
        return key


class InvokeHTTP(Processor):
    def __init__(self, url,
                 method='GET',
                 proxy_host='',
                 proxy_port='',
                 proxy_username='',
                 proxy_password='',
                 ssl_context_service=None):
        properties = {'Remote URL': url,
                      'HTTP Method': method,
                      'Proxy Host': proxy_host,
                      'Proxy Port': proxy_port,
                      'invokehttp-proxy-username': proxy_username,
                      'invokehttp-proxy-password': proxy_password}

        controller_services = []

        if ssl_context_service is not None:
            properties['SSL Context Service'] = ssl_context_service.name
            controller_services.append(ssl_context_service)

        super(InvokeHTTP, self).__init__('InvokeHTTP',
                                         properties=properties,
                                         controller_services=controller_services,
                                         auto_terminate=['success',
                                                         'response',
                                                         'retry',
                                                         'failure',
                                                         'no retry'])


class ListenHTTP(Processor):
    def __init__(self, port, cert=None):
        properties = {'Listening Port': port}

        if cert is not None:
            properties['SSL Certificate'] = cert
            properties['SSL Verify Peer'] = 'no'

        super(ListenHTTP, self).__init__('ListenHTTP',
                                         properties=properties,
                                         auto_terminate=['success'])


class LogAttribute(Processor):
    def __init__(self, ):
        super(LogAttribute, self).__init__('LogAttribute',
                                           auto_terminate=['success'])


class DebugFlow(Processor):
    def __init__(self, ):
        super(DebugFlow, self).__init__('DebugFlow')

class AttributesToJSON(Processor):
    def __init__(self, destination, attributes):
        super(AttributesToJSON, self).__init__('AttributesToJSON',
                                      properties={'Destination': destination, 'Attributes List': attributes},
                                      schedule={'scheduling period': '0 sec'},
                                      auto_terminate=['failure'])
class GetFile(Processor):
    def __init__(self, input_dir):
        super(GetFile, self).__init__('GetFile',
                                      properties={'Input Directory': input_dir, 'Keep Source File': 'true'},
                                      schedule={'scheduling period': '2 sec'},
                                      auto_terminate=['success'])

class GenerateFlowFile(Processor):
    def __init__(self, file_size):
        super(GenerateFlowFile, self).__init__('GenerateFlowFile',
                                      properties={'File Size': file_size},
                                      schedule={'scheduling period': '0 sec'},
                                      auto_terminate=['success'])


class PutFile(Processor):
    def __init__(self, output_dir):
        super(PutFile, self).__init__('PutFile',
                                      properties={'Directory': output_dir},
                                      auto_terminate=['success', 'failure'])

    def nifi_property_key(self, key):
        if key == 'Output Directory':
            return 'Directory'
        else:
            return key


class PublishKafka(Processor):
    def __init__(self):
        super(PublishKafka, self).__init__('PublishKafka',
                                           properties={'Client Name': 'nghiaxlee', 'Known Brokers': 'kafka-broker:9092', 'Topic Name': 'test',
                                                       'Batch Size': '10', 'Compress Codec': 'none', 'Delivery Guarantee': '1',
                                                       'Request Timeout': '10 sec', 'Message Timeout': '12 sec'},
                                           auto_terminate=['success'])


class PublishKafkaSSL(Processor):
    def __init__(self):
        super(PublishKafkaSSL, self).__init__('PublishKafka',
                                              properties={'Client Name': 'LMN', 'Known Brokers': 'kafka-broker:9093',
                                                          'Topic Name': 'test', 'Batch Size': '10',
                                                          'Compress Codec': 'none', 'Delivery Guarantee': '1',
                                                          'Request Timeout': '10 sec', 'Message Timeout': '12 sec',
                                                          'Security CA': '/tmp/resources/certs/ca-cert',
                                                          'Security Cert': '/tmp/resources/certs/client_LMN_client.pem',
                                                          'Security Pass Phrase': 'abcdefgh',
                                                          'Security Private Key': '/tmp/resources/certs/client_LMN_client.key',
                                                          'Security Protocol': 'ssl'},
                                              auto_terminate=['success'])


class InputPort(Connectable):
    def __init__(self, name=None, remote_process_group=None):
        super(InputPort, self).__init__(name=name)

        self.remote_process_group = remote_process_group


class RemoteProcessGroup(object):
    def __init__(self, url,
                 name=None):

        self.uuid = uuid.uuid4()

        if name is None:
            self.name = str(self.uuid)
        else:
            self.name = name

        self.url = url


class ControllerService(object):
    def __init__(self, name=None, properties=None):

        self.id = str(uuid.uuid4())

        if name is None:
            self.name = str(uuid.uuid4())
            logging.info('Controller service name was not provided; using generated name \'%s\'', self.name)
        else:
            self.name = name

        if properties is None:
            properties = {}

        self.properties = properties


class SSLContextService(ControllerService):
    def __init__(self, name=None, cert=None, key=None, ca_cert=None):
        super(SSLContextService, self).__init__(name=name)

        self.service_class = 'SSLContextService'

        if cert is not None:
            self.properties['Client Certificate'] = cert

        if key is not None:
            self.properties['Private Key'] = key

        if ca_cert is not None:
            self.properties['CA Certificate'] = ca_cert


def minifi_flow_yaml(connectable, root=None, visited=None):
    if visited is None:
        visited = []

    if root is None:
        res = {
            'Flow Controller': {
                'name': 'MiNiFi Flow'
            },
            'Processors': [],
            'Connections': [],
            'Remote Processing Groups': [],
            'Controller Services': []
        }
    else:
        res = root

    visited.append(connectable)

    if hasattr(connectable, 'name'):
        connectable_name = connectable.name
    else:
        connectable_name = str(connectable.uuid)

    if isinstance(connectable, InputPort):
        group = connectable.remote_process_group
        res_group = None

        for res_group_candidate in res['Remote Processing Groups']:
            assert isinstance(res_group_candidate, dict)
            if res_group_candidate['id'] == str(group.uuid):
                res_group = res_group_candidate

        if res_group is None:
            res_group = {
                'name': group.name,
                'id': str(group.uuid),
                'url': group.url,
                'timeout': '30 sec',
                'yield period': '3 sec',
                'Input Ports': []
            }

            res['Remote Processing Groups'].append(res_group)

        res_group['Input Ports'].append({
            'id': str(connectable.uuid),
            'name': connectable.name,
            'max concurrent tasks': 1,
            'Properties': {}
        })

    if isinstance(connectable, Processor):
        res['Processors'].append({
            'name': connectable_name,
            'id': str(connectable.uuid),
            'class': 'org.apache.nifi.processors.standard.' + connectable.clazz,
            'scheduling strategy': connectable.schedule['scheduling strategy'],
            'scheduling period': connectable.schedule['scheduling period'],
            'penalization period': connectable.schedule['penalization period'],
            'yield period': connectable.schedule['yield period'],
            'run duration nanos': connectable.schedule['run duration nanos'],
            'Properties': connectable.properties,
            'auto-terminated relationships list': connectable.auto_terminate
        })

        for svc in connectable.controller_services:
            if svc in visited:
                continue

            visited.append(svc)
            res['Controller Services'].append({
                'name': svc.name,
                'id': svc.id,
                'class': svc.service_class,
                'Properties': svc.properties
            })

    for conn_name in connectable.connections:
        conn_procs = connectable.connections[conn_name]

        if isinstance(conn_procs, list):
            for proc in conn_procs:
                res['Connections'].append({
                    'name': str(uuid.uuid4()),
                    'source id': str(connectable.uuid),
                    'source relationship name': conn_name,
                    'destination id': str(proc.uuid),
                    'drop empty': ("true" if proc.drop_empty_flowfiles else "false")
                })
                if proc not in visited:
                    minifi_flow_yaml(proc, res, visited)
        else:
            res['Connections'].append({
                'name': str(uuid.uuid4()),
                'source id': str(connectable.uuid),
                'source relationship name': conn_name,
                'destination id': str(conn_procs.uuid)
            })
            if conn_procs not in visited:
                minifi_flow_yaml(conn_procs, res, visited)

    if root is None:
        return yaml.dump(res, default_flow_style=False)


def nifi_flow_xml(connectable, nifi_version=None, root=None, visited=None):
    if visited is None:
        visited = []

    position = Element('position')
    position.set('x', '0.0')
    position.set('y', '0.0')

    comment = Element('comment')
    styles = Element('styles')
    bend_points = Element('bendPoints')
    label_index = Element('labelIndex')
    label_index.text = '1'
    z_index = Element('zIndex')
    z_index.text = '0'

    if root is None:
        res = Element('flowController')
        max_timer_driven_thread_count = Element('maxTimerDrivenThreadCount')
        max_timer_driven_thread_count.text = '10'
        res.append(max_timer_driven_thread_count)
        max_event_driven_thread_count = Element('maxEventDrivenThreadCount')
        max_event_driven_thread_count.text = '5'
        res.append(max_event_driven_thread_count)
        root_group = Element('rootGroup')
        root_group_id = Element('id')
        root_group_id_text = str(uuid.uuid4())
        root_group_id.text = root_group_id_text
        root_group.append(root_group_id)
        root_group_name = Element('name')
        root_group_name.text = root_group_id_text
        root_group.append(root_group_name)
        res.append(root_group)
        root_group.append(position)
        root_group.append(comment)
        res.append(Element('controllerServices'))
        res.append(Element('reportingTasks'))
        res.set('encoding-version', '1.2')
    else:
        res = root

    visited.append(connectable)

    if hasattr(connectable, 'name'):
        connectable_name_text = connectable.name
    else:
        connectable_name_text = str(connectable.uuid)

    if isinstance(connectable, InputPort):
        input_port = Element('inputPort')

        input_port_id = Element('id')
        input_port_id.text = str(connectable.uuid)
        input_port.append(input_port_id)

        input_port_name = Element('name')
        input_port_name.text = connectable_name_text
        input_port.append(input_port_name)

        input_port.append(position)
        input_port.append(comment)

        input_port_scheduled_state = Element('scheduledState')
        input_port_scheduled_state.text = 'RUNNING'
        input_port.append(input_port_scheduled_state)

        input_port_max_concurrent_tasks = Element('maxConcurrentTasks')
        input_port_max_concurrent_tasks.text = '1'
        input_port.append(input_port_max_concurrent_tasks)
        next( res.iterfind('rootGroup') ).append(input_port)

    if isinstance(connectable, Processor):
        conn_destination = Element('processor')

        proc_id = Element('id')
        proc_id.text = str(connectable.uuid)
        conn_destination.append(proc_id)

        proc_name = Element('name')
        proc_name.text = connectable_name_text
        conn_destination.append(proc_name)

        conn_destination.append(position)
        conn_destination.append(styles)
        conn_destination.append(comment)

        proc_class = Element('class')
        proc_class.text = 'org.apache.nifi.processors.standard.' + connectable.clazz
        conn_destination.append(proc_class)

        proc_bundle = Element('bundle')
        proc_bundle_group = Element('group')
        proc_bundle_group.text = 'org.apache.nifi'
        proc_bundle.append(proc_bundle_group)
        proc_bundle_artifact = Element('artifact')
        proc_bundle_artifact.text = 'nifi-standard-nar'
        proc_bundle.append(proc_bundle_artifact)
        proc_bundle_version = Element('version')
        proc_bundle_version.text = nifi_version
        proc_bundle.append(proc_bundle_version)
        conn_destination.append(proc_bundle)

        proc_max_concurrent_tasks = Element('maxConcurrentTasks')
        proc_max_concurrent_tasks.text = '1'
        conn_destination.append(proc_max_concurrent_tasks)

        proc_scheduling_period = Element('schedulingPeriod')
        proc_scheduling_period.text = connectable.schedule['scheduling period']
        conn_destination.append(proc_scheduling_period)

        proc_penalization_period = Element('penalizationPeriod')
        proc_penalization_period.text = connectable.schedule['penalization period']
        conn_destination.append(proc_penalization_period)

        proc_yield_period = Element('yieldPeriod')
        proc_yield_period.text = connectable.schedule['yield period']
        conn_destination.append(proc_yield_period)

        proc_bulletin_level = Element('bulletinLevel')
        proc_bulletin_level.text = 'WARN'
        conn_destination.append(proc_bulletin_level)

        proc_loss_tolerant = Element('lossTolerant')
        proc_loss_tolerant.text = 'false'
        conn_destination.append(proc_loss_tolerant)

        proc_scheduled_state = Element('scheduledState')
        proc_scheduled_state.text = 'RUNNING'
        conn_destination.append(proc_scheduled_state)

        proc_scheduling_strategy = Element('schedulingStrategy')
        proc_scheduling_strategy.text = connectable.schedule['scheduling strategy']
        conn_destination.append(proc_scheduling_strategy)

        proc_execution_node = Element('executionNode')
        proc_execution_node.text = 'ALL'
        conn_destination.append(proc_execution_node)

        proc_run_duration_nanos = Element('runDurationNanos')
        proc_run_duration_nanos.text = str(connectable.schedule['run duration nanos'])
        conn_destination.append(proc_run_duration_nanos)

        for property_key, property_value in connectable.properties.items():
            proc_property = Element('property')
            proc_property_name = Element('name')
            proc_property_name.text = connectable.nifi_property_key(property_key)
            proc_property.append(proc_property_name)
            proc_property_value = Element('value')
            proc_property_value.text = property_value
            proc_property.append(proc_property_value)
            conn_destination.append(proc_property)

        for auto_terminate_rel in connectable.auto_terminate:
            proc_auto_terminated_relationship = Element('autoTerminatedRelationship')
            proc_auto_terminated_relationship.text = auto_terminate_rel
            conn_destination.append(proc_auto_terminated_relationship)
        next( res.iterfind('rootGroup') ).append(conn_destination)
        """ res.iterfind('rootGroup').next().append(conn_destination) """

        for svc in connectable.controller_services:
            if svc in visited:
                continue

            visited.append(svc)
            controller_service = Element('controllerService')

            controller_service_id = Element('id')
            controller_service_id.text = str(svc.id)
            controller_service.append(controller_service_id)

            controller_service_name = Element('name')
            controller_service_name.text = svc.name
            controller_service.append(controller_service_name)

            controller_service.append(comment)

            controller_service_class = Element('class')
            controller_service_class.text = svc.service_class,
            controller_service.append(controller_service_class)

            controller_service_bundle = Element('bundle')
            controller_service_bundle_group = Element('group')
            controller_service_bundle_group.text = svc.group
            controller_service_bundle.append(controller_service_bundle_group)
            controller_service_bundle_artifact = Element('artifact')
            controller_service_bundle_artifact.text = svc.artifact
            controller_service_bundle.append(controller_service_bundle_artifact)
            controller_service_bundle_version = Element('version')
            controller_service_bundle_version.text = nifi_version
            controller_service_bundle.append(controller_service_bundle_version)
            controller_service.append(controller_service_bundle)

            controller_enabled = Element('enabled')
            controller_enabled.text = 'true',
            controller_service.append(controller_enabled)

            for property_key, property_value in svc.properties:
                controller_service_property = Element('property')
                controller_service_property_name = Element('name')
                controller_service_property_name.text = property_key
                controller_service_property.append(controller_service_property_name)
                controller_service_property_value = Element('value')
                controller_service_property_value.text = property_value
                controller_service_property.append(controller_service_property_value)
                controller_service.append(controller_service_property)
            next( res.iterfind('rootGroup') ).append(controller_service)
            """ res.iterfind('rootGroup').next().append(controller_service)"""

    for conn_name in connectable.connections:
        conn_destinations = connectable.connections[conn_name]

        if isinstance(conn_destinations, list):
            for conn_destination in conn_destinations:
                connection = nifi_flow_xml_connection(res,
                                                      bend_points,
                                                      conn_name,
                                                      connectable,
                                                      label_index,
                                                      conn_destination,
                                                      z_index)
                next( res.iterfind('rootGroup') ).append(connection)
                """ res.iterfind('rootGroup').next().append(connection) """

                if conn_destination not in visited:
                    nifi_flow_xml(conn_destination, nifi_version, res, visited)
        else:
            connection = nifi_flow_xml_connection(res,
                                                  bend_points,
                                                  conn_name,
                                                  connectable,
                                                  label_index,
                                                  conn_destinations,
                                                  z_index)
            next( res.iterfind('rootGroup') ).append(connection)
            """ res.iterfind('rootGroup').next().append(connection) """

            if conn_destinations not in visited:
                nifi_flow_xml(conn_destinations, nifi_version, res, visited)

    if root is None:
        return ('<?xml version="1.0" encoding="UTF-8" standalone="no"?>'
                + "\n"
                + elementTree.tostring(res, encoding='utf-8').decode('utf-8'))


def nifi_flow_xml_connection(res, bend_points, conn_name, connectable, label_index, destination, z_index):
    connection = Element('connection')

    connection_id = Element('id')
    connection_id.text = str(uuid.uuid4())
    connection.append(connection_id)

    connection_name = Element('name')
    connection.append(connection_name)

    connection.append(bend_points)
    connection.append(label_index)
    connection.append(z_index)

    connection_source_id = Element('sourceId')
    connection_source_id.text = str(connectable.uuid)
    connection.append(connection_source_id)

    connection_source_group_id = Element('sourceGroupId')
    connection_source_group_id.text = next( res.iterfind('rootGroup/id') ).text
    """connection_source_group_id.text = res.iterfind('rootGroup/id').next().text"""
    connection.append(connection_source_group_id)

    connection_source_type = Element('sourceType')
    if isinstance(connectable, Processor):
        connection_source_type.text = 'PROCESSOR'
    elif isinstance(connectable, InputPort):
        connection_source_type.text = 'INPUT_PORT'
    else:
        raise Exception('Unexpected source type: %s' % type(connectable))
    connection.append(connection_source_type)

    connection_destination_id = Element('destinationId')
    connection_destination_id.text = str(destination.uuid)
    connection.append(connection_destination_id)

    connection_destination_group_id = Element('destinationGroupId')
    connection_destination_group_id.text = next(res.iterfind('rootGroup/id')).text
    """ connection_destination_group_id.text = res.iterfind('rootGroup/id').next().text """
    connection.append(connection_destination_group_id)

    connection_destination_type = Element('destinationType')
    if isinstance(destination, Processor):
        connection_destination_type.text = 'PROCESSOR'
    elif isinstance(destination, InputPort):
        connection_destination_type.text = 'INPUT_PORT'
    else:
        raise Exception('Unexpected destination type: %s' % type(destination))
    connection.append(connection_destination_type)

    connection_relationship = Element('relationship')
    if not isinstance(connectable, InputPort):
        connection_relationship.text = conn_name
    connection.append(connection_relationship)

    connection_max_work_queue_size = Element('maxWorkQueueSize')
    connection_max_work_queue_size.text = '10000'
    connection.append(connection_max_work_queue_size)

    connection_max_work_queue_data_size = Element('maxWorkQueueDataSize')
    connection_max_work_queue_data_size.text = '1 GB'
    connection.append(connection_max_work_queue_data_size)

    connection_flow_file_expiration = Element('flowFileExpiration')
    connection_flow_file_expiration.text = '0 sec'
    connection.append(connection_flow_file_expiration)

    return connection