Merge pull request #45 from jaketf/demo

Demoprep

Author: Jacob Ferriero

Makefile

@@ -24,17 +24,17 @@ help: ## Prints help for targets with comments
 
 .PHONY: test
 test: ## Test if all files are properly formatted
-	@$$SHELL ./helpers/check_format.sh && python3 -m flake8 --max-line-length=80 && ./helpers/run_tests.sh
+	@$$SHELL ./helpers/check_format.sh && python3 -m flake8 --max-line-length=100 && ./helpers/run_tests.sh
 
 .PHONY: precommit
 precommit: ## Test if all files are properly formatted
-	@$$SHELL ./helpers/check_format.sh && python3 -m flake8 --max-line-length=80 && ./helpers/run_relevant_cloudbuilds.sh precommit_cloudbuild.yaml
+	@$$SHELL ./helpers/check_format.sh && python3 -m flake8 --max-line-length=100 && ./helpers/run_relevant_cloudbuilds.sh precommit_cloudbuild.yaml
 
 .PHONY: push_ci_image
 push_ci_image:
-	@cd ci && gcloud builds submit --project=datapipelines-ci --tag gcr.io/datapipelines-ci/make .
+	@cd ci && gcloud builds submit --project=datapipelines-ci-282719 --tag gcr.io/datapipelines-ci-282719/make .
 
 .PHONY: push_deploydags_image
 push_deploydags_image:
-	@cd composer/cloudbuild/go/dagsdeployer && gcloud builds submit --project=datapipelines-ci --tag gcr.io/datapipelines-ci/deploydags .
+	@cd composer/cloudbuild/go/dagsdeployer && gcloud builds submit --project=datapipelines-ci-282719 --tag gcr.io/datapipelines-ci-282719/deploydags .
 

cd/prod.yaml

@@ -139,7 +139,7 @@ steps:
           'deploy-wordcount-jar'
           ]
   args: [
-          '-dagList=./config/ci_dags.txt',
+          '-dagList=./config/running_dags.txt',
           '-dagsFolder=./dags',
           '-project=${PROJECT_ID}',
           '-region=${_COMPOSER_REGION}',

ci/Dockerfile

@@ -5,11 +5,13 @@ FROM python:buster
 # install core tools
 RUN apt-get update && apt-get install -y build-essential
 
+RUN curl -sSL https://sdk.cloud.google.com | bash
+
 # install shellcheck
 RUN apt-get install shellcheck
 
 # install yapf
-RUN pip3 install yapf flake8 pytest
+RUN pip3 install yapf flake8 pytest apache-airflow[gcp]==1.10.6
 
 # install golang (+gofmt)
 RUN apt-get install -y golang

cloudbuild.yaml

@@ -187,7 +187,7 @@ steps:
   dir: './composer/cloudbuild/go/dagsdeployer'
   args: [
           'build', 
-          '-t', 'gcr.io/${_ARTIFACTS_PROJECT_ID}/deploydags:latest',
+          '-t', 'gcr.io/${PROJECT_ID}/deploydags:latest',
           '--cache-from', 'gcr.io/${_ARTIFACTS_PROJECT_ID}/deploydags:latest',
           '.'
         ]
@@ -201,7 +201,7 @@ steps:
           'clean-up-data-dir-dags',
           ]
   args: [
-          '-dagList=./config/ci_dags.txt',
+          '-dagList=./config/running_dags.txt',
           '-dagsFolder=./dags',
           '-project=${PROJECT_ID}',
           '-region=${_COMPOSER_REGION}',

composer/cloudbuild/bin/run_tests.sh

@@ -109,7 +109,7 @@ function install_airflow() {
 # $1 relative path to directory containing bigquery sql.
 # $2 relative path to JSON file contianing Airflow Variables.
 main() {
-  setup_local_airflow "$1" "$2" "$3" 
+  setup_local_airflow "$1" "$2" "$3"
   run_tests
   TEST_STATUS=$?
   clean_up

composer/cloudbuild/go/dagsdeployer/internal/composerdeployer/composer_ops.go

@@ -195,7 +195,6 @@ func readCommentScrubbedLines(path string) ([]string, error) {
 			lines = append(lines, candidate)
 		}
 	}
-	log.Printf("scrubbed lines: %#v", lines)
 
 	return lines, scanner.Err()
 }
@@ -212,15 +211,18 @@ func FindDagFilesInLocalTree(dagsRoot string, dagNames map[string]bool) (map[str
 	// This should map a dir to the ignore patterns in it's airflow ignore if relevant
 	// this allows us to easily identify the patterns relevant to this dir and it's parents, grandparents, etc.
 	airflowignoreTree := make(map[string][]string)
-	files, err := ioutil.ReadDir(dagsRoot)
+	_, err := ioutil.ReadDir(dagsRoot)
 	if err != nil {
 	  return matches, fmt.Errorf("error reading dagRoot: %v. %v", dagsRoot, err)
 	}
-	log.Printf("found files at dagRott: %v", files)
 	filepath.Walk(dagsRoot, func(path string, info os.FileInfo, err error) error {
 		dagID := strings.TrimSuffix(info.Name(), ".py")
 		relPath, err := filepath.Rel(dagsRoot, path)
 
+    if info == nil {
+			 dur, _ := time.ParseDuration("5s")
+       time.Sleep(dur)
+    }
 		// resepect .airflowignore
 		if info.Name() == ".airflowignore" {
 			log.Printf("found %v, adding to airflowignoreTree", path)
@@ -325,7 +327,6 @@ func FindDagFilesInLocalTree(dagsRoot string, dagNames map[string]bool) (map[str
 				}
 			}
 			if !alreadyMatched {
-				log.Printf("new match for %v: %v", dagID, relPath)
 				matches[dagID] = append(matches[dagID], relPath)
 			}
 		}
@@ -461,8 +462,6 @@ func (c *ComposerEnv) stopDag(dag string, relPath string, pauseOnly bool, wg *sy
 	if err != nil {
 		return fmt.Errorf("error pausing dag %v: %v", dag, string(out))
 	}
-	log.Printf("pauseOnly: %#v", pauseOnly)
-	log.Printf("!pauseOnly: %#v", !pauseOnly)
 	if !pauseOnly {
 		log.Printf("parsing gcs url %v", c.DagBucketPrefix)
 		gcs, err := url.Parse(c.DagBucketPrefix)

composer/cloudbuild/go/dagsdeployer/internal/gcshasher/gcs_hash.go

@@ -44,7 +44,6 @@ func parseGcsPath(gcsPath string) (bucket string, path string, err error) {
 }
 func gcsMD5(gcsPath string) ([]byte, error) {
 	bktName, path, err := parseGcsPath(gcsPath)
-	log.Printf("taking hash of bucket: %s, path: %s", bktName, path)
 	if err != nil {
 		log.Fatalf("%s", err)
 	}

composer/config/AirflowVariables.json

@@ -6,5 +6,9 @@
     "gcs_input_bucket": "${INPUT_BUCKET}",
     "gcs_ref_bucket": "${REF_BUCKET}",
     "gcs_output_bucket": "${RESULT_BUCKET}",
-    "dataflow_staging_bucket": "${DATAFLOW_STAGING_BUCKET}"
+    "dataflow_staging_bucket": "${DATAFLOW_STAGING_BUCKET}",
+    "dataproc_bucket": "${DATFLOW_STAGING_BUCKET}",
+    "gce_zone": "${COMPOSER_REGION}-a",
+    "gcs_bucket": "spark_bucket",
+    "bq_output_table": "${GCP_PROJECT_ID}.nyc_taxi.avg_speed"
 }

composer/config/ci_dags.txt

@@ -0,0 +1,22 @@
+wordcount_dag
+tutorial
+ephemeral_dataproc_spark_dag
+bash_operator
+branch_operator
+branch_python_dop_operator_3
+complex
+http_operator
+kubernetes_executor_config
+latest_only
+latest_only_with_trigger
+nested_branch_dag
+passing_params_via_test_command
+pig_operator
+python_operator
+short_circuit_operator
+skip_dag
+subdag_operator
+trigger_controller_dag
+trigger_target_dag
+tutorial
+xcom