add parameter metadata lookup to param handling

Adds a cache to the DB parameter group hooks.go file that stores
metadata about engine family parameter defaults. This metadata informs
the resource manager whether a particular parameter is modifiable and
whether changes to it can be applied immediately or require a reboot of
the DB instance.

The cache is a simple one, with no support for TTL or expiry and
because there is nowhere to "attach" the cache that has access to the
SDK API object (in order to call RDS `DescribeEngineDefaultParameters`
API), the cache's `get` method has to have a "fetcher" function pointer
passed into it. This isn't super clean but there isn't currently a way
to, say, add the cache as a member of the `AWSResourceManager` struct
since that is all code-generated and this is a specific to the DB
Parameter Group resource manager.

Signed-off-by: Jay Pipes <[email protected]>

Author: jaypipes

apis/v1alpha1/ack-generate-metadata.yaml

@@ -3,11 +3,11 @@ ack_generate_info:
   build_hash: fe61d04673fd4d9848d5f726b01e0689a16d3733
   go_version: go1.18.2
   version: v0.19.3-1-gfe61d04
-api_directory_checksum: 420612ef2bc7cf0b19ebe41c44626560e060c4a1
+api_directory_checksum: 83bde1f77a49ebd8acd36a1c5e231138b67aa883
 api_version: v1alpha1
 aws_sdk_go_version: v1.44.27
 generator_config_info:
-  file_checksum: 90a2d0d868a32e6474738ad70a557035f490780a
+  file_checksum: 5d126ead4715f729c2fda976c422eb14a948f3da
   original_file_name: generator.yaml
 last_modification:
   reason: API generation

pkg/resource/db_parameter_group/hooks.go

@@ -15,8 +15,11 @@ package db_parameter_group
 
 import (
 	"context"
+	"fmt"
+	"sync"
 
 	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
+	ackerr "github.com/aws-controllers-k8s/runtime/pkg/errors"
 	ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
 
 	svcapitypes "github.com/aws-controllers-k8s/rds-controller/apis/v1alpha1"
@@ -27,10 +30,34 @@ import (
 )
 
 const (
+	applyTypeStatic        = "static"
 	sourceUser             = "user"
 	maxResetParametersSize = 20
 )
 
+var (
+	errUnknownParameter      = fmt.Errorf("unknown parameter")
+	errUnmodifiableParameter = fmt.Errorf("parameter is not modifiable")
+)
+
+func newErrUnknownParameter(name string) error {
+	// This is a terminal error because unless the user removes this parameter
+	// from their list of parameter overrides, we will not be able to get the
+	// resource into a synced state.
+	return ackerr.NewTerminalError(
+		fmt.Errorf("%w: %s", errUnknownParameter, name),
+	)
+}
+
+func newErrUnmodifiableParameter(name string) error {
+	// This is a terminal error because unless the user removes this parameter
+	// from their list of parameter overrides, we will not be able to get the
+	// resource into a synced state.
+	return ackerr.NewTerminalError(
+		fmt.Errorf("%w: %s", errUnmodifiableParameter, name),
+	)
+}
+
 // customUpdate is required to fix
 // https://github.com/aws-controllers-k8s/community/issues/869.
 //
@@ -206,7 +233,8 @@ func (rm *resourceManager) syncParameters(
 	exit := rlog.Trace("rm.syncParameters")
 	defer func() { exit(err) }()
 
-	groupName := (*string)(latest.ko.Spec.Name)
+	groupName := latest.ko.Spec.Name
+	family := latest.ko.Spec.Family
 
 	toModify, toDelete := computeParametersDelta(
 		desired.ko.Spec.ParameterOverrides, latest.ko.Spec.ParameterOverrides,
@@ -219,7 +247,8 @@ func (rm *resourceManager) syncParameters(
 	if len(toDelete) > 0 {
 		chunks := sliceStringChunks(toDelete, maxResetParametersSize)
 		for _, chunk := range chunks {
-			if err = rm.resetParameters(ctx, groupName, chunk); err != nil {
+			err = rm.resetParameters(ctx, family, groupName, chunk)
+			if err != nil {
 				return err
 			}
 		}
@@ -228,7 +257,8 @@ func (rm *resourceManager) syncParameters(
 	if len(toModify) > 0 {
 		chunks := mapStringChunks(toModify, maxResetParametersSize)
 		for _, chunk := range chunks {
-			if err = rm.modifyParameters(ctx, groupName, chunk); err != nil {
+			err = rm.modifyParameters(ctx, family, groupName, chunk)
+			if err != nil {
 				return err
 			}
 		}
@@ -283,20 +313,37 @@ func (rm *resourceManager) getParameters(
 // no more than 20 parameters to reset.
 func (rm *resourceManager) resetParameters(
 	ctx context.Context,
+	family *string,
 	groupName *string,
 	toDelete []string,
 ) (err error) {
 	rlog := ackrtlog.FromContext(ctx)
 	exit := rlog.Trace("rm.resetParameters")
 	defer func() { exit(err) }()
 
+	var pMeta *paramMeta
 	inputParams := []*svcsdk.Parameter{}
 	for _, paramName := range toDelete {
+		// default to this if something goes wrong looking up parameter
+		// defaults
+		applyMethod := svcsdk.ApplyMethodImmediate
+		pMeta, err = cachedParamMeta.get(
+			ctx, *family, paramName, rm.getFamilyParameters,
+		)
+		if err != nil {
+			return err
+		}
+		if !pMeta.isModifiable {
+			return newErrUnmodifiableParameter(paramName)
+		}
+		if !pMeta.isDynamic {
+			applyMethod = svcsdk.ApplyMethodPendingReboot
+		}
 		p := &svcsdk.Parameter{
 			ParameterName: aws.String(paramName),
 			// TODO(jaypipes): Look up appropriate apply method for this
 			// parameter...
-			ApplyMethod: aws.String(svcsdk.ApplyMethodImmediate),
+			ApplyMethod: aws.String(applyMethod),
 		}
 		inputParams = append(inputParams, p)
 	}
@@ -323,21 +370,38 @@ func (rm *resourceManager) resetParameters(
 // no more than 20 parameters to modify.
 func (rm *resourceManager) modifyParameters(
 	ctx context.Context,
+	family *string,
 	groupName *string,
 	toModify map[string]*string,
 ) (err error) {
 	rlog := ackrtlog.FromContext(ctx)
 	exit := rlog.Trace("rm.resetParameters")
 	defer func() { exit(err) }()
 
+	var pMeta *paramMeta
 	inputParams := []*svcsdk.Parameter{}
 	for paramName, paramValue := range toModify {
+		// default to this if something goes wrong looking up parameter
+		// defaults
+		applyMethod := svcsdk.ApplyMethodImmediate
+		pMeta, err = cachedParamMeta.get(
+			ctx, *family, paramName, rm.getFamilyParameters,
+		)
+		if err != nil {
+			return err
+		}
+		if !pMeta.isModifiable {
+			return newErrUnmodifiableParameter(paramName)
+		}
+		if !pMeta.isDynamic {
+			applyMethod = svcsdk.ApplyMethodPendingReboot
+		}
 		p := &svcsdk.Parameter{
 			ParameterName:  aws.String(paramName),
 			ParameterValue: paramValue,
 			// TODO(jaypipes): Look up appropriate apply method for this
 			// parameter...
-			ApplyMethod: aws.String(svcsdk.ApplyMethodImmediate),
+			ApplyMethod: aws.String(applyMethod),
 		}
 		inputParams = append(inputParams, p)
 	}
@@ -360,6 +424,42 @@ func (rm *resourceManager) modifyParameters(
 	return nil
 }
 
+// getFamilyParameters calls the RDS DescribeEngineDefaultParameters API to
+// retrieve the set of parameter information for a DB parameter group family.
+func (rm *resourceManager) getFamilyParameters(
+	ctx context.Context,
+	family string,
+) (map[string]paramMeta, error) {
+	var marker *string
+	familyMeta := map[string]paramMeta{}
+
+	for {
+		resp, err := rm.sdkapi.DescribeEngineDefaultParametersWithContext(
+			ctx,
+			&svcsdk.DescribeEngineDefaultParametersInput{
+				DBParameterGroupFamily: aws.String(family),
+				Marker:                 marker,
+			},
+		)
+		rm.metrics.RecordAPICall("GET", "DescribeEngineDefaultParameters", err)
+		if err != nil {
+			return nil, err
+		}
+		for _, param := range resp.EngineDefaults.Parameters {
+			pName := *param.ParameterName
+			familyMeta[pName] = paramMeta{
+				isModifiable: *param.IsModifiable,
+				isDynamic:    *param.ApplyType != applyTypeStatic,
+			}
+		}
+		marker = resp.EngineDefaults.Marker
+		if marker == nil {
+			break
+		}
+	}
+	return familyMeta, nil
+}
+
 // computeParametersDelta compares two Parameter arrays and returns the new
 // parameters to add, to update and the parameter identifiers to delete
 func computeParametersDelta(
@@ -431,3 +531,79 @@ func mapStringChunks(
 
 	return chunks
 }
+
+type paramMeta struct {
+	isModifiable bool
+	isDynamic    bool
+}
+
+// metaFetcher is the functor we pass to the paramMetaCache that allows it to
+// fetch engine default parameter information
+type metaFetcher func(ctx context.Context, family string) (map[string]paramMeta, error)
+
+// paramMetaCache stores information about a parameter for a DB parameter group
+// family. We use this cached information to determine whether a parameter is
+// statically or dynamically defined (whether changes can be applied
+// immediately or pending a reboot) and whether a parameter is modifiable.
+//
+// Keeping things super simple for now and not adding any TTL or expiration
+// behaviour to the cache. Engine defaults are pretty static information...
+type paramMetaCache struct {
+	sync.RWMutex
+	hits   uint64
+	misses uint64
+	cache  map[string]map[string]paramMeta
+}
+
+// get retrieves the metadata for a named parameter group family and parameter
+// name.
+func (c *paramMetaCache) get(
+	ctx context.Context,
+	family string,
+	name string,
+	fetcher metaFetcher,
+) (*paramMeta, error) {
+	c.RLock()
+	defer c.RUnlock()
+
+	var err error
+	var found bool
+	var metas map[string]paramMeta
+	var meta paramMeta
+
+	metas, found = c.cache[family]
+	if !found {
+		c.misses++
+		metas, err = c.loadFamily(ctx, family, fetcher)
+		if err != nil {
+			return nil, err
+		}
+	}
+	meta, found = metas[name]
+	if !found {
+		return nil, newErrUnknownParameter(name)
+	}
+	c.hits++
+	return &meta, nil
+}
+
+// loadFamily fetches parameter information from the AWS RDS
+// DescribeEngineDefaultParameters API and caches that information.
+func (c *paramMetaCache) loadFamily(
+	ctx context.Context,
+	family string,
+	fetcher metaFetcher,
+) (map[string]paramMeta, error) {
+	familyMeta, err := fetcher(ctx, family)
+	if err != nil {
+		return nil, err
+	}
+	c.Lock()
+	defer c.Unlock()
+	c.cache[family] = familyMeta
+	return familyMeta, nil
+}
+
+var cachedParamMeta = paramMetaCache{
+	cache: map[string]map[string]paramMeta{},
+}