Moving from flyteadmin - Add UpdateTag Function

flyteadmin/pkg/manager/impl/execution_manager.go

@@ -1340,12 +1340,16 @@ func (m *ExecutionManager) UpdateExecution(ctx context.Context, request admin.Ex
 		logger.Debugf(ctx, "Failed to get execution model for request [%+v] with err: %v", request, err)
 		return nil, err
 	}
-
+	
 	if err = transformers.UpdateExecutionModelStateChangeDetails(executionModel, request.State, requestedAt,
 		getUser(ctx)); err != nil {
 		return nil, err
 	}
 
+	if err = transformers.UpdateExecutionModelTag(executionModel, request.Tags); err != nil {
+		return nil, err
+	}
+
 	if err := m.db.ExecutionRepo().Update(ctx, *executionModel); err != nil {
 		return nil, err
 	}

flyteadmin/pkg/repositories/transformers/execution.go

@@ -103,7 +103,7 @@ func CreateExecutionModel(input CreateExecutionModelInput) (*models.Execution, e
 	for i, tag := range input.RequestSpec.Tags {
 		tags[i] = models.AdminTag{Name: tag}
 	}
-
+	
 	executionModel := &models.Execution{
 		ExecutionKey: models.ExecutionKey{
 			Project: input.WorkflowExecutionID.Project,
@@ -292,6 +292,37 @@ func UpdateExecutionModelStateChangeDetails(executionModel *models.Execution, st
 	return nil
 }
 
+// Update tag information of existing execution model.
+func UpdateExecutionModelTag(executionModel *models.Execution, tagsUpdatedTo []string) error {
+
+	var spec admin.ExecutionSpec
+	var err error
+	if err = proto.Unmarshal(executionModel.Spec, &spec); err != nil {
+		return errors.NewFlyteAdminErrorf(codes.Internal, "failed to unmarshal spec")
+	}
+
+	tagSet := sets.NewString()
+	for _, tag := range spec.Tags {
+		tagSet.Insert(tag)
+	}
+	for _, tag := range tagsUpdatedTo {
+		// if tag not in tagSet, append into executionModel.Tags
+		if !tagSet.Has(tag) {
+			executionModel.Tags = append(executionModel.Tags, models.AdminTag{Name: tag})
+		}
+		tagSet.Insert(tag)
+	}
+	spec.Tags = tagSet.List()
+
+	marshaledSpec, err := proto.Marshal(&spec)
+	if err != nil {
+		return errors.NewFlyteAdminErrorf(codes.Internal, "Failed to marshal execution spec: %v", err)
+	}
+	executionModel.Spec = marshaledSpec
+
+	return nil
+}
+
 // The execution abort metadata is recorded but the phase is not actually updated *until* the abort event is propagated
 // by flytepropeller. The metadata is preemptively saved at the time of the abort.
 func SetExecutionAborting(execution *models.Execution, cause, principal string) error {