[incubator-devlake] branch main updated: feat: dbt add download packages functions and logging functions

[wa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

warren pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-devlake.git


The following commit(s) were added to refs/heads/main by this push:
     new acd9bbc3 feat: dbt add download packages functions and logging functions
acd9bbc3 is described below

commit acd9bbc324b26a89a64be8fce8c735bd2816a240
Author: abeizn <zi...@merico.dev>
AuthorDate: Thu Sep 1 15:01:30 2022 +0800

    feat: dbt add download packages functions and logging functions
---
 Dockerfile                     |  2 +-
 plugins/dbt/tasks/convertor.go | 23 ++++++++++++++++++-----
 2 files changed, 19 insertions(+), 6 deletions(-)

diff --git a/Dockerfile b/Dockerfile
index 81250fbe..b97ee46d 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -37,7 +37,7 @@ ENV GOBIN=/app/bin
 
 RUN make clean && make all
 
-FROM --platform=linux/amd64 mericodev/alpine-dbt-mysql:0.0.1
+FROM --platform=linux/amd64 mericodev/alpine-dbt:0.0.1
 
 EXPOSE 8080
 
diff --git a/plugins/dbt/tasks/convertor.go b/plugins/dbt/tasks/convertor.go
index 2ada7c9b..c65291ef 100644
--- a/plugins/dbt/tasks/convertor.go
+++ b/plugins/dbt/tasks/convertor.go
@@ -33,6 +33,7 @@ import (
 )
 
 func DbtConverter(taskCtx core.SubTaskContext) error {
+	log := taskCtx.GetLogger()
 	taskCtx.SetProgress(0, -1)
 	data := taskCtx.GetData().(*DbtTaskData)
 	models := data.Options.SelectedModels
@@ -101,6 +102,15 @@ func DbtConverter(taskCtx core.SubTaskContext) error {
 		return err
 	}
 
+	_, err = os.Stat("packages.yml")
+	if err == nil {
+		cmd := exec.Command("dbt", "deps")
+		err = cmd.Start()
+		if err != nil {
+			return err
+		}
+	}
+
 	dbtExecParams := []string{"dbt", "run", "--profiles-dir", projectPath}
 	if projectVars != nil {
 		jsonProjectVars, err := json.Marshal(projectVars)
@@ -113,18 +123,16 @@ func DbtConverter(taskCtx core.SubTaskContext) error {
 	dbtExecParams = append(dbtExecParams, "--select")
 	dbtExecParams = append(dbtExecParams, models...)
 	cmd := exec.Command(dbtExecParams[0], (dbtExecParams[1:])...)
-	stdout, err := cmd.StdoutPipe()
-	if err != nil {
-		return err
-	}
+	log.Info("dbt run script: ", cmd)
+	stdout, _ := cmd.StdoutPipe()
 	err = cmd.Start()
 	if err != nil {
 		return err
 	}
-
 	scanner := bufio.NewScanner(stdout)
 	for scanner.Scan() {
 		line := scanner.Text()
+		log.Info(line)
 		if strings.Contains(line, "of") && strings.Contains(line, "OK") {
 			taskCtx.IncProgress(1)
 		}
@@ -133,6 +141,11 @@ func DbtConverter(taskCtx core.SubTaskContext) error {
 		return err
 	}
 
+	cmd.Wait()
+	if !cmd.ProcessState.Success() {
+		log.Error("dbt run task error, please check!!!")
+	}
+
 	return nil
 }