build handlers: split buld step in two phases: submit and run

pkg/apis/camel/v1alpha1/integration_types.go

@@ -131,8 +131,10 @@ const (
 	IntegrationPhaseWaitingForPlatform IntegrationPhase = "Waiting For Platform"
 	// IntegrationPhaseBuildingContext --
 	IntegrationPhaseBuildingContext IntegrationPhase = "Building Context"
-	// IntegrationPhaseBuildingImage --
-	IntegrationPhaseBuildingImage IntegrationPhase = "Building Image"
+	// IntegrationPhaseBuildImageSubmitted --
+	IntegrationPhaseBuildImageSubmitted IntegrationPhase = "Build Image Submitted"
+	// IntegrationPhaseBuildImageRunning --
+	IntegrationPhaseBuildImageRunning IntegrationPhase = "Build Image Running"
 	// IntegrationPhaseDeploying --
 	IntegrationPhaseDeploying IntegrationPhase = "Deploying"
 	// IntegrationPhaseRunning --

pkg/apis/camel/v1alpha1/integrationcontext_types.go

@@ -64,8 +64,10 @@ const (
 	// IntegrationContextTypeExternal --
 	IntegrationContextTypeExternal = "external"
 
-	// IntegrationContextPhaseBuilding --
-	IntegrationContextPhaseBuilding IntegrationContextPhase = "Building"
+	// IntegrationContextPhaseBuildSubmitted --
+	IntegrationContextPhaseBuildSubmitted IntegrationContextPhase = "Build Submitted"
+	// IntegrationContextPhaseBuildRunning --
+	IntegrationContextPhaseBuildRunning IntegrationContextPhase = "Build Running"
 	// IntegrationContextPhaseReady --
 	IntegrationContextPhaseReady IntegrationContextPhase = "Ready"
 	// IntegrationContextPhaseError --

pkg/builder/builder.go

@@ -41,7 +41,7 @@ import (
 // ********************************
 
 type buildTask struct {
-	handler func(Result)
+	handler func(*Result)
 	request Request
 }
 
@@ -78,22 +78,26 @@ func (b *defaultBuilder) IsBuilding(object v1.ObjectMeta) bool {
 }
 
 // Submit --
-func (b *defaultBuilder) Submit(request Request, handler func(Result)) {
+func (b *defaultBuilder) Submit(request Request, handler func(*Result)) {
 	if atomic.CompareAndSwapInt32(&b.running, 0, 1) {
 		go b.loop()
 	}
 
 	result, present := b.request.Load(request.Meta.Name)
 	if !present || result == nil {
-		result = Result{
+		r := Result{
 			Builder: b,
 			Request: request,
 			Status:  StatusSubmitted,
 		}
 
 		b.log.Infof("submitting request: %+v", request)
 
-		b.request.Store(request.Meta.Name, result)
+		if handler != nil {
+			handler(&r)
+		}
+
+		b.request.Store(request.Meta.Name, r)
 		b.tasks <- buildTask{handler: handler, request: request}
 	}
 }
@@ -127,7 +131,7 @@ func (b *defaultBuilder) loop() {
 	}
 }
 
-func (b *defaultBuilder) process(request Request, handler func(Result)) {
+func (b *defaultBuilder) process(request Request, handler func(*Result)) {
 	result, present := b.request.Load(request.Meta.Name)
 	if !present || result == nil {
 		b.log.Panicf("no info found for: %+v", request.Meta.Name)
@@ -139,7 +143,7 @@ func (b *defaultBuilder) process(request Request, handler func(Result)) {
 	r.Task.StartedAt = time.Now()
 
 	if handler != nil {
-		handler(r)
+		handler(&r)
 	}
 
 	// create tmp path
@@ -189,7 +193,7 @@ func (b *defaultBuilder) process(request Request, handler func(Result)) {
 
 	if r.Status == StatusError {
 		if handler != nil {
-			handler(r)
+			handler(&r)
 		}
 
 		return
@@ -202,13 +206,15 @@ func (b *defaultBuilder) process(request Request, handler func(Result)) {
 
 	b.log.Infof("steps: %v", request.Steps)
 	for _, step := range request.Steps {
-		if c.Error != nil {
+		if c.Error != nil || r.Status == StatusInterrupted {
 			break
 		}
 
 		select {
 		case <-b.interrupt:
 			c.Error = errors.New("build canceled")
+		case <-request.C.Done():
+			r.Status = StatusInterrupted
 		default:
 			l := b.log.WithFields(logrus.Fields{
 				"step":    step.ID(),
@@ -229,33 +235,38 @@ func (b *defaultBuilder) process(request Request, handler func(Result)) {
 		}
 	}
 
-	r.Status = StatusCompleted
-	r.BaseImage = c.BaseImage
-	r.Image = c.Image
-	r.PublicImage = c.PublicImage
-	r.Error = c.Error
 	r.Task.CompletedAt = time.Now()
 
-	if r.Error != nil {
-		r.Status = StatusError
-	}
+	if r.Status != StatusInterrupted {
+		r.Status = StatusCompleted
+		r.BaseImage = c.BaseImage
+		r.Image = c.Image
+		r.PublicImage = c.PublicImage
+		r.Error = c.Error
 
-	r.Artifacts = make([]v1alpha1.Artifact, 0, len(c.Artifacts))
-	r.Artifacts = append(r.Artifacts, c.Artifacts...)
+		if r.Error != nil {
+			r.Status = StatusError
+		}
+
+		r.Artifacts = make([]v1alpha1.Artifact, 0, len(c.Artifacts))
+		r.Artifacts = append(r.Artifacts, c.Artifacts...)
+
+		b.log.Infof("build request %s executed in %f seconds", request.Meta.Name, r.Task.Elapsed().Seconds())
+		b.log.Infof("dependencies: %s", request.Dependencies)
+		b.log.Infof("artifacts: %s", ArtifactIDs(c.Artifacts))
+		b.log.Infof("artifacts selected: %s", ArtifactIDs(c.SelectedArtifacts))
+		b.log.Infof("requested image: %s", request.Image)
+		b.log.Infof("base image: %s", c.BaseImage)
+		b.log.Infof("resolved image: %s", c.Image)
+		b.log.Infof("resolved public image: %s", c.PublicImage)
+	} else {
+		b.log.Infof("build request %s interrupted after %f seconds", request.Meta.Name, r.Task.Elapsed().Seconds())
+	}
 
 	// update the cache
 	b.request.Store(request.Meta.Name, r)
 
-	b.log.Infof("build request %s executed in %f seconds", request.Meta.Name, r.Task.Elapsed().Seconds())
-	b.log.Infof("dependencies: %s", request.Dependencies)
-	b.log.Infof("artifacts: %s", ArtifactIDs(c.Artifacts))
-	b.log.Infof("artifacts selected: %s", ArtifactIDs(c.SelectedArtifacts))
-	b.log.Infof("requested image: %s", request.Image)
-	b.log.Infof("base image: %s", c.BaseImage)
-	b.log.Infof("resolved image: %s", c.Image)
-	b.log.Infof("resolved public image: %s", c.PublicImage)
-
 	if handler != nil {
-		handler(r)
+		handler(&r)
 	}
 }

pkg/builder/builder_types.go

@@ -47,7 +47,7 @@ const (
 // Builder --
 type Builder interface {
 	IsBuilding(object v1.ObjectMeta) bool
-	Submit(request Request, handler func(Result))
+	Submit(request Request, handler func(*Result))
 	Close()
 }
 
@@ -102,6 +102,7 @@ type Resource struct {
 
 // Request --
 type Request struct {
+	C            context.Context
 	Meta         v1.ObjectMeta
 	Platform     v1alpha1.IntegrationPlatformSpec
 	Dependencies []string
@@ -190,4 +191,7 @@ const (
 
 	// StatusError --
 	StatusError
+
+	// StatusInterrupted --
+	StatusInterrupted
 )

pkg/controller/integration/build_image.go

@@ -52,11 +52,41 @@ func (action *buildImageAction) Name() string {
 }
 
 func (action *buildImageAction) CanHandle(integration *v1alpha1.Integration) bool {
-	return integration.Status.Phase == v1alpha1.IntegrationPhaseBuildingImage
+	if integration.Status.Phase == v1alpha1.IntegrationPhaseBuildImageSubmitted {
+		return true
+	}
+	if integration.Status.Phase == v1alpha1.IntegrationPhaseBuildImageRunning {
+		return true
+	}
+
+	return false
 }
 
 func (action *buildImageAction) Handle(ctx context.Context, integration *v1alpha1.Integration) error {
+	if integration.Status.Phase == v1alpha1.IntegrationPhaseBuildImageSubmitted {
+		return action.handleBuildImageSubmitted(ctx, integration)
+	}
+	if integration.Status.Phase == v1alpha1.IntegrationPhaseBuildImageRunning {
+		return action.handleBuildImageRunning(ctx, integration)
+	}
+
+	return nil
+}
+
+func (action *buildImageAction) handleBuildImageRunning(ctx context.Context, integration *v1alpha1.Integration) error {
+	b, err := platform.GetPlatformBuilder(action.client, integration.Namespace)
+	if err != nil {
+		return err
+	}
+
+	if b.IsBuilding(integration.ObjectMeta) {
+		logrus.Infof("Build for integration %s is running", integration.Name)
+	}
 
+	return nil
+}
+
+func (action *buildImageAction) handleBuildImageSubmitted(ctx context.Context, integration *v1alpha1.Integration) error {
 	// in this phase the integration need to be associated to a context whose image
 	// will be used as base image for the integration images
 	if integration.Status.Context == "" {
@@ -81,9 +111,14 @@ func (action *buildImageAction) Handle(ctx context.Context, integration *v1alpha
 			return err
 		}
 
-		// This build do not require to determine dependencies nor a project, the builder
-		// step do remove them
+		// This build do not require to determine dependencies nor a project, the
+		// builder step do remove them
+		//
+		// the context given to the handler is per reconcile loop and as the build
+		// happens asynchronously, a new context has to be created. the new context
+		// can be used also to stop the build.
 		r := builder.Request{
+			C:        context.TODO(),
 			Meta:     integration.ObjectMeta,
 			Steps:    env.Steps,
 			BuildDir: env.BuildDir,
@@ -99,15 +134,11 @@ func (action *buildImageAction) Handle(ctx context.Context, integration *v1alpha
 			return err
 		}
 
-		b.Submit(r, func(result builder.Result) {
-			// we can't use the handler ctx as this happen asynchronously so we
-			// need a new context
-			ctx := context.TODO()
-
+		b.Submit(r, func(result *builder.Result) {
 			//
 			// this function is invoked synchronously for every state change
 			//
-			if err := action.handleBuildStateChange(ctx, result); err != nil {
+			if err := action.handleBuildStateChange(result.Request.C, result); err != nil {
 				logrus.Warnf("Error while building integration image %s, reason: %s", ictx.Name, err.Error())
 			}
 		})
@@ -116,7 +147,7 @@ func (action *buildImageAction) Handle(ctx context.Context, integration *v1alpha
 	return nil
 }
 
-func (action *buildImageAction) handleBuildStateChange(ctx context.Context, res builder.Result) error {
+func (action *buildImageAction) handleBuildStateChange(ctx context.Context, res *builder.Result) error {
 	//
 	// Get the latest status of the integration
 	//
@@ -129,7 +160,11 @@ func (action *buildImageAction) handleBuildStateChange(ctx context.Context, res
 	case builder.StatusSubmitted:
 		logrus.Info("Build submitted")
 	case builder.StatusStarted:
-		logrus.Info("Build started")
+		target.Status.Phase = v1alpha1.IntegrationPhaseBuildImageRunning
+
+		logrus.Infof("Integration %s transitioning to state %s", target.Name, target.Status.Phase)
+
+		return action.client.Status().Update(ctx, target)
 	case builder.StatusError:
 		target.Status.Phase = v1alpha1.IntegrationPhaseBuildFailureRecovery