[WIP] Fix feedback on GCP CD implementation in cloud build

src/go.mod

@@ -39,6 +39,7 @@ require (
 	github.com/docker/cli v27.3.1+incompatible
 	github.com/docker/docker v27.3.1+incompatible
 	github.com/firebase/genkit/go v1.2.0
+	github.com/goccy/go-yaml v1.17.1
 	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/google/uuid v1.6.0
 	github.com/googleapis/gax-go/v2 v2.14.2
@@ -99,7 +100,6 @@ require (
 	github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
 	github.com/go-jose/go-jose/v4 v4.1.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
-	github.com/goccy/go-yaml v1.17.1 // indirect
 	github.com/google/dotprompt/go v0.0.0-20251014011017-8d056e027254 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/go-querystring v1.1.0 // indirect

src/pkg/cli/client/byoc/gcp/byoc.go

@@ -14,29 +14,25 @@ import (
 	"time"
 
 	"cloud.google.com/go/logging/apiv2/loggingpb"
-	run "cloud.google.com/go/run/apiv2"
-	"cloud.google.com/go/run/apiv2/runpb"
 	"cloud.google.com/go/storage"
 	"github.com/DefangLabs/defang/src/pkg"
 	"github.com/DefangLabs/defang/src/pkg/cli/client"
 	"github.com/DefangLabs/defang/src/pkg/cli/client/byoc"
 	"github.com/DefangLabs/defang/src/pkg/cli/compose"
-	"github.com/DefangLabs/defang/src/pkg/clouds"
-	"github.com/DefangLabs/defang/src/pkg/clouds/aws/ecs"
 	"github.com/DefangLabs/defang/src/pkg/clouds/gcp"
 	"github.com/DefangLabs/defang/src/pkg/dns"
 	"github.com/DefangLabs/defang/src/pkg/http"
 	"github.com/DefangLabs/defang/src/pkg/logs"
 	"github.com/DefangLabs/defang/src/pkg/term"
 	"github.com/DefangLabs/defang/src/pkg/types"
 	defangv1 "github.com/DefangLabs/defang/src/protos/io/defang/v1"
-	"github.com/aws/smithy-go/ptr"
 	"github.com/bufbuild/connect-go"
 	"google.golang.org/api/googleapi"
 	auditpb "google.golang.org/genproto/googleapis/cloud/audit"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
+	"gopkg.in/yaml.v3"
 )
 
 var _ client.Provider = (*ByocGcp)(nil)
@@ -186,6 +182,7 @@ func (b *ByocGcp) SetUpCD(ctx context.Context) error {
 		"roles/certificatemanager.owner",        // For creating certificates
 		"roles/serviceusage.serviceUsageAdmin",  // For allowing cd to Enable APIs
 		"roles/datastore.owner",                 // For creating firestore database
+		"roles/logging.logWriter",               // For allowing cloudbuild to write logs
 	}); err != nil {
 		return err
 	}
@@ -232,20 +229,6 @@ func (b *ByocGcp) SetUpCD(ctx context.Context) error {
 	// 5. Setup Cloud Run Job
 	term.Debugf("Using CD image: %q", b.CDImage)
 
-	serviceAccount := path.Base(b.cdServiceAccount)
-	if err := b.driver.SetupJob(ctx, "defang-cd", serviceAccount, []clouds.Container{
-		{
-			Image:     b.CDImage,
-			Name:      ecs.CdContainerName,
-			Cpus:      2.0,
-			Memory:    2048_000_000, // 2G
-			Essential: ptr.Bool(true),
-			WorkDir:   "/app",
-		},
-	}); err != nil {
-		return err
-	}
-
 	b.setupDone = true
 	return nil
 }
@@ -344,11 +327,18 @@ func (b *ByocGcp) CdCommand(ctx context.Context, req client.CdCommandRequest) (t
 type cdCommand struct {
 	command        []string
 	delegateDomain string
-	envOverride    map[string]string
+	etag           types.ETag
 	mode           defangv1.DeploymentMode
 	project        string
 }
 
+type CloudBuildStep struct {
+	Name       string   `yaml:"name,omitempty"`
+	Entrypoint string   `yaml:"entrypoint,omitempty"`
+	Args       []string `yaml:"args,omitempty"`
+	Env        []string `yaml:"env,omitempty"`
+}
+
 func (b *ByocGcp) runCdCommand(ctx context.Context, cmd cdCommand) (string, error) {
 	defangStateUrl := `gs://` + b.bucket
 	pulumiBackendKey, pulumiBackendValue, err := byoc.GetPulumiBackend(defangStateUrl)
@@ -384,8 +374,8 @@ func (b *ByocGcp) runCdCommand(ctx context.Context, cmd cdCommand) (string, erro
 		env["DOMAIN"] = "dummy.domain"
 	}
 
-	for k, v := range cmd.envOverride {
-		env[k] = v
+	if cmd.etag != "" {
+		env["DEFANG_ETAG"] = cmd.etag
 	}
 
 	if os.Getenv("DEFANG_PULUMI_DIR") != "" {
@@ -401,12 +391,35 @@ func (b *ByocGcp) runCdCommand(ctx context.Context, cmd cdCommand) (string, erro
 		}
 	}
 
-	execution, err := b.driver.Run(ctx, gcp.JobNameCD, env, cmd.command...)
+	var envs []string
+	for k, v := range env {
+		envs = append(envs, fmt.Sprintf("%s=%s", k, v))
+	}
+
+	steps, err := yaml.Marshal([]CloudBuildStep{
+		{
+			Name: b.CDImage,
+			Args: cmd.command,
+			Env:  envs,
+		},
+	})
+	if err != nil {
+		return "", err
+	}
+	execution, err := b.driver.RunCloudBuild(ctx, gcp.CloudBuildArgs{
+		Steps:          string(steps),
+		ServiceAccount: &b.cdServiceAccount,
+		Tags: []string{
+			fmt.Sprintf("%v_%v_%v_%v", b.PulumiStack, cmd.project, "cd", cmd.etag), // For cd logs, consistent with cloud build tagging
+			"defang-cd", // To indicate this is the actual cd service
+		},
+	})
 	if err != nil {
 		return "", err
 	}
 	b.cdExecution = execution
 	// term.Printf("CD Execution: %s\n", execution)
+
 	return execution, nil
 }
 
@@ -433,38 +446,7 @@ func (b *ByocGcp) Preview(ctx context.Context, req *defangv1.DeployRequest) (*de
 }
 
 func (b *ByocGcp) GetDeploymentStatus(ctx context.Context) error {
-	execClient, err := run.NewExecutionsClient(ctx)
-	if err != nil {
-		return err
-	}
-	defer execClient.Close()
-
-	execution, err := execClient.GetExecution(ctx, &runpb.GetExecutionRequest{Name: b.cdExecution})
-	if err != nil {
-		return err
-	}
-
-	var completionTime = execution.GetCompletionTime()
-	if completionTime != nil {
-		// cd is done
-		var failedTasks = execution.GetFailedCount()
-		if failedTasks > 0 {
-			var msgs []string
-			for _, condition := range execution.GetConditions() {
-				if condition.GetType() == "Completed" && condition.GetMessage() != "" {
-					msgs = append(msgs, condition.GetMessage())
-				}
-			}
-
-			return client.ErrDeploymentFailed{Message: strings.Join(msgs, ",")}
-		}
-
-		// completed successfully
-		return io.EOF
-	}
-
-	// still running
-	return nil
+	return b.driver.GetBuildStatus(ctx, b.cdExecution)
 }
 
 func (b *ByocGcp) deploy(ctx context.Context, req *defangv1.DeployRequest, command string) (*defangv1.DeployResponse, error) {
@@ -521,7 +503,7 @@ func (b *ByocGcp) deploy(ctx context.Context, req *defangv1.DeployRequest, comma
 	cdCmd := cdCommand{
 		command:        []string{command, payload},
 		delegateDomain: req.DelegateDomain,
-		envOverride:    map[string]string{"DEFANG_ETAG": etag},
+		etag:           etag,
 		mode:           req.Mode,
 		project:        project.Name,
 	}

src/pkg/cli/client/byoc/gcp/byoc_test.go

@@ -74,12 +74,12 @@ func (m MockGcpLogsClient) GetBuildInfo(ctx context.Context, buildId string) (*g
 }
 
 type MockGcpLoggingLister struct {
-	logEntries []loggingpb.LogEntry
+	logEntries []*loggingpb.LogEntry
 }
 
 func (m *MockGcpLoggingLister) Next() (*loggingpb.LogEntry, error) {
 	if len(m.logEntries) > 0 {
-		entry := &m.logEntries[0]
+		entry := m.logEntries[0]
 		m.logEntries = m.logEntries[1:]
 		return entry, nil
 	}

src/pkg/cli/client/byoc/gcp/stream.go

@@ -3,6 +3,7 @@ package gcp
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"path"
 	"regexp"
@@ -378,6 +379,7 @@ var cdExecutionNamePattern = regexp.MustCompile(`^defang-cd-[a-z0-9]{5}$`)
 
 func getLogEntryParser(ctx context.Context, gcpClient GcpLogsClient) func(entry *loggingpb.LogEntry) ([]*defangv1.TailResponse, error) {
 	envCache := make(map[string]map[string]string)
+	cdStarted := false
 	return func(entry *loggingpb.LogEntry) ([]*defangv1.TailResponse, error) {
 		if entry == nil {
 			return nil, nil
@@ -398,9 +400,12 @@ func getLogEntryParser(ctx context.Context, gcpClient GcpLogsClient) func(entry
 		}
 
 		var serviceName, etag, host string
+		var buildTags []string
 		serviceName = entry.Labels["defang-service"]
 		executionName := entry.Labels["run.googleapis.com/execution_name"]
-		buildTags := entry.Labels["build_tags"]
+		if entry.Labels["build_tags"] != "" {
+			buildTags = strings.Split(entry.Labels["build_tags"], ",")
+		}
 		// Log from service
 		if serviceName != "" {
 			etag = entry.Labels["defang-etag"]
@@ -421,7 +426,7 @@ func getLogEntryParser(ctx context.Context, gcpClient GcpLogsClient) func(entry
 				var err error
 				env, err = gcpClient.GetExecutionEnv(ctx, executionName)
 				if err != nil {
-					return nil, err
+					return nil, fmt.Errorf("failed to get execution environment variables: %w", err)
 				}
 				envCache[executionName] = env
 			}
@@ -435,14 +440,24 @@ func getLogEntryParser(ctx context.Context, gcpClient GcpLogsClient) func(entry
 			// use kaniko build job environment to get etag
 			etag = env["DEFANG_ETAG"]
 			host = "pulumi" // Hardcoded to match end condition detector in cmd/cli/command/compose.go
-		} else if buildTags != "" {
+		} else if len(buildTags) > 0 {
 			var bt gcp.BuildTag
 			if err := bt.Parse(buildTags); err != nil {
 				return nil, err
 			}
 			serviceName = bt.Service
 			etag = bt.Etag
 			host = "cloudbuild"
+			if bt.IsDefangCD {
+				host = "pulumi"
+			}
+			// HACK: Detect cd start from cloudbuild logs to skip the cloud build image pulling logs
+			if strings.HasPrefix(msg, " ** Update started") {
+				cdStarted = true
+			}
+			if !cdStarted {
+				return nil, nil // Skip cloudbuild logs (like pulling cd image) before cd started
+			}
 		} else {
 			var err error
 			_, msg, err = LogEntryToString(entry)
@@ -478,6 +493,23 @@ func getActivityParser(ctx context.Context, gcpLogsClient GcpLogsClient, waitFor
 
 	computeEngineRootTriggers := make(map[string]string)
 
+	getReadyServicesCompletedResps := func(status string) []*defangv1.SubscribeResponse {
+		resps := make([]*defangv1.SubscribeResponse, 0, len(readyServices))
+		for serviceName, status := range readyServices {
+			resps = append(resps, &defangv1.SubscribeResponse{
+				Name:   serviceName,
+				State:  defangv1.ServiceState_DEPLOYMENT_COMPLETED,
+				Status: status,
+			})
+		}
+		resps = append(resps, &defangv1.SubscribeResponse{
+			Name:   defangCD,
+			State:  defangv1.ServiceState_DEPLOYMENT_COMPLETED,
+			Status: status,
+		})
+		return resps
+	}
+
 	return func(entry *loggingpb.LogEntry) ([]*defangv1.SubscribeResponse, error) {
 		if entry == nil {
 			return nil, nil
@@ -575,20 +607,7 @@ func getActivityParser(ctx context.Context, gcpLogsClient GcpLogsClient, waitFor
 				}
 				cdSuccess = true
 				// Report all ready services when CD is successful, prevents cli deploy stop before cd is done
-				resps := make([]*defangv1.SubscribeResponse, 0, len(readyServices))
-				for serviceName, status := range readyServices {
-					resps = append(resps, &defangv1.SubscribeResponse{
-						Name:   serviceName,
-						State:  defangv1.ServiceState_DEPLOYMENT_COMPLETED,
-						Status: status,
-					})
-				}
-				resps = append(resps, &defangv1.SubscribeResponse{
-					Name:   defangCD,
-					State:  defangv1.ServiceState_DEPLOYMENT_COMPLETED,
-					Status: auditLog.GetStatus().GetMessage(),
-				})
-				return resps, nil // Ignore success cd status when we are waiting for service status
+				return getReadyServicesCompletedResps(auditLog.GetStatus().GetMessage()), nil // Ignore success cd status when we are waiting for service status
 			} else {
 				term.Warnf("unexpected execution name in audit log : %v", executionName)
 				return nil, nil
@@ -672,30 +691,43 @@ func getActivityParser(ctx context.Context, gcpLogsClient GcpLogsClient, waitFor
 				return nil, nil
 			}
 
-			var state defangv1.ServiceState
-			status := ""
-			if entry.Operation.First {
-				state = defangv1.ServiceState_BUILD_ACTIVATING
-			} else if entry.Operation.Last {
+			if bt.IsDefangCD {
+				if !entry.Operation.Last { // Ignore non-final cloud build event for CD
+					return nil, nil
+				}
+				// When cloud build fails, the last log message is an error message
 				if entry.Severity == logtype.LogSeverity_ERROR {
-					state = defangv1.ServiceState_BUILD_FAILED
-					if auditLog.GetStatus() != nil {
-						status = auditLog.GetStatus().String()
+					return nil, client.ErrDeploymentFailed{Message: auditLog.GetStatus().GetMessage()}
+				}
+
+				cdSuccess = true
+				return getReadyServicesCompletedResps(auditLog.GetStatus().String()), nil
+			} else {
+				var state defangv1.ServiceState
+				status := ""
+				if entry.Operation.First {
+					state = defangv1.ServiceState_BUILD_ACTIVATING
+				} else if entry.Operation.Last {
+					if entry.Severity == logtype.LogSeverity_ERROR {
+						state = defangv1.ServiceState_BUILD_FAILED
+						if auditLog.GetStatus() != nil {
+							status = auditLog.GetStatus().String()
+						}
+					} else {
+						state = defangv1.ServiceState_BUILD_STOPPING
 					}
 				} else {
-					state = defangv1.ServiceState_BUILD_STOPPING
+					state = defangv1.ServiceState_BUILD_RUNNING
 				}
-			} else {
-				state = defangv1.ServiceState_BUILD_RUNNING
-			}
-			if status == "" {
-				status = state.String()
+				if status == "" {
+					status = state.String()
+				}
+				return []*defangv1.SubscribeResponse{{
+					Name:   bt.Service,
+					State:  state,
+					Status: status,
+				}}, nil
 			}
-			return []*defangv1.SubscribeResponse{{
-				Name:   bt.Service,
-				State:  state,
-				Status: status,
-			}}, nil
 		default:
 			term.Warnf("unexpected resource type : %v", entry.Resource.Type)
 			return nil, nil

src/pkg/cli/client/byoc/gcp/stream_test.go

@@ -53,10 +53,10 @@ func TestServiceNameRestorer(t *testing.T) {
 		})
 	}
 }
-func makeMockLogEntries(n int) []loggingpb.LogEntry {
-	logEntries := make([]loggingpb.LogEntry, n)
+func makeMockLogEntries(n int) []*loggingpb.LogEntry {
+	logEntries := make([]*loggingpb.LogEntry, n)
 	for i := range logEntries {
-		logEntries[i] = loggingpb.LogEntry{
+		logEntries[i] = &loggingpb.LogEntry{
 			Payload: &loggingpb.LogEntry_TextPayload{
 				TextPayload: "Log entry number " + strconv.Itoa(i),
 			},