Feb/pipeline(unzip, pipeline log for error and success) (#6678)

* Zip file * Zip file and migration done status Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * PR Changes * PR Changes Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * PR Commect fixed Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * Code smell Fixed Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * Update utility.go * Unzip Destination change Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * Some fixes Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * Changes Issues and PR comments Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * Error log modified Signed-off-by: Pappu Kumar <pappu.kumar@progress.com> * Code smell fixed Signed-off-by: Pappu Kumar <pappu.kumar@progress.com>
chef · Mar 29, 2022 · 24e53fe · 24e53fe
1 parent bb4d0a6
commit 24e53fe
Show file tree

Hide file tree

Showing 7 changed files with 168 additions and 178 deletions.
diff --git a/components/infra-proxy-service/migrations/migrations.go b/components/infra-proxy-service/migrations/migrations.go
@@ -8,11 +8,12 @@ import (
 	"os"
 	"path"
 
+	"github.com/chef/automate/components/infra-proxy-service/pipeline"
+
 	"github.com/chef/automate/api/interservice/infra_proxy/migrations/request"
 	"github.com/chef/automate/api/interservice/infra_proxy/migrations/response"
 	"github.com/chef/automate/api/interservice/infra_proxy/migrations/service"
 	"github.com/chef/automate/components/infra-proxy-service/constants"
-	"github.com/chef/automate/components/infra-proxy-service/migrations/pipeline"
 	pipeline_model "github.com/chef/automate/components/infra-proxy-service/pipeline"
 
 	"github.com/chef/automate/components/infra-proxy-service/validation"

diff --git a/components/infra-proxy-service/migrations/pipeline/models.go b/components/infra-proxy-service/migrations/pipeline/models.go
diff --git a/components/infra-proxy-service/migrations/pipeline/phaseonemigration.go b/components/infra-proxy-service/migrations/pipeline/phaseonemigration.go
@@ -3,10 +3,20 @@ package pipeline
 import (
 	"context"
 	"fmt"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/chef/automate/components/infra-proxy-service/pipeline"
+	"github.com/chef/automate/components/infra-proxy-service/storage"
+)
+
+var (
+	Storage storage.Storage
+	Mig     storage.MigrationStorage
 )
 
 type PipelineData struct {
-	Result Result
+	Result pipeline.Result
 	Done   chan<- error
 	Ctx    context.Context
 }
@@ -189,7 +199,7 @@ func adminUsers(result <-chan PipelineData) <-chan PipelineData {
 
 func migrationPipeline(source <-chan PipelineData, pipes ...PhaseOnePipelineProcessor) {
 	fmt.Println("Pipeline started...")
-
+	status := make(chan string)
 	go func() {
 		for _, pipe := range pipes {
 			source = pipe(source)
@@ -198,7 +208,9 @@ func migrationPipeline(source <-chan PipelineData, pipes ...PhaseOnePipelineProc
 		for s := range source {
 			s.Done <- nil
 		}
+		status <- "Done"
 	}()
+	<-status
 }
 
 func SetupPhaseOnePipeline() PhaseOnePipleine {
@@ -214,7 +226,8 @@ func SetupPhaseOnePipeline() PhaseOnePipleine {
 	return PhaseOnePipleine{in: c}
 }
 
-func (p *PhaseOnePipleine) Run(result Result) {
+func (p *PhaseOnePipleine) Run(result pipeline.Result) {
+	status := make(chan string)
 	go func() {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
@@ -224,8 +237,25 @@ func (p *PhaseOnePipleine) Run(result Result) {
 		}
 		err := <-done
 		if err != nil {
-			fmt.Println("received error")
+			MigrationError(err, Mig, ctx, result.Meta.MigrationID, result.Meta.ServerID)
+			log.Errorf("Phase one pipeline received error for migration %s: %s", result.Meta.MigrationID, err)
 		}
-		fmt.Println("received done")
+		log.Println("received done")
+		status <- "Done"
 	}()
+	<-status
+}
+
+func MigrationError(err error, st storage.MigrationStorage, ctx context.Context, migrationId, serviceId string) {
+	_, err = st.FailedMigration(ctx, migrationId, serviceId, err.Error(), 0, 0, 0)
+	if err != nil {
+		log.Errorf("received error while updating for migration id %s: %s", migrationId, err)
+	}
+}
+
+func MigrationSuccess(st storage.MigrationStorage, ctx context.Context, migrationId, serviceId string) {
+	_, err := st.CompleteMigration(ctx, migrationId, serviceId, 0, 0, 0)
+	if err != nil {
+		log.Errorf("received error while updating for migration id %s: %s", migrationId, err)
+	}
 }
diff --git a/components/infra-proxy-service/migrations/pipeline/phasetwomigration.go b/components/infra-proxy-service/migrations/pipeline/phasetwomigration.go
@@ -3,6 +3,10 @@ package pipeline
 import (
 	"context"
 	"fmt"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/chef/automate/components/infra-proxy-service/pipeline"
 )
 
 type PhaseTwoPipleine struct {
@@ -143,7 +147,7 @@ func populateMembersPolicy(result <-chan PipelineData) <-chan PipelineData {
 
 func migrationTwoPipeline(source <-chan PipelineData, pipes ...PhaseTwoPipelineProcessor) {
 	fmt.Println("Pipeline started...")
-
+	status := make(chan string)
 	go func() {
 		for _, pipe := range pipes {
 			source = pipe(source)
@@ -152,7 +156,9 @@ func migrationTwoPipeline(source <-chan PipelineData, pipes ...PhaseTwoPipelineP
 		for s := range source {
 			s.Done <- nil
 		}
+		status <- "Done"
 	}()
+	<-status
 }
 
 func SetupPhaseTwoPipeline() PhaseTwoPipleine {
@@ -167,7 +173,7 @@ func SetupPhaseTwoPipeline() PhaseTwoPipleine {
 	return PhaseTwoPipleine{in: c}
 }
 
-func (p *PhaseTwoPipleine) Run(result Result) {
+func (p *PhaseTwoPipleine) Run(result pipeline.Result) {
 	status := make(chan string)
 	go func() {
 		ctx, cancel := context.WithCancel(context.Background())
@@ -178,9 +184,11 @@ func (p *PhaseTwoPipleine) Run(result Result) {
 		}
 		err := <-done
 		if err != nil {
-			fmt.Println("received error")
+			MigrationError(err, Mig, ctx, result.Meta.MigrationID, result.Meta.ServerID)
+			log.Errorf("Phase two pipeline received error for migration %s: %s", result.Meta.MigrationID, err)
 		}
-		fmt.Println("received done")
+		MigrationSuccess(Mig, ctx, result.Meta.MigrationID, result.Meta.ServerID)
+		log.Println("received done")
 		status <- "Done"
 	}()
 	<-status