resource_bigtable_gc_policy.go

// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: MPL-2.0
package bigtable

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"reflect"
	"strings"
	"time"

	"cloud.google.com/go/bigtable"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/structure"
	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"

	"github.com/hashicorp/terraform-provider-google-beta/google-beta/tpgresource"
	transport_tpg "github.com/hashicorp/terraform-provider-google-beta/google-beta/transport"
	"github.com/hashicorp/terraform-provider-google-beta/google-beta/verify"
)

const (
	GCPolicyModeIntersection = "INTERSECTION"
	GCPolicyModeUnion        = "UNION"
)

func resourceBigtableGCPolicyCustomizeDiffFunc(diff tpgresource.TerraformResourceDiff) error {
	count := diff.Get("max_age.#").(int)
	if count < 1 {
		return nil
	}

	oldDays, newDays := diff.GetChange("max_age.0.days")
	oldDuration, newDuration := diff.GetChange("max_age.0.duration")
	log.Printf("days: %v %v", oldDays, newDays)
	log.Printf("duration: %v %v", oldDuration, newDuration)

	if oldDuration == "" && newDuration != "" {
		// flatten the old days and the new duration to duration... if they are
		// equal then do nothing.
		do, err := ParseDuration(newDuration.(string))
		if err != nil {
			return err
		}
		dn := time.Hour * 24 * time.Duration(oldDays.(int))
		if do == dn {
			err := diff.Clear("max_age.0.days")
			if err != nil {
				return err
			}
			err = diff.Clear("max_age.0.duration")
			if err != nil {
				return err
			}
		}
	}

	return nil
}
func gcRulesDiffSuppress(oldRules, newRules interface{}) bool {
	var oldPolicyRaw map[string]interface{}
	if err := json.Unmarshal([]byte(oldRules.(string)), &oldPolicyRaw); err != nil {
		return false
	}
	oldPolicy, err := getGCPolicyFromJSON(oldPolicyRaw /*isTopLevel=*/, true)
	if err != nil {
		return false
	}
	var newPolicyRaw map[string]interface{}
	if err := json.Unmarshal([]byte(newRules.(string)), &newPolicyRaw); err != nil {
		return false
	}
	newPolicy, err := getGCPolicyFromJSON(newPolicyRaw /*isTopLevel=*/, true)
	if err != nil {
		return false
	}
	oldPolicyString, err := GcPolicyToGCRuleString(oldPolicy /*isTopLevel=*/, true)
	if err != nil {
		return false
	}
	newPolicyString, err := GcPolicyToGCRuleString(newPolicy /*isTopLevel=*/, true)
	if err != nil {
		return false
	}
	if reflect.DeepEqual(oldPolicyString, newPolicyString) {
		return true
	}
	return false
}

func resourceBigtableGCPolicyCustomizeDiff(_ context.Context, d *schema.ResourceDiff, meta interface{}) error {
	return resourceBigtableGCPolicyCustomizeDiffFunc(d)
}

func ResourceBigtableGCPolicy() *schema.Resource {
	return &schema.Resource{
		Create:        resourceBigtableGCPolicyUpsert,
		Read:          resourceBigtableGCPolicyRead,
		Delete:        resourceBigtableGCPolicyDestroy,
		Update:        resourceBigtableGCPolicyUpsert,
		CustomizeDiff: resourceBigtableGCPolicyCustomizeDiff,

		Timeouts: &schema.ResourceTimeout{
			Create: schema.DefaultTimeout(20 * time.Minute),
			Delete: schema.DefaultTimeout(20 * time.Minute),
		},

		Schema: map[string]*schema.Schema{
			"instance_name": {
				Type:             schema.TypeString,
				Required:         true,
				ForceNew:         true,
				DiffSuppressFunc: tpgresource.CompareResourceNames,
				Description:      `The name of the Bigtable instance.`,
			},

			"table": {
				Type:        schema.TypeString,
				Required:    true,
				ForceNew:    true,
				Description: `The name of the table.`,
			},

			"column_family": {
				Type:        schema.TypeString,
				Required:    true,
				ForceNew:    true,
				Description: `The name of the column family.`,
			},

			"gc_rules": {
				Type:          schema.TypeString,
				Optional:      true,
				Description:   `Serialized JSON string for garbage collection policy. Conflicts with "mode", "max_age" and "max_version".`,
				ValidateFunc:  validation.StringIsJSON,
				ConflictsWith: []string{"mode", "max_age", "max_version"},
				StateFunc: func(v interface{}) string {
					json, _ := structure.NormalizeJsonString(v)
					return json
				},
				DiffSuppressFunc: func(k, old, new string, _ *schema.ResourceData) bool {
					return gcRulesDiffSuppress(old, new)
				},
			},
			"mode": {
				Type:          schema.TypeString,
				Optional:      true,
				ForceNew:      true,
				Description:   `NOTE: 'gc_rules' is more flexible, and should be preferred over this field for new resources. This field may be deprecated in the future. If multiple policies are set, you should choose between UNION OR INTERSECTION.`,
				ValidateFunc:  validation.StringInSlice([]string{GCPolicyModeIntersection, GCPolicyModeUnion}, false),
				ConflictsWith: []string{"gc_rules"},
			},

			"max_age": {
				Type:        schema.TypeList,
				Optional:    true,
				ForceNew:    true,
				Description: `NOTE: 'gc_rules' is more flexible, and should be preferred over this field for new resources. This field may be deprecated in the future. GC policy that applies to all cells older than the given age.`,
				MaxItems:    1,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"days": {
							Type:         schema.TypeInt,
							Optional:     true,
							Computed:     true,
							ForceNew:     true,
							Deprecated:   "Deprecated in favor of duration",
							Description:  `Number of days before applying GC policy.`,
							ExactlyOneOf: []string{"max_age.0.days", "max_age.0.duration"},
						},
						"duration": {
							Type:         schema.TypeString,
							Optional:     true,
							Computed:     true,
							ForceNew:     true,
							Description:  `Duration before applying GC policy`,
							ValidateFunc: verify.ValidateDuration(),
							ExactlyOneOf: []string{"max_age.0.days", "max_age.0.duration"},
						},
					},
				},
				ConflictsWith: []string{"gc_rules"},
			},

			"max_version": {
				Type:        schema.TypeList,
				Optional:    true,
				ForceNew:    true,
				Description: `NOTE: 'gc_rules' is more flexible, and should be preferred over this field for new resources. This field may be deprecated in the future. GC policy that applies to all versions of a cell except for the most recent.`,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"number": {
							Type:        schema.TypeInt,
							Required:    true,
							ForceNew:    true,
							Description: `Number of version before applying the GC policy.`,
						},
					},
				},
				ConflictsWith: []string{"gc_rules"},
			},

			"project": {
				Type:        schema.TypeString,
				Optional:    true,
				Computed:    true,
				ForceNew:    true,
				Description: `The ID of the project in which the resource belongs. If it is not provided, the provider project is used.`,
			},

			"deletion_policy": {
				Type:     schema.TypeString,
				Optional: true,
				Description: `The deletion policy for the GC policy. Setting ABANDON allows the resource
				to be abandoned rather than deleted. This is useful for GC policy as it cannot be deleted
				in a replicated instance. Possible values are: "ABANDON".`,
				ValidateFunc: validation.StringInSlice([]string{"ABANDON", ""}, false),
			},

			"ignore_warnings": {
				Type:     schema.TypeBool,
				Optional: true,
				Description: `Allows ignoring warnings when updating the GC policy. This can be used
				to increase the gc policy on replicated clusters. Doing this may make clusters be
				inconsistent for a longer period of time, before using this make sure you understand
				the risks listed at https://cloud.google.com/bigtable/docs/garbage-collection#increasing`,
				Default: false,
			},
		},
		UseJSONNumber: true,
	}
}

func resourceBigtableGCPolicyUpsert(d *schema.ResourceData, meta interface{}) error {
	config := meta.(*transport_tpg.Config)
	userAgent, err := tpgresource.GenerateUserAgentString(d, config.UserAgent)
	if err != nil {
		return err
	}

	ctx := context.Background()

	project, err := tpgresource.GetProject(d, config)
	if err != nil {
		return err
	}

	instanceName := tpgresource.GetResourceNameFromSelfLink(d.Get("instance_name").(string))
	c, err := config.BigTableClientFactory(userAgent).NewAdminClient(project, instanceName)
	if err != nil {
		return fmt.Errorf("Error starting admin client. %s", err)
	}
	if err := d.Set("instance_name", instanceName); err != nil {
		return fmt.Errorf("Error setting instance_name: %s", err)
	}

	defer c.Close()

	gcPolicy, err := generateBigtableGCPolicy(d)
	if err != nil {
		return err
	}

	tableName := d.Get("table").(string)
	columnFamily := d.Get("column_family").(string)
	ignoreWarnings := d.Get("ignore_warnings").(bool)
	updateOpts := []bigtable.GCPolicyOption{}
	if ignoreWarnings {
		updateOpts = append(updateOpts, bigtable.IgnoreWarnings())
	}

	retryFunc := func() error {
		reqErr := c.SetGCPolicyWithOptions(ctx, tableName, columnFamily, gcPolicy, updateOpts...)
		return reqErr
	}
	// The default create timeout is 20 minutes.
	timeout := d.Timeout(schema.TimeoutCreate)
	pollInterval := time.Duration(30) * time.Second
	// Mutations to gc policies can only happen one-at-a-time and take some amount of time.
	// Use a fixed polling rate of 30s based on the RetryInfo returned by the server rather than
	// the standard up-to-10s exponential backoff for those operations.
	err = transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc:            retryFunc,
		Timeout:              timeout,
		PollInterval:         pollInterval,
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsBigTableRetryableError},
	})
	if err != nil {
		return err
	}

	table, err := c.TableInfo(ctx, tableName)
	if err != nil {
		return fmt.Errorf("Error retrieving table. Could not find %s in %s. %s", tableName, instanceName, err)
	}

	for _, i := range table.FamilyInfos {
		if i.Name == columnFamily {
			d.SetId(i.GCPolicy)
		}
	}

	return resourceBigtableGCPolicyRead(d, meta)
}

func resourceBigtableGCPolicyRead(d *schema.ResourceData, meta interface{}) error {
	config := meta.(*transport_tpg.Config)
	userAgent, err := tpgresource.GenerateUserAgentString(d, config.UserAgent)
	if err != nil {
		return err
	}
	ctx := context.Background()

	project, err := tpgresource.GetProject(d, config)
	if err != nil {
		return err
	}

	instanceName := tpgresource.GetResourceNameFromSelfLink(d.Get("instance_name").(string))
	c, err := config.BigTableClientFactory(userAgent).NewAdminClient(project, instanceName)
	if err != nil {
		return fmt.Errorf("Error starting admin client. %s", err)
	}

	defer c.Close()

	name := d.Get("table").(string)
	columnFamily := d.Get("column_family").(string)
	ti, err := c.TableInfo(ctx, name)
	if err != nil {
		if tpgresource.IsNotFoundGrpcError(err) {
			log.Printf("[WARN] Removing the GC policy because the parent table %s is gone", name)
			d.SetId("")
			return nil
		}
		return err
	}

	for _, fi := range ti.FamilyInfos {
		if fi.Name != columnFamily {
			continue
		}

		d.SetId(fi.GCPolicy)

		// No GC Policy.
		if fi.FullGCPolicy.String() == "" {
			return nil
		}

		// Only set `gc_rules`` when the legacy fields are not set. We are not planning to support legacy fields.
		maxAge := d.Get("max_age")
		maxVersion := d.Get("max_version")
		if len(maxAge.([]interface{})) == 0 && len(maxVersion.([]interface{})) == 0 {
			gcRuleString, err := GcPolicyToGCRuleString(fi.FullGCPolicy, true)
			if err != nil {
				return err
			}
			gcRuleJsonString, err := json.Marshal(gcRuleString)
			if err != nil {
				return fmt.Errorf("Error marshaling GC policy to json: %s", err)
			}
			d.Set("gc_rules", string(gcRuleJsonString))
		}
		break
	}

	if err := d.Set("project", project); err != nil {
		return fmt.Errorf("Error setting project: %s", err)
	}

	return nil
}

// Recursively convert Bigtable GC policy to JSON format in a map.
func GcPolicyToGCRuleString(gc bigtable.GCPolicy, topLevel bool) (map[string]interface{}, error) {
	result := make(map[string]interface{})
	switch bigtable.GetPolicyType(gc) {
	case bigtable.PolicyMaxAge:
		age := gc.(bigtable.MaxAgeGCPolicy).GetDurationString()
		if topLevel {
			rule := make(map[string]interface{})
			rule["max_age"] = age
			rules := []interface{}{}
			rules = append(rules, rule)
			result["rules"] = rules
		} else {
			result["max_age"] = age
		}
		break
	case bigtable.PolicyMaxVersion:
		// bigtable.MaxVersionsGCPolicy is an int.
		// Not sure why max_version is a float64.
		// TODO: Maybe change max_version to an int.
		version := float64(int(gc.(bigtable.MaxVersionsGCPolicy)))
		if topLevel {
			rule := make(map[string]interface{})
			rule["max_version"] = version
			rules := []interface{}{}
			rules = append(rules, rule)
			result["rules"] = rules
		} else {
			result["max_version"] = version
		}
		break
	case bigtable.PolicyUnion:
		result["mode"] = "union"
		rules := []interface{}{}
		for _, c := range gc.(bigtable.UnionGCPolicy).Children {
			gcRuleString, err := GcPolicyToGCRuleString(c, false)
			if err != nil {
				return nil, err
			}
			rules = append(rules, gcRuleString)
		}
		result["rules"] = rules
		break
	case bigtable.PolicyIntersection:
		result["mode"] = "intersection"
		rules := []interface{}{}
		for _, c := range gc.(bigtable.IntersectionGCPolicy).Children {
			gcRuleString, err := GcPolicyToGCRuleString(c, false)
			if err != nil {
				return nil, err
			}
			rules = append(rules, gcRuleString)
		}
		result["rules"] = rules
	default:
		break
	}

	if err := validateNestedPolicy(result, topLevel); err != nil {
		return nil, err
	}

	return result, nil
}

func resourceBigtableGCPolicyDestroy(d *schema.ResourceData, meta interface{}) error {
	config := meta.(*transport_tpg.Config)

	if deletionPolicy := d.Get("deletion_policy"); deletionPolicy == "ABANDON" {
		// Allows for the GC policy to be abandoned without deletion to avoid possible
		// deletion failure in a replicated instance.
		log.Printf("[WARN] The GC policy is abandoned")
		return nil
	}

	userAgent, err := tpgresource.GenerateUserAgentString(d, config.UserAgent)
	if err != nil {
		return err
	}
	ctx := context.Background()

	project, err := tpgresource.GetProject(d, config)
	if err != nil {
		return err
	}

	instanceName := tpgresource.GetResourceNameFromSelfLink(d.Get("instance_name").(string))
	c, err := config.BigTableClientFactory(userAgent).NewAdminClient(project, instanceName)
	if err != nil {
		return fmt.Errorf("Error starting admin client. %s", err)
	}

	defer c.Close()

	retryFunc := func() error {
		reqErr := c.SetGCPolicy(ctx, d.Get("table").(string), d.Get("column_family").(string), bigtable.NoGcPolicy())
		return reqErr
	}
	// The default delete timeout is 20 minutes.
	timeout := d.Timeout(schema.TimeoutDelete)
	pollInterval := time.Duration(30) * time.Second
	err = transport_tpg.Retry(transport_tpg.RetryOptions{
		RetryFunc:            retryFunc,
		Timeout:              timeout,
		PollInterval:         pollInterval,
		ErrorRetryPredicates: []transport_tpg.RetryErrorPredicateFunc{transport_tpg.IsBigTableRetryableError},
	})
	if err != nil {
		return err
	}

	d.SetId("")

	return nil
}

func generateBigtableGCPolicy(d *schema.ResourceData) (bigtable.GCPolicy, error) {
	var policies []bigtable.GCPolicy
	mode := d.Get("mode").(string)
	ma, aok := d.GetOk("max_age")
	mv, vok := d.GetOk("max_version")
	gcRules, gok := d.GetOk("gc_rules")

	if !aok && !vok && !gok {
		return bigtable.NoGcPolicy(), nil
	}

	if mode == "" && aok && vok {
		return nil, fmt.Errorf("if multiple policies are set, mode can't be empty")
	}

	if gok {
		var topLevelPolicy map[string]interface{}
		if err := json.Unmarshal([]byte(gcRules.(string)), &topLevelPolicy); err != nil {
			return nil, err
		}
		return getGCPolicyFromJSON(topLevelPolicy /*isTopLevel=*/, true)
	}

	if aok {
		l, _ := ma.([]interface{})
		d, err := getMaxAgeDuration(l[0].(map[string]interface{}))
		if err != nil {
			return nil, err
		}

		policies = append(policies, bigtable.MaxAgePolicy(d))
	}

	if vok {
		l, _ := mv.([]interface{})
		n, _ := l[0].(map[string]interface{})["number"].(int)

		policies = append(policies, bigtable.MaxVersionsPolicy(n))
	}

	switch mode {
	case GCPolicyModeUnion:
		return bigtable.UnionPolicy(policies...), nil
	case GCPolicyModeIntersection:
		return bigtable.IntersectionPolicy(policies...), nil
	}

	return policies[0], nil
}

func getGCPolicyFromJSON(inputPolicy map[string]interface{}, isTopLevel bool) (bigtable.GCPolicy, error) {
	policy := []bigtable.GCPolicy{}

	if err := validateNestedPolicy(inputPolicy, isTopLevel); err != nil {
		return nil, err
	}

	for _, p := range inputPolicy["rules"].([]interface{}) {
		childPolicy := p.(map[string]interface{})
		if err := validateNestedPolicy(childPolicy /*isTopLevel=*/, false); err != nil {
			return nil, err
		}

		if childPolicy["max_age"] != nil {
			maxAge := childPolicy["max_age"].(string)
			duration, err := ParseDuration(maxAge)
			if err != nil {
				return nil, fmt.Errorf("invalid duration string: %v", maxAge)
			}
			policy = append(policy, bigtable.MaxAgePolicy(duration))
		}

		if childPolicy["max_version"] != nil {
			version := childPolicy["max_version"].(float64)
			policy = append(policy, bigtable.MaxVersionsPolicy(int(version)))
		}

		if childPolicy["mode"] != nil {
			n, err := getGCPolicyFromJSON(childPolicy /*isTopLevel=*/, false)
			if err != nil {
				return nil, err
			}
			policy = append(policy, n)
		}
	}

	switch inputPolicy["mode"] {
	case strings.ToLower(GCPolicyModeUnion):
		return bigtable.UnionPolicy(policy...), nil
	case strings.ToLower(GCPolicyModeIntersection):
		return bigtable.IntersectionPolicy(policy...), nil
	default:
		return policy[0], nil
	}
}

func validateNestedPolicy(p map[string]interface{}, isTopLevel bool) error {
	if len(p) > 2 {
		return fmt.Errorf("rules has more than 2 fields")
	}
	maxVersion, maxVersionOk := p["max_version"]
	maxAge, maxAgeOk := p["max_age"]
	rulesObj, rulesOk := p["rules"]

	_, modeOk := p["mode"]
	rules, arrOk := rulesObj.([]interface{})
	_, vCastOk := maxVersion.(float64)
	_, aCastOk := maxAge.(string)

	if rulesOk && !arrOk {
		return fmt.Errorf("`rules` must be array")
	}

	if modeOk && len(rules) < 2 {
		return fmt.Errorf("`rules` need at least 2 GC rule when mode is specified")
	}

	if isTopLevel && !rulesOk {
		return fmt.Errorf("invalid nested policy, need `rules`")
	}

	if isTopLevel && !modeOk && len(rules) != 1 {
		return fmt.Errorf("when `mode` is not specified, `rules` can only have 1 child rule")
	}

	if !isTopLevel && len(p) == 2 && (!modeOk || !rulesOk) {
		return fmt.Errorf("need `mode` and `rules` for child nested policies")
	}

	if !isTopLevel && len(p) == 1 && !maxVersionOk && !maxAgeOk {
		return fmt.Errorf("need `max_version` or `max_age` for the rule")
	}

	if maxVersionOk && !vCastOk {
		return fmt.Errorf("`max_version` must be a number")
	}

	if maxAgeOk && !aCastOk {
		return fmt.Errorf("`max_age must be a string")
	}

	return nil
}

func getMaxAgeDuration(values map[string]interface{}) (time.Duration, error) {
	d := values["duration"].(string)
	if d != "" {
		return ParseDuration(d)
	}

	days := values["days"].(int)

	return time.Hour * 24 * time.Duration(days), nil
}