Merge pull request #7 from teamclairvoyant/feature

Feature

Author: rssanders3

.gitignore

@@ -1,2 +1,6 @@
 target/
 *.swp
+.gitignore
+*.DS_Store
+.idea/*
+*.iml

README.md

@@ -5,9 +5,12 @@ This repository allows you to install [Apache Airflow](https://airflow.apache.or
 ## Requirements
 - A supported operating system.
 - MySQL or PostgreSQL database in which to store Airflow metadata.
+- RabbitMQ
+- [Airflow Parcel](https://github.com/teamclairvoyant/apache-airflow-parcels)
 
 ### Currently Supported Versions of Airflow
-- Airflow 1.10
+- Airflow 1.9.0
+- Airflow 1.10.3
 
 ### Currently Supported Operating Systems
 - CentOS/RHEL 6 & 7
@@ -56,6 +59,7 @@ create_postgresql_dbs-airflow.sh --host <host_name> --user <username> --password
 ```
 
 ## Roles
+
 There are six roles available for deployment:
 
 1. Webserver
@@ -132,19 +136,16 @@ Update the `version` file before running `make dist` if creating a new release.
 2. Only 'airflow.contrib.auth.backends.password_auth' mechanism is supported for Airflow user authentication.
 
 ## Future work:
-1. Test Database connection.
-2. Add the support for more Airflow user authentication methods.
+1. Build RabbitMQ parcel.
+2. Test Database connection.
+3. Add the support for more Airflow user authentication methods.
 
 ## Known Errors:
 
 ### Markup already exists Error:
 
 Upon many deployments, you may face an error called 'Markup file already exists' while trying to stop a role and the process never stops. In that case, stop the process using the "Abort" command and navigate to `/var/run/cloudera-scm-agent/process` and delete all the `GracefulRoleStopRunner` directories.
 
-### Lag in DAG Execution:
-
-Occasionally, we experienced some delay in DAG execution. We are working to fix this.
-
 ## Resources:
 1. https://github.com/teamclairvoyant/apache-airflow-parcels
 2. https://github.com/cloudera/cm_ext/wiki/The-Structure-of-a-CSD

src/descriptor/service.sdl

@@ -14,7 +14,7 @@
 {
   "name" : "AIRFLOW",
   "label" : "Airflow",
-  "description" : "Airflow is a platform to programmatically author, schedule and monitor workflows.  <span class=\"error\">Before adding this service, ensure that the Airflow parcel has been activated.</span>",
+  "description" : "Airflow is a platform to programmatically author, schedule and monitor workflows.  <span class=\"error\">Before adding this service, please ensure that the Airflow parcel is activated and RabbitMQ is installed</span>",
   "version" : "{{ version }}",
   "runAs" : {
     "user" : "root",
@@ -136,6 +136,22 @@
       "type" : "string",
       "default" : ""
     },
+    {
+      "name" : "Airflow.password",
+      "label" : "Airflow Password",
+      "description" : "Password of the Airflow user.",
+      "type" : "password",
+      "default" : ""
+    },
+    {
+      "name" : "broker_url",
+      "label" : "Broker URL",
+      "description" : "Broker URL value for Celery ",
+      "type" : "string",
+      "default" : "sqla+mysql://username:password@host:3306/dbname",	
+      "configurableInWizard": true,
+      "required" : "true"
+    },
     {
       "name" : "celery_broker_port",
       "label" : "Celery Broker Port",
@@ -483,6 +499,40 @@
       "program" : "scripts/control.sh",
       "args" : [ "client" ],
       "environmentVariables" : {
+        "dbType" : "${dbType}",
+        "dbHost" : "${dbHost}",
+        "dbUser" : "${dbUser}",
+        "dbName" : "${dbName}",
+        "dbPort" : "${dbPort}",
+        "dbPass" : "${dbPass}",
+	"broker_url": "${broker_url}",
+        "authenticate" : "${Airflow.authenticate}",
+        "AIRFLOW_USER" : "${Airflow.username}",
+        "AIRFLOW_PASS" : "${Airflow.password}",
+        "AIRFLOW_EMAIL" : "${Airflow.email}",
+        "airflow_home" : "${airflow_home}",
+        "AIRFLOW_HOME" : "${airflow_home}",
+        "dags_folder" : "${dags_folder}",
+        "base_log_folder" : "${base_log_folder}",
+        "remote_base_log_folder" : "${remote_base_log_folder}",
+        "remote_log_conn_id" : "${remote_log_conn_id}",
+        "encrypt_s3_logs" : "${encrypt_s3_logs}",
+        "executor" : "${executor}",
+        "sql_alchemy_pool_size" : "${sql_alchemy_pool_size}",
+        "sql_alchemy_pool_recycle" : "${sql_alchemy_pool_recycle}",
+        "parallelism" : "${parallelism}",
+        "dag_concurrency" : "${dag_concurrency}",
+        "dags_are_paused_at_creation" : "${dags_are_paused_at_creation}",
+        "non_pooled_task_slot_count" :  "${non_pooled_task_slot_count}",
+        "max_active_runs_per_dag" : "${max_active_runs_per_dag}",
+        "default_owner" : "${default_owner}",
+        "web_server_port" : "${web_server_port}",
+        "security" : "${security}",
+        "keytab" : "${keytab}",
+        "reinit_frequency" : "${reinit_frequency}",
+        "principal" : "${principal}",
+        "expose_config" : "${expose_config}",
+        "load_examples" : "${load_examples}"
         "AIRFLOW_CONFIG" : "/etc/airflow/conf/airflow.cfg",
         "DB_PASS" : "${database_password}",
         "CELERY_BROKER_PASS" : "${celery_broker_password}"
@@ -811,6 +861,11 @@
           }
         ]
       },
+      "stopRunner" : {
+         "runner" : {
+             "program" : "scripts/stop_airflow_scheduler.sh"
+        }
+      },
       "parameters" : [
         {
           "name" : "job_heartbeat_sec",

src/scripts/stop_airflow_scheduler.sh

@@ -0,0 +1,99 @@
+#!/bin/bash
+
+
+# Creating the AIRFLOW_HOME directory
+if [ ! -d ${airflow_home} ]; then
+    mkdir -p ${airflow_home}
+    chown -R airflow:airflow ${airflow_home}
+fi
+
+# Creating the Airflow log directory
+if [ ! -d ${base_log_folder} ]; then
+    mkdir -p ${base_log_folder}
+    chown -R airflow:airflow ${base_log_folder}
+fi
+
+# Creating dags folder
+if [ ! -d ${dags_folder} ]; then
+    mkdir -p ${dags_folder}
+    chown -R airflow:airflow ${dags_folder}
+    chmod 775 ${dags_folder}
+fi
+
+# Initializing the Airflow database if not existed
+if [ ! -f ${airflow_home}/airflow.cfg ]; then
+    sudo -Eu airflow bash -c 'export PATH=${AIRFLOW_DIR}/usr/bin:$PATH;
+    exec airflow initdb'
+fi
+
+# Building connection string
+if [ "$dbType" == "mysql" ]; then
+    sql_alchemy_conn="mysql://${dbUser}:${dbPass}@${dbHost}:${dbPort}/${dbName}"
+elif [ "$dbType" == "postgresql" ]; then
+    sql_alchemy_conn="postgresql+psycopg2://${dbUser}:${dbPass}@${dbHost}:${dbPort}/${dbName}"
+fi
+
+echo $sql_alchemy_conn
+   
+# Updating airflow.cfg
+
+#broker_url="amqp://${RABBITMQ_USER}:${RABBITMQ_PASS}@${RABBITMQ_HOST}:${RABBITMQ_PORT}/"
+celery_result_backend="db+${sql_alchemy_conn}"
+
+PWCMD='< /dev/urandom tr -dc A-Za-z0-9 | head -c 20;echo'
+secret_key=`eval $PWCMD`
+
+function replace {
+    sed -i "s#${1}#${2}#g" $3
+}
+
+function remove_line {
+    sed -i "/${1}/d" $2
+}
+
+echo "Updating airflow.cfg..."
+auth_backend_flag=false
+auth_backend=airflow.contrib.auth.backends.password_auth
+
+# Updating airflow.cfg
+while read line; do
+    key=$(cut -d "=" -f1 <<< "$line" | xargs)
+    value=$(cut -d "=" -f2 <<< "$line")
+    if [[ ${!key} ]]; then
+        if [[ "$key" != "#" ]]; then
+            if [[ "$key" == "authenticate" ]]; then
+                replace "$key =${value}" "$key = ${!key}\nauth_backend = airflow.contrib.auth.backends.password_auth" ${airflow_home}/airflow.cfg
+            elif [[ "$key" == "auth_backend" ]]; then
+                if [[ "$auth_backend_flag" = true ]]; then
+                    sed -i "/auth_backend = airflow.contrib.auth.backends.password_auth/d" ${airflow_home}/airflow.cfg
+                fi
+                auth_backend_flag=true
+            else
+                replace "$key =${value}" "$key = ${!key}" ${airflow_home}/airflow.cfg
+            fi
+        fi
+    fi
+done < ${airflow_home}/airflow.cfg
+
+
+echo "Creating Airflow user..."
+# Creating Airflow User
+
+if [[ ! -z "$AIRFLOW_USER" ]];
+then 
+    ${AIRFLOW_DIR}/usr/bin/python2.7 ../scripts/mkuser.py $AIRFLOW_USER $AIRFLOW_EMAIL $AIRFLOW_PASS
+fi
+
+
+echo "Updating AIRFLOW_HOME in binaries..."
+# Creating the airflow binary
+
+sed -i "1s+.*+export AIRFLOW_HOME=${AIRFLOW_HOME}\n+" ${AIRFLOW_DIR}/bin/airflow.sh
+chmod 777  ${AIRFLOW_DIR}/bin/airflow.sh
+
+echo "export PYTHONPATH=${PYTHONPATH}" > /usr/bin/airflow-mkuser
+echo "export PATH=${AIRFLOW_DIR}/bin:\$PATH" >> /usr/bin/airflow-mkuser
+echo "${AIRFLOW_DIR}/bin/mkuser.sh \$@" >> /usr/bin/airflow-mkuser
+
+
+echo "update.cfg script ends"