heat_template_version: 2017-02-24

parameters:

  image_to_use: { type: string, label: "Image name or ID" }

  flavor_to_use: { type: string, label: "Flavor name" }

  key_name_user: { type: string, label: "User ssh public key" }

  root_pw: { type: string, label: "Root password" }

  avail_zone: { type: string, label: "Availability Zone", default: "nova" }

  host_prefix: { type: string, label: "Hostname prefix", default: "bigdata-test" }

  host_name: { type: string, label: "Server Hostname", default: "undefined" }

  lan_net_id: { type: string, label: "LAN Network ID", default: "undefined" }
  
  lan_subnet_name: { type: string, label: "LAN Sub network", default: "undefined" }

  lan_net_ippref: { type: string, label: "LAN IPv4 prefix" }

  wan_net_id: { type: string, label: "WAN Network ID", default: "undefined" }

  wan_subnet_name: { type: string, label: "WAN Sub network", default: "undefined" }

  wan_net_ippref: { type: string, label: "WAN IPv4 prefix", default: "undefined" }

  nameserver_list: { type: string, label: "Name server ip list" }

  sec_group_id: { type: string, label: "Security Group ID" }

  node_type: { type: string, label: "Node type", default: "generic" }

  node_id: { type: number, label: "Node ID" }

  kafka_ids: { type: string, label: "Kafka host id" }

  spark_ids: { type: string, label: "Spark host id" }

  admin_id: { type: string, label: "Administrator ID", default: "undefined" }

  fip_id: { type: string, label: "Floating point ID", default: "undefined" }

  oidc_client_id: { type: string, label: "OpenID Connect Client ID", default: "undefined" }

  oidc_client_secret: { type: string, label: "OpenID Connect Client secret", default: "undefined" }

  oidc_idp_url: { type: string, label: "OpenID Connect IdP metadata", default: "undefined" }

conditions:

  public_access: { not: { equals: [ { get_param: wan_net_id }, "undefined" ] } }

  internal_access: { not: { equals: [ { get_param: lan_net_id }, "undefined" ] } }


resources:

  internal_port:
    type: OS::Neutron::Port
    condition: internal_access
    properties:
      name: { str_replace: { template: "internal-port-<%node_id%>", params: { <%node_id%>: { get_param: node_id } } } }
      network_id: { get_param: lan_net_id }
      fixed_ips:
        - {
            subnet: { get_param: lan_subnet_name },
            ip_address: {
              str_replace: {
                template: "<%ippref%>.<%node_id%>",
                params: { <%ippref%>: { get_param: lan_net_ippref }, <%node_id%>: { get_param: node_id } }
              }
            }
          }
      security_groups: [ { get_param: sec_group_id }, ]


  public_port:
    type: OS::Neutron::Port
    condition: public_access
    properties:
      name: { str_replace: { template: "public-port-<%node_id%>", params: { <%node_id%>: { get_param: node_id } } } }
      network_id: { get_param: wan_net_id }
      fixed_ips:
        - {
            subnet: { get_param: wan_subnet_name },
            ip_address: {
                str_replace: {
                  template: "<%ippref%>.<%node_id%>",
                  params: { <%ippref%>: { get_param: wan_net_ippref }, <%node_id%>: { get_param: node_id } }
                }
            }
          }
      security_groups: [ { get_param: sec_group_id }, ]

  public_fip_ass:
    type: OS::Neutron::FloatingIPAssociation
    condition: public_access
    properties:
      floatingip_id: { get_param: fip_id }
      port_id: { get_resource: public_port }

  server_instance:
    type: OS::Nova::Server
    properties:
      name: {
        str_replace: {
          template: "<%host_pref%>-<%node_id%>",
          params: { <%host_pref%> : { get_param: host_prefix }, <%node_id%>: { get_param: node_id } }
        }
      }
      key_name: { get_param: key_name_user }
      image: { get_param: image_to_use }
      flavor: { get_param: flavor_to_use }
      admin_pass: { get_param: root_pw }
      networks:
        - port: { if: [ "internal_access", { get_resource: internal_port }, { get_resource: public_port } ] }
      user_data_format: RAW
      user_data:
        str_replace:
          template: |
            #!/usr/bin/python
            import sys, os, os.path, re, traceback
            from subprocess import check_call
            try:
              #
              with open("/tmp/heat_multi_node.log", "a") as resfile:
                resfile.write(" Configuring ssh\n")
              #
              host_prefix = "<%host_pref%>"
              node_id = <%node_id%>
              lan_ippre = "<%lan_ippre%>"
              nameservers = "<%ns_list%>"
              node_type = re.split("\s+", "<%node_type%>")
              rootpwd = "<%root_pwd%>"
              admin_id = "<%admin_id%>"
              oidc_id = "<%oidc_id%>" if "<%oidc_id%>".lower() <> "undefined" else ""
              oidc_sec = "<%oidc_sec%>" if "<%oidc_sec%>".lower() <> "undefined" else ""
              oidc_idp = "<%oidc_idp%>" if "<%oidc_idp%>".lower() <> "undefined" else ""
              servername = "<%host_name%>".strip()
              if not servername or servername == "undefined":
                servername = "%s-%d.pd.infn.it" % (host_prefix, node_id)
              workdir = "/opt/bigdata"
              bguser = "bigdatausr"
              #
              kafkaIDs = map(lambda x: int(x), re.findall(r'\d+', "<%kafka_ids%>"))
              sparkIDs = map(lambda x: int(x), re.findall(r'\d+', "<%spark_ids%>"))
              allIDs = set(kafkaIDs) | set(sparkIDs)
              #
              repo_url = "http://artifacts.pd.infn.it/packages/SMACT/misc"
              #
              check_call("hostnamectl set-hostname %s" % servername, shell=True)
              #
              with open("/etc/hosts", "w") as hostsfile:
                hostsfile.write("127.0.0.1 localhost.pd.infn.it localhost\n")
                hostsfile.write("::1       localhost.pd.infn.it localhost\n")
                for item in allIDs:
                  tmpn = "%s-%d" % (host_prefix, item)
                  hostsfile.write("%s.%d %s.pd.infn.it %s\n" % (lan_ippre, item, tmpn, tmpn))
              #
              with open("/etc/ssh/shosts.equiv", "w") as equivfile:
                for item in allIDs:
                  equivfile.write("%s-%d.pd.infn.it\n" % (host_prefix, item))
              #
              with open("/etc/resolv.conf", "w") as resolvfile:
                resolvfile.write("search pd.infn.it\n")
                resolvfile.write("nameserver %s\n" % nameservers)
              #
              with open("/etc/ssh/sshd_config", "a") as sshdfile:
                sshdfile.write("# workaround\n")
                sshdfile.write("HostbasedAuthentication yes\n")
                sshdfile.write("IgnoreUserKnownHosts yes\n")
                sshdfile.write("IgnoreRhosts yes\n")
              #
              with open("/etc/ssh/ssh_config", "w") as sshfile:
                sshfile.write("Host *\n")
                sshfile.write("    HostbasedAuthentication yes\n")
                sshfile.write("    EnableSSHKeysign yes\n")
                sshfile.write("    GSSAPIAuthentication no\n")
                sshfile.write("    ForwardX11Trusted yes\n")
                sshfile.write("    SendEnv LANG LC_CTYPE LC_NUMERIC LC_TIME LC_COLLATE LC_MONETARY LC_MESSAGES\n")
                sshfile.write("    SendEnv LC_PAPER LC_NAME LC_ADDRESS LC_TELEPHONE LC_MEASUREMENT\n")
                sshfile.write("    SendEnv LC_IDENTIFICATION LC_ALL LANGUAGE\n")
                sshfile.write("    SendEnv XMODIFIERS\n")
              #
              with open("/etc/sysctl.conf", "a") as sysfile:
                sysfile.write("net.ipv6.conf.all.disable_ipv6 = 1\n")
                sysfile.write("net.ipv6.conf.default.disable_ipv6 = 1\n")
              #
              check_call("useradd -d %s %s" % (workdir, bguser), shell=True)
              runcmd = "runuser -s /bin/bash -c \"%s\" -- " + bguser
              #
              pkglist = "epel-release yum-priorities vim-enhanced wget java-1.8.0-openjdk"
              check_call("yum -q -y install %s" % pkglist, shell=True)
              #
              #####################################################################################
              # Kafka setup
              #####################################################################################
              #
              if "kafka" in node_type:
                #
                with open("/tmp/heat_multi_node.log", "a") as resfile:
                  resfile.write(" Installing Kafka\n")
                #
                kafkaver = "2.11-1.1.0"
                #
                check_call("wget -O /tmp/kafka_%s.tgz %s/kafka_%s.tgz" % (kafkaver, repo_url, kafkaver), shell=True)
                check_call("tar -C %s -zxf /tmp/kafka_%s.tgz" % (workdir, kafkaver), shell=True)
                os.rename("%s/kafka_%s" % (workdir, kafkaver), "%s/kafka" % workdir)
                for d_item in [ "/var/lib/kafka", "/var/lib/kafka/kafka-logs", "/var/cache/zookeeper" ]:
                  os.makedirs(d_item, 0770)
                  check_call("chown %s.%s %s" % (bguser, bguser, d_item), shell=True)
                #
                with open("%s/kafka/config/server.properties" % workdir, "w") as kconf:
                  kconf.write("broker.id=%d\n" % node_id)
                  kconf.write("num.network.threads=3\n")
                  kconf.write("num.io.threads=8\n")
                  kconf.write("socket.send.buffer.bytes=102400\n")
                  kconf.write("socket.receive.buffer.bytes=102400\n")
                  kconf.write("socket.request.max.bytes=104857600\n")
                  kconf.write("log.dirs=/var/lib/kafka/kafka-logs\n")
                  kconf.write("num.partitions=1\n")
                  kconf.write("num.recovery.threads.per.data.dir=1\n")
                  kconf.write("offsets.topic.replication.factor=1\n")
                  kconf.write("transaction.state.log.replication.factor=1\n")
                  kconf.write("transaction.state.log.min.isr=1\n")
                  kconf.write("log.retention.hours=168\n")
                  kconf.write("log.segment.bytes=1073741824\n")
                  kconf.write("log.retention.check.interval.ms=300000\n")
                  kconf.write("zookeeper.connect=localhost:2181\n")
                  kconf.write("zookeeper.connection.timeout.ms=6000\n")
                  kconf.write("group.initial.rebalance.delay.ms=0\n")
                #
                with open("%s/kafka/config/zookeeper.properties" % workdir, "w") as zconf:
                  zconf.write("dataDir=/var/cache/zookeeper\n")
                  zconf.write("clientPort=2181\n")
                  zconf.write("tickTime=2000\n")
                  zconf.write("initLimit=5\n")
                  zconf.write("syncLimit=2\n")
                  for item in kafkaIDs:
                    zconf.write("server.%d=%s-%d.pd.infn.it:2888:3888\n" % (item, host_prefix, item))
                #
                with open("/var/cache/zookeeper/myid", "w") as idfile:
                  idfile.write("%d\n" % node_id)
                #
                with open("/usr/lib/systemd/system/zookeeper.service", "w") as zsrvfile:
                  zsrvfile.write("[Unit]\nDescription=Zookeeper service\n\n")
                  zsrvfile.write("[Service]\nExecStart=/usr/sbin/runuser -s /bin/bash ")
                  zsrvfile.write("-c \"%s/kafka/bin/zookeeper-server-start.sh " % workdir)
                  zsrvfile.write("%s/kafka/config/zookeeper.properties\" -- %s\n" % (workdir, bguser))
                  zsrvfile.write("ExecStop=/usr/sbin/runuser -s /bin/bash ")
                  zsrvfile.write("-c \"%s/kafka/bin/zookeeper-server-stop.sh\" -- %s\n\n" % (workdir, bguser))
                  zsrvfile.write("[Install]\nWantedBy=multi-user.target\n")
                #
                with open("/usr/lib/systemd/system/kafka.service", "w") as ksrvfile:
                  ksrvfile.write("[Unit]\nDescription=Kafka broker\n")
                  ksrvfile.write("Wants=zookeeper.service\nAfter=zookeeper.service\n\n")
                  ksrvfile.write("[Service]\nExecStart=/usr/sbin/runuser -s /bin/bash ")
                  ksrvfile.write("-c \"%s/kafka/bin/kafka-server-start.sh " %workdir)
                  ksrvfile.write("%s/kafka/config/server.properties\" -- %s\n"% (workdir, bguser))
                  ksrvfile.write("ExecStop=/usr/sbin/runuser -s /bin/bash ")
                  ksrvfile.write("-c \"%s/kafka/bin/kafka-server-start.sh\" -- %s\n\n" % (workdir, bguser))
                  ksrvfile.write("[Install]\nWantedBy=multi-user.target\n")
                #
                with open("/etc/bashrc", "a") as rcfile:
                  rcfile.write("export PATH=$PATH:%s/kafka/bin/\n" % workdir) 
              #
              #####################################################################################
              # Spark setup
              #####################################################################################
              #
              corecfg =  "<configuration>\n"
              corecfg += "  <property>\n"
              corecfg += "    <name>fs.defaultFS</name>\n"
              corecfg += "    <value>hdfs://%s-%d.pd.infn.it:9000/</value>\n" % (host_prefix, sparkIDs[0])
              corecfg += "  </property>\n"
              corecfg += "</configuration>\n"
              #
              hdfscfg =  "<configuration>\n"
              hdfscfg += "  <property><name>dfs.replication</name><value>1</value></property>\n"
              hdfscfg += "  <property><name>dfs.permissions</name><value>false</value></property>\n"
              hdfscfg += "  <property><name>dfs.datanode.data.dir</name>"
              hdfscfg += "<value>%s/datanode</value></property>\n" % workdir
              hdfscfg += "  <property><name>dfs.namenode.name.dir</name>"
              hdfscfg += "<value>%s/namenode</value></property>\n" % workdir
              hdfscfg += "  <property> <name>dfs.secondary.http.address</name>"
              hdfscfg += "<value>%s.%d:50090</value></property>\n" % (lan_ippre, sparkIDs[0])
              hdfscfg += "</configuration>\n"
              #
              if "spark" in node_type:
                #
                with open("/tmp/heat_multi_node.log", "a") as resfile:
                  resfile.write(" Installing Spark\n")
                #
                hdver = "2.9.1"
                spkver = "2.3.2"
                spk_wport = 6066
                #
                check_call("wget -O /tmp/hadoop-%s.tar.gz %s/hadoop-%s.tar.gz" % (hdver, repo_url, hdver), shell=True)
                check_call(runcmd % ("tar -C %s -zxf /tmp/hadoop-%s.tar.gz" % (workdir, hdver)), shell=True)
                os.rename("%s/hadoop-%s" % (workdir, hdver), "%s/hadoop" % workdir)
                #
                hdhome = workdir + "/hadoop"
                spkhome = workdir + "/spark"
                with open("/etc/profile.d/hadoop.sh", "w") as proffile:
                  proffile.write("export JAVA_HOME=/usr/lib/jvm/jre-1.8.0-openjdk\n")
                  proffile.write("export JRE_HOME=/usr/lib/jvm/jre-1.8.0-openjdk/jre\n")
                  proffile.write("export HADOOP_PREFIX=%s\n" % hdhome)
                  proffile.write("export HADOOP_HOME=%s\n" % hdhome)
                  proffile.write("export HADOOP_COMMON_HOME=%s\n" % hdhome)
                  proffile.write("export HADOOP_CONF_DIR=%s/etc/hadoop\n" % hdhome)
                  proffile.write("export HADOOP_HDFS_HOME=%s\n" % hdhome)
                  proffile.write("export HADOOP_MAPRED_HOME=%s\n" % hdhome)
                  proffile.write("export HADOOP_YARN_HOME=%s\n" % hdhome)
                  proffile.write("export PYTHONPATH=$PYTHONPATH:%s/python\n" % spkhome)
                  proffile.write("export PATH=$PATH:%s/bin:%s/sbin\n" % (hdhome, hdhome))
                  proffile.write("export PATH=$PATH:%s/bin:%s/sbin\n" % (spkhome, spkhome))
                #
                os.makedirs("%s/datanode" % workdir, 0770)
                os.makedirs("%s/namenode" % workdir, 0770)
                check_call("chown %s.%s %s/datanode %s/namenode" % (bguser, bguser, workdir, workdir), shell=True)
                #
                with open("%s/etc/hadoop/core-site.xml" % hdhome, "w") as corefile:
                  corefile.write(corecfg)
                #
                with open("%s/etc/hadoop/hdfs-site.xml" % hdhome, "w") as hdfscfile:
                  hdfscfile.write(hdfscfg)
                #
                with open("%s/etc/hadoop/mapred-site.xml" % hdhome, "w") as mrcfile:
                  mrcfile.write("<configuration>\n")
                  mrcfile.write("  <property><name>mapreduce.framework.name</name><value>yarn</value></property>\n")
                  mrcfile.write("</configuration>\n")
                #
                with open("%s/etc/hadoop/yarn-site.xml" % hdhome, "w") as yarnfile:
                  yarnfile.write("<configuration>\n")
                  yarnfile.write("  <property><name>yarn.resourcemanager.hostname</name>")
                  yarnfile.write("<value>%s-%d</value></property>\n" % (host_prefix, sparkIDs[0]))
                  yarnfile.write("  <property><name>yarn.nodemanager.hostname</name>")
                  yarnfile.write("<value>%s-%d</value></property>\n" % (host_prefix, sparkIDs[0]))
                  yarnfile.write("  <property><name>yarn.nodemanager.aux-services</name>")
                  yarnfile.write("<value>mapreduce_shuffle</value></property>\n")
                  yarnfile.write("</configuration>\n")
                #
                with open("%s/etc/hadoop/slaves" % hdhome, "w") as slavefile:
                  for item in sparkIDs[1:]:
                    slavefile.write("%s-%d.pd.infn.it\n" % (host_prefix, item))
                #
                if node_id == sparkIDs[0]:
                  check_call(runcmd % ("JAVA_HOME=/usr/lib/jvm/jre-1.8.0-openjdk %s/bin/hdfs namenode -format" % hdhome),
                       shell=True)
                #
                check_call("wget -O /tmp/spark-%s.tar.gz %s/spark-%s-bin-without-hadoop.tgz" % (spkver, repo_url, spkver), shell=True)
                check_call(runcmd % ("tar -C %s -zxf /tmp/spark-%s.tar.gz" % (workdir, spkver)), shell=True)
                os.rename("%s/spark-%s-bin-without-hadoop" % (workdir, spkver), "%s/spark" % workdir)
                #
                with open("%s/conf/spark-env.sh" % spkhome, "w") as spenvfile:
                  spenvfile.write("export SPARK_DIST_CLASSPATH=")
                  spenvfile.write("$(%s/bin/hadoop --config %s/etc/hadoop classpath)\n" % (hdhome, hdhome))
                  spenvfile.write("export SPARK_WORKER_PORT=%d\n" % spk_wport)
                #
                with open("%s/conf/slaves" % spkhome, "w") as slavefile:
                  for item in sparkIDs[1:]:
                    slavefile.write("%s-%d.pd.infn.it\n" % (host_prefix, item))
              #
              #####################################################################################
              # Nifi setup
              #####################################################################################
              #
              if "nifi" in node_type:
                #
                with open("/tmp/heat_multi_node.log", "a") as resfile:
                  resfile.write(" Installing NiFi\n")
                #
                nifiver = "1.7.1"
                #
                check_call("wget -O /tmp/nifi-%s.tar.gz %s/nifi-%s-bin.tar.gz" % (nifiver, repo_url, nifiver), shell=True)
                check_call(runcmd % ("tar -C %s -zxf /tmp/nifi-%s.tar.gz" % (workdir, nifiver)), shell=True)
                os.rename("%s/nifi-%s" % (workdir, nifiver), "%s/nifi" % workdir)
                #
                with open("%s/nifi/conf/nifi.properties" % workdir, "w") as nififile:
                  nififile.write("nifi.administrative.yield.duration=30 sec\n")
                  nififile.write("nifi.authorizer.configuration.file=%s/nifi/conf/authorizers.xml\n" % workdir)
                  nififile.write("nifi.bored.yield.duration=10 millis\n")
                  nififile.write("nifi.cluster.firewall.file=\n")
                  nififile.write("nifi.cluster.flow.election.max.candidates=\n")
                  nififile.write("nifi.cluster.flow.election.max.wait.time=5 mins\n")
                  nififile.write("nifi.cluster.is.node=false\n")
                  nififile.write("nifi.cluster.node.address=\n")
                  nififile.write("nifi.cluster.node.connection.timeout=5 sec\n")
                  nififile.write("nifi.cluster.node.event.history.size=25\n")
                  nififile.write("nifi.cluster.node.max.concurrent.requests=100\n")
                  nififile.write("nifi.cluster.node.protocol.max.threads=50\n")
                  nififile.write("nifi.cluster.node.protocol.port=\n")
                  nififile.write("nifi.cluster.node.protocol.threads=10\n")
                  nififile.write("nifi.cluster.node.read.timeout=5 sec\n")
                  nififile.write("nifi.cluster.protocol.heartbeat.interval=5 sec\n")
                  nififile.write("nifi.cluster.protocol.is.secure=true\n")
                  nififile.write("nifi.components.status.repository.buffer.size=1440\n")
                  nififile.write("nifi.components.status.repository.implementation=")
                  nififile.write("org.apache.nifi.controller.status.history.VolatileComponentStatusRepository\n")
                  nififile.write("nifi.components.status.snapshot.frequency=1 min\n")
                  nififile.write("nifi.content.claim.max.appendable.size=1 MB\n")
                  nififile.write("nifi.content.claim.max.flow.files=100\n")
                  nififile.write("nifi.content.repository.always.sync=false\n")
                  nififile.write("nifi.content.repository.archive.enabled=true\n")
                  nififile.write("nifi.content.repository.archive.max.retention.period=12 hours\n")
                  nififile.write("nifi.content.repository.archive.max.usage.percentage=50%\n")
                  nififile.write("nifi.content.repository.directory.default=%s/nifi/content_repository\n" % workdir)
                  nififile.write("nifi.content.repository.implementation=")
                  nififile.write("org.apache.nifi.controller.repository.FileSystemRepository\n")
                  nififile.write("nifi.content.viewer.url=../nifi-content-viewer/\n")
                  nififile.write("nifi.database.directory=./database_repository\n")
                  nififile.write("nifi.documentation.working.directory=%s/nifi/work/docs/components\n" % workdir)
                  nififile.write("nifi.flow.configuration.archive.dir=%s/nifi/conf/archive/\n" % workdir)
                  nififile.write("nifi.flow.configuration.archive.enabled=true\n")
                  nififile.write("nifi.flow.configuration.archive.max.count=\n")
                  nififile.write("nifi.flow.configuration.archive.max.storage=500 MB\n")
                  nififile.write("nifi.flow.configuration.archive.max.time=30 days\n")
                  nififile.write("nifi.flow.configuration.file=./conf/flow.xml.gz\n")
                  nififile.write("nifi.flowcontroller.autoResumeState=true\n")
                  nififile.write("nifi.flowcontroller.graceful.shutdown.period=10 sec\n")
                  nififile.write("nifi.flowfile.repository.always.sync=false\n")
                  nififile.write("nifi.flowfile.repository.checkpoint.interval=2 mins\n")
                  nififile.write("nifi.flowfile.repository.directory=%s/nifi/flowfile_repository\n" % workdir)
                  nififile.write("nifi.flowfile.repository.implementation=")
                  nififile.write("org.apache.nifi.controller.repository.WriteAheadFlowFileRepository\n")
                  nififile.write("nifi.flowfile.repository.partitions=256\n")
                  nififile.write("nifi.flowfile.repository.wal.implementation=")
                  nififile.write("org.apache.nifi.wali.SequentialAccessWriteAheadLog\n")
                  nififile.write("nifi.flowservice.writedelay.interval=500 ms\n")
                  nififile.write("nifi.h2.url.append=;LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE\n")
                  nififile.write("nifi.kerberos.krb5.file=\n")
                  nififile.write("nifi.kerberos.service.keytab.location=\n")
                  nififile.write("nifi.kerberos.service.principal=\n")
                  nififile.write("nifi.kerberos.spnego.authentication.expiration=12 hours\n")
                  nififile.write("nifi.kerberos.spnego.keytab.location=\n")
                  nififile.write("nifi.kerberos.spnego.principal=\n")
                  nififile.write("nifi.login.identity.provider.configuration.file=")
                  nififile.write("%s/nifi/conf/login-identity-providers.xml\n" % workdir)
                  nififile.write("nifi.nar.library.directory=%s/nifi/lib\n" % workdir)
                  nififile.write("nifi.nar.working.directory=%s/nifi/work/nar/\n" % workdir)
                  nififile.write("nifi.provenance.repository.always.sync=false\n")
                  nififile.write("nifi.provenance.repository.buffer.size=100000\n")
                  nififile.write("nifi.provenance.repository.compress.on.rollover=true\n")
                  nififile.write("nifi.provenance.repository.concurrent.merge.threads=2\n")
                  nififile.write("nifi.provenance.repository.debug.frequency=1_000_000\n")
                  nififile.write("nifi.provenance.repository.directory.default=")
                  nififile.write("%s/nifi/provenance_repository\n" % workdir)
                  nififile.write("nifi.provenance.repository.encryption.key=\n")
                  nififile.write("nifi.provenance.repository.encryption.key.id=\n")
                  nififile.write("nifi.provenance.repository.encryption.key.provider.implementation=\n")
                  nififile.write("nifi.provenance.repository.encryption.key.provider.location=\n")
                  nififile.write("nifi.provenance.repository.implementation=")
                  nififile.write("org.apache.nifi.provenance.PersistentProvenanceRepository\n")
                  nififile.write("nifi.provenance.repository.indexed.attributes=\n")
                  nififile.write("nifi.provenance.repository.indexed.fields=")
                  nififile.write("EventType, FlowFileUUID, Filename, ProcessorID, Relationship\n")
                  nififile.write("nifi.provenance.repository.index.shard.size=500 MB\n")
                  nififile.write("nifi.provenance.repository.index.threads=2\n")
                  nififile.write("nifi.provenance.repository.journal.count=16\n")
                  nififile.write("nifi.provenance.repository.max.attribute.length=65536\n")
                  nififile.write("nifi.provenance.repository.max.storage.size=1 GB\n")
                  nififile.write("nifi.provenance.repository.max.storage.time=24 hours\n")
                  nififile.write("nifi.provenance.repository.query.threads=2\n")
                  nififile.write("nifi.provenance.repository.rollover.size=100 MB\n")
                  nififile.write("nifi.provenance.repository.rollover.time=30 secs\n")
                  nififile.write("nifi.provenance.repository.warm.cache.frequency=1 hour\n")
                  nififile.write("nifi.queue.backpressure.count=10000\n")
                  nififile.write("nifi.queue.backpressure.size=1 GB\n")
                  nififile.write("nifi.queue.swap.threshold=20000\n")
                  nififile.write("nifi.remote.contents.cache.expiration=30 secs\n")
                  nififile.write("nifi.remote.input.host=\n")
                  nififile.write("nifi.remote.input.http.enabled=true\n")
                  nififile.write("nifi.remote.input.http.transaction.ttl=30 sec\n")
                  nififile.write("nifi.remote.input.secure=true\n")
                  nififile.write("nifi.remote.input.socket.port=\n")
                  nififile.write("nifi.security.keyPasswd=%s\n" % rootpwd)
                  nififile.write("nifi.security.keystore=%s/nifi/conf/service.jks\n" % workdir)
                  nififile.write("nifi.security.keystorePasswd=%s\n" % rootpwd)
                  nififile.write("nifi.security.keystoreType=JKS\n")
                  nififile.write("nifi.security.needClientAuth=false\n")
                  nififile.write("nifi.security.ocsp.responder.certificate=\n")
                  nififile.write("nifi.security.ocsp.responder.url=\n")
                  nififile.write("nifi.security.truststore=/etc/pki/java/cacerts\n")
                  nififile.write("nifi.security.truststorePasswd=changeit\n")
                  nififile.write("nifi.security.truststoreType=JKS\n")
                  nififile.write("nifi.security.user.authorizer=managed-authorizer\n")
                  nififile.write("nifi.security.user.knox.audiences=\n")
                  nififile.write("nifi.security.user.knox.cookieName=hadoop-jwt\n")
                  nififile.write("nifi.security.user.knox.publicKey=\n")
                  nififile.write("nifi.security.user.knox.url=\n")
                  nififile.write("nifi.security.user.login.identity.provider=\n")
                  nififile.write("nifi.security.user.oidc.client.id=%s\n" % oidc_id)
                  nififile.write("nifi.security.user.oidc.client.secret=%s\n" % oidc_sec)
                  nififile.write("nifi.security.user.oidc.connect.timeout=5 secs\n")
                  nififile.write("nifi.security.user.oidc.discovery.url=%s\n" % oidc_idp)
                  nififile.write("nifi.security.user.oidc.preferred.jwsalgorithm=\n")
                  nififile.write("nifi.security.user.oidc.read.timeout=5 secs\n")
                  nififile.write("nifi.sensitive.props.additional.keys=\n")
                  nififile.write("nifi.sensitive.props.algorithm=PBEWITHMD5AND256BITAES-CBC-OPENSSL\n")
                  nififile.write("nifi.sensitive.props.key=%s\n" % rootpwd)
                  nififile.write("nifi.sensitive.props.key.protected=\n")
                  nififile.write("nifi.sensitive.props.provider=BC\n")
                  nififile.write("nifi.state.management.configuration.file=")
                  nififile.write("%s/nifi/conf/state-management.xml\n" % workdir)
                  nififile.write("nifi.state.management.embedded.zookeeper.properties=")
                  nififile.write("%s/nifi/conf/zookeeper.properties\n" % workdir)
                  nififile.write("nifi.state.management.embedded.zookeeper.start=false\n")
                  nififile.write("nifi.state.management.provider.cluster=zk-provider\n")
                  nififile.write("nifi.state.management.provider.local=local-provider\n")
                  nififile.write("nifi.swap.in.period=5 sec\n")
                  nififile.write("nifi.swap.in.threads=1\n")
                  nififile.write("nifi.swap.manager.implementation=")
                  nififile.write("org.apache.nifi.controller.FileSystemSwapManager\n")
                  nififile.write("nifi.swap.out.period=5 sec\n")
                  nififile.write("nifi.swap.out.threads=4\n")
                  nififile.write("nifi.templates.directory=%s/nifi/conf/templates\n" % workdir)
                  nififile.write("nifi.ui.autorefresh.interval=30 sec\n")
                  nififile.write("nifi.ui.banner.text=\n")
                  nififile.write("nifi.variable.registry.properties=\n")
                  nififile.write("nifi.web.http.host=\n")
                  nififile.write("nifi.web.http.network.interface.default=\n")
                  nififile.write("nifi.web.http.port=\n")
                  nififile.write("nifi.web.https.host=0.0.0.0\n")
                  nififile.write("nifi.web.https.network.interface.default=\n")
                  nififile.write("nifi.web.https.port=8443\n")
                  nififile.write("nifi.web.jetty.threads=200\n")
                  nififile.write("nifi.web.jetty.working.directory=%s/nifi/work/jetty\n" % workdir)
                  nififile.write("nifi.web.max.header.size=16 KB\n")
                  nififile.write("nifi.web.proxy.context.path=\n")
                  nififile.write("nifi.web.proxy.host=\n")
                  nififile.write("nifi.web.war.directory=%s/nifi/lib\n" % workdir)
                  nififile.write("nifi.zookeeper.auth.type=\n")
                  nififile.write("nifi.zookeeper.connect.string=\n")
                  nififile.write("nifi.zookeeper.connect.timeout=3 secs\n")
                  nififile.write("nifi.zookeeper.kerberos.removeHostFromPrincipal=\n")
                  nififile.write("nifi.zookeeper.kerberos.removeRealmFromPrincipal=\n")
                  nififile.write("nifi.zookeeper.root.node=/nifi\n")
                  nififile.write("nifi.zookeeper.session.timeout=3 secs\n")
                #
                with open("%s/nifi/conf/authorizers.xml" % workdir, "w") as authfile:
                  authfile.write("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?>\n")
                  authfile.write("<authorizers>\n")
                  authfile.write("  <userGroupProvider>\n")
                  authfile.write("    <identifier>file-user-group-provider</identifier>\n")
                  authfile.write("    <class>org.apache.nifi.authorization.FileUserGroupProvider</class>\n")
                  authfile.write("    <property name=\"Users File\">./conf/users.xml</property>\n")
                  authfile.write("    <property name=\"Legacy Authorized Users File\"></property>\n")
                  authfile.write("    <property name=\"Legacy Authorized Users File\"></property>\n")
                  authfile.write("    <property name=\"Legacy Authorized Users File\"></property>\n")
                  authfile.write("    <property name=\"Legacy Authorized Users File\"></property>\n")
                  authfile.write("    <property name=\"Initial User Identity 1\">%s</property>\n" % admin_id)
                  authfile.write("  </userGroupProvider>\n")
                  authfile.write("  <accessPolicyProvider>\n")
                  authfile.write("    <identifier>file-access-policy-provider</identifier>\n")
                  authfile.write("    <class>org.apache.nifi.authorization.FileAccessPolicyProvider</class>\n")
                  authfile.write("    <property name=\"User Group Provider\">file-user-group-provider</property>\n")
                  authfile.write("    <property name=\"Authorizations File\">./conf/authorizations.xml</property>\n")
                  authfile.write("    <property name=\"Initial Admin Identity\">%s</property>\n" % admin_id)
                  authfile.write("    <property name=\"Legacy Authorized Users File\"></property>\n")
                  authfile.write("    <property name=\"Node Identity 1\"></property>\n")
                  authfile.write("  </accessPolicyProvider>\n")
                  authfile.write("  <authorizer>\n")
                  authfile.write("    <identifier>managed-authorizer</identifier>\n")
                  authfile.write("    <class>org.apache.nifi.authorization.StandardManagedAuthorizer</class>\n")
                  authfile.write("    <property name=\"Access Policy Provider\">file-access-policy-provider</property>\n")
                  authfile.write("   </authorizer>\n")
                  authfile.write("</authorizers>\n")
                #
                with open("%s/nifi/conf/hdfs-core-site.xml" % workdir, "w") as corefile:
                  corefile.write(corecfg)
                #
                with open("%s/nifi/conf/hdfs-site.xml" % workdir, "w") as hdfscfile:
                  hdfscfile.write(hdfscfg)
                #
              #
              #####################################################################################
              # ML setup
              #####################################################################################
              #
              if "ml" in node_type:
                #
                bdlver = "0.7.0"
                #
                with open("/tmp/heat_multi_node.log", "a") as resfile:
                  resfile.write(" Installing ML packages\n")
                #
                check_call("wget -O /tmp/bigdl-%s.tar.gz %s/bigdl-%s.tar.gz" % (bdlver, repo_url, bdlver), shell=True)
                check_call(runcmd % ("tar -C %s -zxf /tmp/bigdl-%s.tar.gz" % (workdir, bdlver)), shell=True)
                os.rename("%s/bigdl-%s" % (workdir, bdlver), "%s/bigdl" % workdir)
                #
                with open("/etc/profile.d/bigdl.sh", "w") as prfile:
                  prfile.write("export SPARK_HOME=%s/spark\n" % workdir)
                  prfile.write("export BIGDL_HOME=%s/bigdl\n" % workdir)
                  prfile.write("export PATH=$PATH:%s/bigdl/bin" % workdir)
                #
                extrapkgs = "python-devel python2-pip python2-py4j gcc cmake hdf5-devel"
                #extrapkgs += "root root-tmva python2-root"
                check_call("yum -q -y install %s" % extrapkgs, shell=True)
                #
                check_call([sys.executable, '-m', 'pip', 'install', '-q', '-U', 'pip'])
                #
                c_args = [sys.executable, '-m', 'pip', 'install', '-q', '-U',
                  'scikit-learn', 'scikit-image', 'pandas',
                  'seaborn', 'tables', 'lime', 'graphviz'
                ]
                # 'root-pandas', 'rootpy', 'root_numpy'
                check_call(c_args)
              #
              with open("/tmp/heat_multi_node.log", "a") as resfile:
                resfile.write(" Installation completed\n")
              #
            except:
              # TODO use signal or wc for stacktrace output
              etype, evalue, etraceback = sys.exc_info()
              if etraceback:
                with open("/tmp/heat_multi_node.log", "a") as tbfile:
                  traceback.print_exception(etype, evalue, etraceback, None, tbfile)
          params:
            <%host_pref%>: { get_param: host_prefix }
            <%host_name%>: { get_param: host_name }
            <%lan_ippre%>: { get_param: lan_net_ippref }
            <%node_id%>:   { get_param: node_id }
            <%kafka_ids%>: { get_param: kafka_ids }
            <%spark_ids%>: { get_param: spark_ids }
            <%ns_list%>:   { get_param: nameserver_list }
            <%node_type%>: { get_param: node_type }
            <%root_pwd%>:  { get_param: root_pw }
            <%admin_id%>:  { get_param: admin_id }
            <%oidc_id%>:   { get_param: oidc_client_id }
            <%oidc_sec%>:  { get_param: oidc_client_secret }
            <%oidc_idp%>:  { get_param: oidc_idp_url }



#
# Post-inst commands:
#   ssh-keyscan -t rsa1,rsa,dsa -f /etc/ssh/shosts.equiv >> /etc/ssh/ssh_known_hosts
#   systemctl restart sshd
#
#   systemctl start zookeeper
#   systemctl start kafka
#
# (bigdatausr on master) start-dfs.sh
# (bigdatausr on master) start-master.sh
# (bigdatausr on master) start-slaves.sh
#
# https://www.linode.com/docs/databases/hadoop/how-to-install-and-set-up-hadoop-cluster/
# https://pandas.pydata.org/
# http://www.numpy.org/
# https://github.com/cerndb/dist-keras
# https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-rdd-partitions.html
#
# latest package for spark-root in maven central: "org.diana-hep:spark-root_2.11:0.1.16"
# python call:
#   spark.read.format("org.dianahep.sparkroot.experimental").load("hdfs:///path/to/file.root")
#
# Simple tests
#   kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic mytopic
#   kafka-console-producer.sh --broker-list localhost:9092 --topic mytopic
#   kafka-topics.sh --list --zookeeper localhost:2181
#   kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic mytopic
#
# Nifi post-inst operations:
# Install the keystore in /opt/bigdata/nifi/conf/service.jks:
#   the keystore pwd MUST BE equal to the root_pwd
#   the owner MUST BE bigdatausr.bigdatausr
# Run "/opt/bigdata/nifi/bin/nifi.sh start" as bigdatausr
#







