feat: grant ownership on tasks (#2684)

Changes:
- The follow-up grant ownership PR adds support to grant ownership on
Tasks.
- Add a reminder to the documentation to always copy the state file
before applying any state-manipulating functions.
- Added GrantOptionFor option during revoke in privilege-granting
resources, so only with_grant_option will be removed instead of the
whole grant, making the process less destructive

## Test Plan
<!-- detail ways in which this PR has been tested or needs to be tested
-->
* [x] acceptance tests
* [x] integration tests

## References
*
#2670
Another solution for granting ownership `on all tasks` - to be
considered in the future.
* [Grant
Ownership](https://docs.snowflake.com/en/sql-reference/sql/grant-ownership)

docs/resources/grant_ownership.md

@@ -6,14 +6,10 @@ description: |-
   
 ---
 
-
-!> **Warning** We're in a process of implementing this resource, so it's not available yet.
-
 ~> **Note** This is a preview resource. It's ready for general use. In case of any errors, please file an issue in our GitHub repository.
 ~> **Note** For more details about granting ownership, please visit [`GRANT OWNERSHIP` Snowflake documentation page](https://docs.snowflake.com/en/sql-reference/sql/grant-ownership).
 
-
-
+!> **Warning** Grant ownership resource still has some limitations. Delete operation is not implemented for on_future grants (you have to remove the config and then revoke ownership grant on future X manually).
 
 # snowflake_grant_ownership (Resource)
 
@@ -226,8 +222,18 @@ To transfer ownership of a pipe(s) **semi-automatically** you have to:
 2. Create Terraform configuration with the `snowflake_grant_ownership` resource and perform ownership transfer with the `terraform apply`.
 3. To resume the pipe(s) after ownership transfer use [PIPE_FORCE_RESUME system function](https://docs.snowflake.com/en/sql-reference/functions/system_pipe_force_resume).
 
-
-
+## Granting ownership on task
+Granting ownership on single task requires:
+- Either OWNERSHIP or OPERATE privilege to suspend the task (and its root)
+- Role that will be granted ownership has to have USAGE granted on the warehouse assigned to the task, as well as EXECUTE TASK granted globally
+- The outbound privileges set to `outbound_privileges = "COPY"` if you want to move grants automatically to the owner (also enables the provider to resume the task automatically)
+If originally the first owner won't be granted with OPERATE, USAGE (on the warehouse), EXECUTE TASK (on the account), and outbound privileges won't be set to `COPY`, then you have to resume suspended tasks manually.
+
+## Granting ownership on all tasks in database/schema
+Granting ownership on all tasks requires less privileges than granting ownership on one task, because it does a little bit less and requires additional work to be done after.
+The only thing you have to take care of is to resume tasks after grant ownership transfer. If all of your tasks are managed by the Snowflake Terraform Plugin, this should
+be as simple as running `terraform apply` second time (assuming the currently used role is privileged enough to be able to resume the tasks).
+If your tasks are not managed by the Snowflake Terraform Plugin, you should resume them yourself manually.
 
 ## Granting ownership on external tables
 Transferring ownership on an external table or its parent database blocks automatic refreshes of the table metadata by setting the `AUTO_REFRESH` property to `FALSE`.

docs/technical-documentation/resource_migration.md

@@ -15,6 +15,8 @@ resource "snowflake_database_grant" "old_resource" {
 }
 ```
 
+> **Important note:** **Always** save your state, before any state manipulation, so in case of failed migration, you will be able to recover from having incorrect state files.
+
 #### 1. terraform list
 
 First, we need to list all the grant resources that will need to be migrated.

pkg/resources/grant_ownership.go

@@ -291,6 +291,7 @@ func DeleteGrantOwnership(ctx context.Context, d *schema.ResourceData, meta any)
 
 	if grantOn.Future != nil {
 		// TODO (SNOW-1182623): Still waiting for the response on the behavior/SQL syntax we should use here
+		log.Printf("[WARN] Unsupported operation, please revoke ownership transfer manually")
 	} else {
 		accountRoleName, err := client.ContextFunctions.CurrentRole(ctx)
 		if err != nil {

pkg/resources/grant_ownership_acceptance_test.go

@@ -986,6 +986,96 @@ func TestAcc_GrantOwnership_OnAllPipes(t *testing.T) {
 	})
 }
 
+func TestAcc_GrantOwnership_OnTask(t *testing.T) {
+	accountRoleName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+	taskName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+	accountRoleFullyQualifiedName := sdk.NewAccountObjectIdentifier(accountRoleName).FullyQualifiedName()
+	taskFullyQualifiedName := sdk.NewSchemaObjectIdentifier(acc.TestDatabaseName, acc.TestSchemaName, taskName).FullyQualifiedName()
+
+	configVariables := config.Variables{
+		"account_role_name": config.StringVariable(accountRoleName),
+		"database":          config.StringVariable(acc.TestDatabaseName),
+		"schema":            config.StringVariable(acc.TestSchemaName),
+		"task":              config.StringVariable(taskName),
+		"warehouse":         config.StringVariable(acc.TestWarehouseName),
+	}
+	resourceName := "snowflake_grant_ownership.test"
+
+	resource.Test(t, resource.TestCase{
+		ProtoV6ProviderFactories: acc.TestAccProtoV6ProviderFactories,
+		PreCheck:                 func() { acc.TestAccPreCheck(t) },
+		TerraformVersionChecks: []tfversion.TerraformVersionCheck{
+			tfversion.RequireAbove(tfversion.Version1_5_0),
+		},
+		Steps: []resource.TestStep{
+			{
+				ConfigDirectory: acc.ConfigurationDirectory("TestAcc_GrantOwnership/OnTask"),
+				ConfigVariables: configVariables,
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "account_role_name", accountRoleName),
+					resource.TestCheckResourceAttr(resourceName, "on.0.object_type", sdk.ObjectTypeTask.String()),
+					resource.TestCheckResourceAttr(resourceName, "on.0.object_name", taskFullyQualifiedName),
+					resource.TestCheckResourceAttr(resourceName, "id", fmt.Sprintf("ToAccountRole|%s||OnObject|TASK|%s", accountRoleFullyQualifiedName, taskFullyQualifiedName)),
+					checkResourceOwnershipIsGranted(&sdk.ShowGrantOptions{
+						On: &sdk.ShowGrantsOn{
+							Object: &sdk.Object{
+								ObjectType: sdk.ObjectTypeTask,
+								Name:       sdk.NewSchemaObjectIdentifierFromFullyQualifiedName(taskFullyQualifiedName),
+							},
+						},
+						// TODO(SNOW-999049): Fix this identifier
+					}, sdk.ObjectTypeTask, accountRoleName, fmt.Sprintf("%s\".\"%s\".%s", acc.TestDatabaseName, acc.TestSchemaName, taskName)),
+				),
+			},
+		},
+	})
+}
+
+func TestAcc_GrantOwnership_OnAllTasks(t *testing.T) {
+	accountRoleName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+	taskName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+	secondTaskName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+	accountRoleFullyQualifiedName := sdk.NewAccountObjectIdentifier(accountRoleName).FullyQualifiedName()
+	schemaFullyQualifiedName := sdk.NewDatabaseObjectIdentifier(acc.TestDatabaseName, acc.TestSchemaName).FullyQualifiedName()
+
+	configVariables := config.Variables{
+		"account_role_name": config.StringVariable(accountRoleName),
+		"database":          config.StringVariable(acc.TestDatabaseName),
+		"schema":            config.StringVariable(acc.TestSchemaName),
+		"task":              config.StringVariable(taskName),
+		"second_task":       config.StringVariable(secondTaskName),
+	}
+	resourceName := "snowflake_grant_ownership.test"
+
+	resource.Test(t, resource.TestCase{
+		ProtoV6ProviderFactories: acc.TestAccProtoV6ProviderFactories,
+		PreCheck:                 func() { acc.TestAccPreCheck(t) },
+		TerraformVersionChecks: []tfversion.TerraformVersionCheck{
+			tfversion.RequireAbove(tfversion.Version1_5_0),
+		},
+		Steps: []resource.TestStep{
+			{
+				ConfigDirectory: acc.ConfigurationDirectory("TestAcc_GrantOwnership/OnAllTasks"),
+				ConfigVariables: configVariables,
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr(resourceName, "account_role_name", accountRoleName),
+					resource.TestCheckResourceAttr(resourceName, "id", fmt.Sprintf("ToAccountRole|%s|REVOKE|OnAll|TASKS|InSchema|%s", accountRoleFullyQualifiedName, schemaFullyQualifiedName)),
+					checkResourceOwnershipIsGranted(&sdk.ShowGrantOptions{
+						To: &sdk.ShowGrantsTo{
+							Role: sdk.NewAccountObjectIdentifier(accountRoleName),
+						},
+					},
+						sdk.ObjectTypeTask, accountRoleName,
+						// TODO(SNOW-999049): Fix this identifier
+						fmt.Sprintf("%s\".\"%s\".%s", acc.TestDatabaseName, acc.TestSchemaName, taskName),
+						fmt.Sprintf("%s\".\"%s\".%s", acc.TestDatabaseName, acc.TestSchemaName, secondTaskName),
+					),
+				),
+			},
+		},
+	})
+}
+
 func createDatabaseWithRoleAsOwner(t *testing.T, roleName string, databaseName string) func() {
 	t.Helper()
 	client, err := sdk.NewDefaultClient()

pkg/resources/grant_privileges_to_account_role.go

@@ -527,7 +527,9 @@ func UpdateGrantPrivilegesToAccountRole(ctx context.Context, d *schema.ResourceD
 				)
 
 				if !id.WithGrantOption {
-					if err = client.Grants.RevokePrivilegesFromAccountRole(ctx, privilegesToGrant, grantOn, id.RoleName, new(sdk.RevokePrivilegesFromAccountRoleOptions)); err != nil {
+					if err = client.Grants.RevokePrivilegesFromAccountRole(ctx, privilegesToGrant, grantOn, id.RoleName, &sdk.RevokePrivilegesFromAccountRoleOptions{
+						GrantOptionFor: sdk.Bool(true),
+					}); err != nil {
 						return diag.Diagnostics{
 							diag.Diagnostic{
 								Severity: diag.Error,

pkg/resources/grant_privileges_to_database_role.go

@@ -453,7 +453,9 @@ func UpdateGrantPrivilegesToDatabaseRole(ctx context.Context, d *schema.Resource
 				)
 
 				if !id.WithGrantOption {
-					if err = client.Grants.RevokePrivilegesFromDatabaseRole(ctx, privilegesToGrant, grantOn, id.DatabaseRoleName, new(sdk.RevokePrivilegesFromDatabaseRoleOptions)); err != nil {
+					if err = client.Grants.RevokePrivilegesFromDatabaseRole(ctx, privilegesToGrant, grantOn, id.DatabaseRoleName, &sdk.RevokePrivilegesFromDatabaseRoleOptions{
+						GrantOptionFor: sdk.Bool(true),
+					}); err != nil {
 						return diag.Diagnostics{
 							diag.Diagnostic{
 								Severity: diag.Error,

pkg/resources/testdata/TestAcc_GrantOwnership/OnAllTasks/test.tf

@@ -0,0 +1,31 @@
+resource "snowflake_role" "test" {
+  name = var.account_role_name
+}
+
+resource "snowflake_task" "test" {
+  database      = var.database
+  schema        = var.schema
+  name          = var.task
+  sql_statement = "SELECT CURRENT_TIMESTAMP"
+}
+
+resource "snowflake_task" "second_test" {
+  database      = var.database
+  schema        = var.schema
+  name          = var.second_task
+  sql_statement = "SELECT CURRENT_TIMESTAMP"
+}
+
+resource "snowflake_grant_ownership" "test" {
+  depends_on        = [snowflake_task.test, snowflake_task.second_test]
+  account_role_name = snowflake_role.test.name
+
+  on {
+    all {
+      object_type_plural = "TASKS"
+      in_schema          = "\"${var.database}\".\"${var.schema}\""
+    }
+  }
+
+  outbound_privileges = "REVOKE"
+}

pkg/resources/testdata/TestAcc_GrantOwnership/OnAllTasks/variables.tf

@@ -0,0 +1,19 @@
+variable "account_role_name" {
+  type = string
+}
+
+variable "database" {
+  type = string
+}
+
+variable "schema" {
+  type = string
+}
+
+variable "task" {
+  type = string
+}
+
+variable "second_task" {
+  type = string
+}

pkg/resources/testdata/TestAcc_GrantOwnership/OnTask/test.tf

@@ -0,0 +1,20 @@
+resource "snowflake_role" "test" {
+  name = var.account_role_name
+}
+
+resource "snowflake_task" "test" {
+  database      = var.database
+  schema        = var.schema
+  name          = var.task
+  warehouse     = var.warehouse
+  sql_statement = "SELECT CURRENT_TIMESTAMP"
+}
+
+resource "snowflake_grant_ownership" "test" {
+  account_role_name = snowflake_role.test.name
+
+  on {
+    object_type = "TASK"
+    object_name = "\"${snowflake_task.test.database}\".\"${snowflake_task.test.schema}\".\"${snowflake_task.test.name}\""
+  }
+}

pkg/resources/testdata/TestAcc_GrantOwnership/OnTask/variables.tf

@@ -0,0 +1,19 @@
+variable "account_role_name" {
+  type = string
+}
+
+variable "database" {
+  type = string
+}
+
+variable "schema" {
+  type = string
+}
+
+variable "task" {
+  type = string
+}
+
+variable "warehouse" {
+  type = string
+}