diff --git a/mmv1/third_party/terraform/acctest/resource_inventory_reader.go.tmpl b/mmv1/third_party/terraform/acctest/resource_inventory_reader.go.tmpl
index fc5926098a06..3db0f7aec236 100644
--- a/mmv1/third_party/terraform/acctest/resource_inventory_reader.go.tmpl
+++ b/mmv1/third_party/terraform/acctest/resource_inventory_reader.go.tmpl
@@ -4,56 +4,67 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"sync"
 
 	"gopkg.in/yaml.v2"
 )
 
-// ResourceYAMLMetadata represents the structure of the metadata files
-type ResourceYAMLMetadata struct {
-	Resource           string `yaml:"resource"`
-	ApiServiceName     string `yaml:"api_service_name"`
-	CaiAssetNameFormat string `yaml:"cai_asset_name_format"`
-	SourceFile         string `yaml:"source_file"`
-}
-
-// Cache structures to avoid repeated file system operations
 var (
-	// Cache for API service names (resourceName -> apiServiceName)
-	ApiServiceNameCache = NewGenericCache("unknown")
-	// Cache for CAI resource name format (resourceName -> CaiAssetNameFormat)
-	CaiAssetNameFormatCache = NewGenericCache("")
-	// Cache for service packages (resourceType -> servicePackage)
-	ServicePackageCache = NewGenericCache("unknown")
-	// Flag to track if cache has been populated
-	cachePopulated = false
-	// Mutex to protect cache access
-	cacheMutex sync.RWMutex
-
-	iamSuffixes = []string{
-		"_iam_member",
-		"_iam_binding",
-		"_iam_policy",
+	// The GlobalMetadataCache is used by VCR tests to avoid loading metadata once per test run.
+	// Because of the way VCR tests are run, it's difficult to avoid a global variable.
+	GlobalMetadataCache = MetadataCache{
+		mutex: &sync.Mutex{},
 	}
-
 	{{ if eq $.TargetVersionName `ga` -}}
 	providerName = "google"
 	{{- else }}
 	providerName = "google-beta"
 	{{- end }}
-
 )
 
-// PopulateMetadataCache walks through all metadata files once and populates
-// both the API service name and service package caches for improved performance
-func PopulateMetadataCache() error {
-	cacheMutex.Lock()
-	defer cacheMutex.Unlock()
+// Metadata represents the structure of the metadata files
+type Metadata struct {
+	Resource            string          `yaml:"resource"`
+	GenerationType      string          `yaml:"generation_type"`
+	SourceFile          string          `yaml:"source_file"`
+	ApiServiceName      string          `yaml:"api_service_name"`
+	ApiVersion          string          `yaml:"api_version"`
+	ApiResourceTypeKind string          `yaml:"api_resource_type_kind"`
+	CaiAssetNameFormat  string          `yaml:"cai_asset_name_format"`
+	ApiVariantPatterns  []string        `yaml:"api_variant_patterns"`
+	AutogenStatus       bool            `yaml:"autogen_status,omitempty"`
+	Fields              []MetadataField `yaml:"fields"`
+
+	// These keys store information about the metadata file itself.
+
+	// Path is the path of the loaded metadata file
+	Path string
+	// ServicePackage is the folder within services/ that the metadata file is in
+	ServicePackage string
+}
 
-	// If cache is already populated, we can skip
-	if cachePopulated {
-		return nil
+type MetadataField struct {
+	ApiField     string `yaml:"api_field"`
+	Field        string `yaml:"field"`
+	ProviderOnly string `yaml:"provider_only"`
+	Json         string `yaml:"json"`
+}
+
+type MetadataCache struct {
+	mutex          *sync.Mutex
+	cache          map[string]Metadata
+	populated      bool
+	populatedError error
+}
+
+func (mc *MetadataCache) Populate() error {
+	mc.mutex.Lock()
+	defer mc.mutex.Unlock()
+
+	if mc.populated {
+		return mc.populatedError
 	}
 
 	baseDir, err := getServicesDir()
@@ -61,124 +72,86 @@ func PopulateMetadataCache() error {
 		return fmt.Errorf("failed to find services directory: %v", err)
 	}
 
-	// Count for statistics
-	apiNameCount := 0
-	servicePkgCount := 0
+	mc.cache = make(map[string]Metadata)
+
+	var malformed_yaml_errs []string
 
 	// Walk through all service directories once
 	err = filepath.Walk(baseDir, func(path string, info os.FileInfo, err error) error {
 		if err != nil {
-			return nil // Skip files with errors but continue walking
+			return err // Fail immediately if there's an OS error.
+		}
+
+		// Skip non-metadata files
+		if info.IsDir() || !strings.HasPrefix(info.Name(), "resource_") || !strings.HasSuffix(info.Name(), "_meta.yaml") {
+			return nil
+		}
+
+		// Read the file
+		content, err := os.ReadFile(path)
+		if err != nil {
+			return err // Fail immediately if there's an OS error.
 		}
 
-		// Look for metadata files
-		if !info.IsDir() && strings.HasPrefix(info.Name(), "resource_") && strings.HasSuffix(info.Name(), "_meta.yaml") {
-			// Read the file
-			content, err := os.ReadFile(path)
-			if err != nil {
-				return nil // Continue to next file
-			}
-
-			// Parse YAML
-			var metadata ResourceYAMLMetadata
-			if err := yaml.Unmarshal(content, &metadata); err != nil {
-				return nil // Continue to next file
-			}
-
-			// Skip if resource is empty
-			if metadata.Resource == "" {
-				return nil
-			}
-
-			iamResources := make([]string, 0)
-			for _, suffix := range iamSuffixes {
-				iamResources = append(iamResources, fmt.Sprintf("%s%s", metadata.Resource, suffix))
-			}
-
-			// Store API service name in cache
-			if metadata.ApiServiceName != "" {
-				ApiServiceNameCache.Set(metadata.Resource, metadata.ApiServiceName)
-				for _, iamResource := range iamResources {
-					ApiServiceNameCache.Set(iamResource, metadata.ApiServiceName)
-				}
-				apiNameCount++
-			}
-
-			if metadata.CaiAssetNameFormat != "" {
-				CaiAssetNameFormatCache.Set(metadata.Resource, metadata.CaiAssetNameFormat)
-				for _, iamResource := range iamResources {
-					CaiAssetNameFormatCache.Set(iamResource, metadata.CaiAssetNameFormat)
-				}
-			}
-
-			// Extract and store service package in cache
-			pathParts := strings.Split(path, string(os.PathSeparator))
-			servicesIndex := -1
-			for i, part := range pathParts {
-				if part == "services" {
-					servicesIndex = i
-					break
-				}
-			}
-
-			if servicesIndex >= 0 && len(pathParts) > servicesIndex+1 {
-				servicePackage := pathParts[servicesIndex+1] // The part after "services"
-				ServicePackageCache.Set(metadata.Resource, servicePackage)
-				for _, iamResource := range iamResources {
-					ServicePackageCache.Set(iamResource, servicePackage)
-				}
-				servicePkgCount++
-			}
+		// Parse YAML
+		var metadata Metadata
+		if err := yaml.Unmarshal(content, &metadata); err != nil {
+			// note but keep walking
+			malformed_yaml_errs = append(malformed_yaml_errs, fmt.Sprintf("%s: %v", path, err.Error()))
+			return nil
 		}
+
+		// Skip if resource is empty
+		if metadata.Resource == "" {
+			return nil
+		}
+
+		if _, ok := mc.cache[metadata.Resource]; ok {
+			return fmt.Errorf("duplicate resource: %s in %s", metadata.Resource, path)
+		}
+
+		metadata.Path = path
+		pathParts := strings.Split(path, string(os.PathSeparator))
+		servicesIndex := slices.Index(pathParts, "services")
+		if servicesIndex == -1 {
+			return fmt.Errorf("no service found for %s (%s)", metadata.Resource, path)
+		}
+		metadata.ServicePackage = pathParts[servicesIndex+1]
+		mc.cache[metadata.Resource] = metadata
 		return nil
 	})
 
 	if err != nil {
-		return fmt.Errorf("error walking directory: %v", err)
+		mc.populatedError = fmt.Errorf("error walking directory: %v", err)
 	}
 
-	// Mark cache as populated
-	cachePopulated = true
-
-	return nil
-}
+	if len(malformed_yaml_errs) > 0 {
+		mc.populatedError = fmt.Errorf("YAML parsing errors encountered:\n%v", strings.Join(malformed_yaml_errs, "\n"))
+	}
 
-type GenericCache struct {
-	mu           sync.RWMutex
-	data         map[string]string
-	defaultValue string
-}
+	// Mark cache as populated
+	mc.populated = true
 
-// NewGenericCache initializes a new GenericCache with a default value.
-func NewGenericCache(defaultValue string) *GenericCache {
-	return &GenericCache{
-		data:         make(map[string]string),
-		defaultValue: defaultValue,
-	}
+	return mc.populatedError
 }
 
-// Get retrieves a value from the cache, returning the default if not found.
-func (c *GenericCache) Get(key string) string {
-	// Make sure cache is populated
-	if !cachePopulated {
-		if err := PopulateMetadataCache(); err != nil {
-			return "failed_to_populate_metadata_cache"
-		}
-	}
-
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	value, ok := c.data[key]
-	if !ok {
-		return c.defaultValue
-	}
-	return value
+// Get takes a resource name (like google_compute_instance) and returns
+// the metadata for that resource and an `ok` bool of whether the metadata
+// exisxts in the cache.
+func (mc *MetadataCache) Get(key string) (Metadata, bool) {
+	// For IAM resources, return the parent resource's metadata. This could change if we
+	// start generating separate IAM metadata in the future. This is primarily for
+	// backwards-compatibility with the previous cache behavior and a workaround for _not_
+	// generating IAM resource metadata.
+	key, _ = strings.CutSuffix(key, "_iam_member")
+	key, _ = strings.CutSuffix(key, "_iam_binding")
+	key, _ = strings.CutSuffix(key, "_iam_policy")
+	m, ok := mc.cache[key]
+	return m, ok
 }
 
-func (c *GenericCache) Set(key, value string) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.data[key] = value
+func (mc *MetadataCache) Cache() map[string]Metadata {
+	return mc.cache
 }
 
 // getServicesDir returns the path to the services directory
diff --git a/mmv1/third_party/terraform/acctest/resource_inventory_test.go b/mmv1/third_party/terraform/acctest/resource_inventory_test.go
index afdf1eee3307..0defd4fd11f0 100644
--- a/mmv1/third_party/terraform/acctest/resource_inventory_test.go
+++ b/mmv1/third_party/terraform/acctest/resource_inventory_test.go
@@ -1,113 +1,242 @@
 package acctest_test
 
 import (
+	"context"
+	"fmt"
+	"maps"
+	"regexp"
+	"slices"
 	"strings"
 	"testing"
 
+	fwattr "github.com/hashicorp/terraform-plugin-framework/attr"
+	"github.com/hashicorp/terraform-plugin-framework/resource"
+	fwschema "github.com/hashicorp/terraform-plugin-framework/resource/schema"
+	fwtypes "github.com/hashicorp/terraform-plugin-framework/types"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-provider-google/google/acctest"
+	"github.com/hashicorp/terraform-provider-google/google/fwprovider"
 	"github.com/hashicorp/terraform-provider-google/google/provider"
 )
 
-func TestResourceInventoryMetadataFound(t *testing.T) {
-	resources := provider.ResourceMap()
-
-	// Track statistics
-	var (
-		totalResources     = 0
-		missingServicePkg  = 0
-		missingServiceName = 0
-	)
-
-	// Create a map to store missing resources for summary report
-	missingServicePkgResources := make(map[string]bool)
-	missingServiceNameResources := make(map[string]bool)
+type ProviderResource struct {
+	Name   string
+	Fields map[string]bool
+}
 
-	for resourceType := range resources {
-		if strings.HasSuffix(resourceType, "_iam_member") ||
-			strings.HasSuffix(resourceType, "_iam_policy") ||
-			strings.HasSuffix(resourceType, "_iam_binding") {
+func buildProviderResources(pluginSdk map[string]*schema.Resource, pluginFramework []func() resource.Resource) map[string]ProviderResource {
+	ctx := context.Background()
+	resources := map[string]ProviderResource{}
+	// terraform-plugin-sdk-v2
+	for resourceName, resource := range pluginSdk {
+		// We don't care about IAM resources because they don't have their own meta.yaml files
+		if acctest.IsIamResource(resourceName) {
 			continue
 		}
-		totalResources++
-
-		// Log each resource being checked
-		// t.Logf("Checking metadata for resource: %s", resourceType)
-
-		// Check for service package
-		servicePackage := acctest.ServicePackageCache.Get(resourceType)
-		if servicePackage == "unknown" {
-			// t.Logf("WARNING: Could not find service package for resource %s: %v", resourceType)
-			missingServicePkg++
-			missingServicePkgResources[resourceType] = true
+		r := ProviderResource{
+			Name:   resourceName,
+			Fields: map[string]bool{},
+		}
+		flattened := flattenSchema("", resource.Schema)
+		for f := range flattened {
+			r.Fields[f] = true
+		}
+		resources[resourceName] = r
+	}
+	// plugin-framework
+	for _, f := range pluginFramework {
+		fwResource := f()
+		metadataReq := resource.MetadataRequest{ProviderTypeName: "google"}
+		metadataResp := &resource.MetadataResponse{}
+		fwResource.Metadata(ctx, metadataReq, metadataResp)
+
+		resourceName := metadataResp.TypeName
+		r := ProviderResource{
+			Name:   resourceName,
+			Fields: map[string]bool{},
 		}
 
-		apiServiceName := acctest.ApiServiceNameCache.Get(resourceType)
-		// Check for API service name
-		if apiServiceName == "unknown" {
-			// t.Logf("WARNING: Could not find API service name for resource %s: %v", resourceType)
-			missingServiceName++
-			missingServiceNameResources[resourceType] = true
+		schemaReq := resource.SchemaRequest{}
+		schemaResp := &resource.SchemaResponse{}
+		fwResource.Schema(ctx, schemaReq, schemaResp)
+		flattened := flattenPluginFrameworkSchema(schemaResp.Schema.Attributes)
+		for f, _ := range flattened {
+			r.Fields[f] = true
 		}
-		t.Logf(" %s servicePackage: %s apiServiceName: %s", resourceType, servicePackage, apiServiceName)
+		resources[resourceName] = r
+	}
+	return resources
+}
+
+// from mmv1/tools/diff-processor/diff - modified to only include leaf fields
+func flattenSchema(parentKey string, schemaObj map[string]*schema.Schema) map[string]*schema.Schema {
+	flattened := make(map[string]*schema.Schema)
 
+	if parentKey != "" {
+		parentKey += "."
 	}
 
-	// Generate a summary report
-	t.Logf("\n--- RESOURCE METADATA TEST SUMMARY ---")
-	t.Logf("Total resources checked: %d", totalResources)
-	t.Logf("Resources missing service package: %d (%.1f%%)",
-		missingServicePkg,
-		float64(missingServicePkg)/float64(totalResources)*100)
-	t.Logf("Resources missing API service name: %d (%.1f%%)",
-		missingServiceName,
-		float64(missingServiceName)/float64(totalResources)*100)
-
-	// List resources missing metadata (limited to first 10 for readability)
-	if len(missingServicePkgResources) > 0 {
-		t.Log("\nResources missing service package (first 10):")
-		count := 0
-		for res := range missingServicePkgResources {
-			t.Logf("  - %s", res)
-			count++
-			if count >= 10 {
-				remaining := len(missingServicePkgResources) - 10
-				if remaining > 0 {
-					t.Logf("  ... and %d more", remaining)
-				}
-				break
+	for fieldName, field := range schemaObj {
+		key := parentKey + fieldName
+		childResource, hasNestedFields := field.Elem.(*schema.Resource)
+		if field.Elem != nil && hasNestedFields && len(childResource.Schema) > 0 {
+			for childKey, childField := range flattenSchema(key, childResource.Schema) {
+				flattened[childKey] = childField
 			}
+		} else {
+			flattened[key] = field
 		}
 	}
 
-	if len(missingServiceNameResources) > 0 {
-		t.Log("\nResources missing API service name (first 10):")
-		count := 0
-		for res := range missingServiceNameResources {
-			t.Logf("  - %s", res)
-			count++
-			if count >= 10 {
-				remaining := len(missingServiceNameResources) - 10
-				if remaining > 0 {
-					t.Logf("  ... and %d more", remaining)
+	return flattened
+}
+
+// This is overly simplified - we only have two plugin framework resources so we don't need to handle everything
+func flattenPluginFrameworkTypes(parentKey string, attrTypes map[string]fwattr.Type) map[string]bool {
+	flattened := make(map[string]bool)
+	parentKey += "."
+	for fieldName := range attrTypes {
+		key := parentKey + fieldName
+		flattened[key] = true
+	}
+	return flattened
+}
+
+// This is overly simplified - we only have two plugin framework resources so we don't need to handle everything
+func flattenPluginFrameworkSchema(schemaObj map[string]fwschema.Attribute) map[string]bool {
+	flattened := make(map[string]bool)
+	for fieldName, attribute := range schemaObj {
+		key := fieldName
+		if l, ok := attribute.(fwschema.ListAttribute); ok {
+			if child, ok := l.ElementType.(fwtypes.ObjectType); ok {
+				for childKey, _ := range flattenPluginFrameworkTypes(key, child.AttrTypes) {
+					flattened[childKey] = true
 				}
-				break
 			}
+		} else {
+			flattened[key] = true
 		}
 	}
+	return flattened
+}
 
-	// Decide whether to fail the test based on coverage percentage
-	const requiredCoveragePercent = 90.0
+// set of google_resource.field.path fields to ignore during analysis
+var ignoredFields = map[string]bool{
+	// These fields are ignored because they're client-side only fields added with
+	// extra_schema_entry. Shouldn't impact coverage.
+	"google_api_gateway_api_config.api_config_id_prefix":                  true,
+	"google_bigquery_dataset_access.api_updated_member":                   true,
+	"google_bigtable_app_profile.multi_cluster_routing_cluster_ids":       true,
+	"google_bigtable_app_profile.row_affinity":                            true,
+	"google_compute_firewall.enable_logging":                              true,
+	"google_compute_region_ssl_certificate.name_prefix":                   true,
+	"google_compute_route.next_hop_instance_zone":                         true,
+	"google_compute_ssl_certificate.name_prefix":                          true,
+	"google_compute_subnetwork.fingerprint":                               true,
+	"google_redis_instance.auth_string":                                   true,
+	"google_secret_manager_regional_secret_version.is_secret_data_base64": true,
+	"google_secret_manager_secret_version.is_secret_data_base64":          true,
+	"google_sourcerepo_repository.create_ignore_already_exists":           true,
+	"google_vertex_ai_featurestore_entitytype.region":                     true,
+	"google_vertex_ai_featurestore_entitytype_feature.region":             true,
+	"google_workflows_workflow.name_prefix":                               true,
+
+	// query / URL params - need to decide how to handle, but don't impact coverage
+	"google_compute_instance_from_template.source_instance_template": true,
+	"google_bigquery_table.table_metadata_view":                      true,
+	"google_container_registry.bucket_self_link":                     true,
+	"google_container_registry.location":                             true,
+
+	// Something weird about this one, probably related to flatten_object usage.
+	"google_data_catalog_tag.fields.enum_value": true,
+}
 
-	servicePkgCoverage := (float64(totalResources-missingServicePkg) / float64(totalResources)) * 100
-	serviceNameCoverage := (float64(totalResources-missingServiceName) / float64(totalResources)) * 100
+func ignoreField(r, f string) bool {
+	if f == "project" {
+		return true
+	}
+	if ignoredFields[fmt.Sprintf("%s.%s", r, f)] {
+		return true
+	}
+	return false
+}
 
-	if servicePkgCoverage < requiredCoveragePercent {
-		t.Errorf("Service package metadata coverage (%.1f%%) is below required threshold (%.1f%%)",
-			servicePkgCoverage, requiredCoveragePercent)
+// Slightly modified from mmv1/google/string_utils.go
+func underscore(source string) string {
+	tmp := regexp.MustCompile(`([A-Z]+)([A-Z][a-z])`).ReplaceAllString(source, "${1}_${2}")
+	tmp = regexp.MustCompile(`([a-z\d])([A-Z])`).ReplaceAllString(tmp, "${1}_${2}")
+	tmp = strings.Replace(tmp, "-", "_", 1)
+	// skip this because we want to operate on nested api fields
+	// tmp = strings.Replace(tmp, ".", "_", 1)
+	tmp = strings.ToLower(tmp)
+	return tmp
+}
+
+func TestValidateResourceMetadata(t *testing.T) {
+	err := acctest.GlobalMetadataCache.Populate()
+	if err != nil {
+		t.Fatalf("Failed to populate metadata cache: %v", err)
+	}
+
+	ctx := context.Background()
+	resources := buildProviderResources(provider.ResourceMap(), fwprovider.New(provider.Provider()).Resources(ctx))
+	metaResources := acctest.GlobalMetadataCache.Cache()
+
+	// Check for resources that are only in the provider / only in metadata
+	sortedResourceNames := slices.Sorted(maps.Keys(resources))
+	for _, resourceName := range sortedResourceNames {
+		if _, ok := metaResources[resourceName]; !ok {
+			t.Errorf("Resource %q has no meta.yaml file", resourceName)
+		}
+	}
+	for _, resourceName := range slices.Sorted(maps.Keys(metaResources)) {
+		if _, ok := resources[resourceName]; !ok {
+			t.Errorf("Resource %q has meta.yaml file but isn't present in the provider", resourceName)
+		}
+	}
+
+	// Check for fields that are in the provider but not in metadata (or vice versa)
+	for _, resourceName := range sortedResourceNames {
+		r := resources[resourceName]
+		m := metaResources[resourceName]
+		mFields := map[string]bool{}
+		for _, f := range m.Fields {
+			terraformField := f.Field
+			if terraformField == "" {
+				terraformField = underscore(f.ApiField)
+			}
+			mFields[terraformField] = true
+		}
+		for _, f := range slices.Sorted(maps.Keys(r.Fields)) {
+			if ignoreField(r.Name, f) {
+				continue
+			}
+			if _, ok := mFields[f]; !ok {
+				t.Errorf("Field in provider resource; missing in meta.yaml: %s.%s", r.Name, f)
+			}
+		}
+		for f, _ := range mFields {
+			if ignoreField(r.Name, f) {
+				continue
+			}
+			if _, ok := r.Fields[f]; !ok {
+				t.Errorf("Field in meta.yaml; missing in provider resource: %s.%s", r.Name, f)
+			}
+		}
 	}
 
-	if serviceNameCoverage < requiredCoveragePercent {
-		t.Errorf("API service name metadata coverage (%.1f%%) is below required threshold (%.1f%%)",
-			serviceNameCoverage, requiredCoveragePercent)
+	// Validate yaml files
+	for resourceName, m := range metaResources {
+		if m.ServicePackage == "" {
+			t.Errorf("%s isn't in a service package", resourceName)
+		}
+
+		if m.ApiServiceName == "" {
+			// Allowlist google_container_registry because it doesn't clearly correspond to a service
+			if resourceName != "google_container_registry" {
+				t.Errorf("%s is missing `api_service_name`", m.Path)
+			}
+		}
+
 	}
 }
diff --git a/mmv1/third_party/terraform/acctest/tgc_utils.go b/mmv1/third_party/terraform/acctest/tgc_utils.go
index 5b27f962dbee..558963a1111c 100644
--- a/mmv1/third_party/terraform/acctest/tgc_utils.go
+++ b/mmv1/third_party/terraform/acctest/tgc_utils.go
@@ -105,16 +105,20 @@ func CollectAllTgcMetadata(tgcPayload TgcMetadataPayload) resource.TestCheckFunc
 				continue
 			}
 
-			// Resolve the CAI asset name
-			caiAssetNameFormat := CaiAssetNameFormatCache.Get(metadata.ResourceType)
-			if caiAssetNameFormat == "" || caiAssetNameFormat == "failed_to_populate_metadata_cache" {
-				log.Printf("[DEBUG]TGC Terraform error: unknown CAI asset name format for resource type %s", caiAssetNameFormat)
-
-				apiServiceName := ApiServiceNameCache.Get(metadata.ResourceType)
-				if apiServiceName == "unknown" || apiServiceName == "failed_to_populate_metadata_cache" {
+			// Technically this will always be populated before this point but keeping it here for now so
+			// the implementations are more similar.
+			err := GlobalMetadataCache.Populate()
+			if err != nil {
+				log.Printf("[DEBUG] TGC Terraform error: couldn't populate metadata cache: %v", err)
+				metadata.CaiAssetNames = []string{"failed_to_populate_metadata_cache"}
+			} else {
+				yamlMetadata, ok := GlobalMetadataCache.Get(metadata.ResourceType)
+				if !ok {
 					log.Printf("[DEBUG]TGC Terraform error: unknown resource type %s", metadata.ResourceType)
-					metadata.CaiAssetNames = []string{apiServiceName}
-				} else {
+					metadata.CaiAssetNames = []string{"unknown"}
+				} else if yamlMetadata.CaiAssetNameFormat == "" {
+					log.Printf("[DEBUG]TGC Terraform error: unknown CAI asset name format for resource type %s", yamlMetadata.CaiAssetNameFormat)
+
 					var rName string
 					switch metadata.ResourceType {
 					case "google_project":
@@ -131,23 +135,23 @@ func CollectAllTgcMetadata(tgcPayload TgcMetadataPayload) resource.TestCheckFunc
 						rName = getIamResourceId(metadata.ResourceType, rName)
 					}
 
-					metadata.CaiAssetNames = []string{fmt.Sprintf("//%s/%s", apiServiceName, rName)}
-				}
-			} else {
-				paramsMap := make(map[string]any, 0)
-				params := extractIdentifiers(caiAssetNameFormat)
-				for _, param := range params {
-					v := rState.Primary.Attributes[param]
-					paramsMap[param] = v
-				}
+					metadata.CaiAssetNames = []string{fmt.Sprintf("//%s/%s", yamlMetadata.ApiServiceName, rName)}
+				} else {
+					paramsMap := make(map[string]any, 0)
+					params := extractIdentifiers(yamlMetadata.CaiAssetNameFormat)
+					for _, param := range params {
+						v := rState.Primary.Attributes[param]
+						paramsMap[param] = v
+					}
 
-				caiAssetName := replacePlaceholders(caiAssetNameFormat, paramsMap)
+					caiAssetName := replacePlaceholders(yamlMetadata.CaiAssetNameFormat, paramsMap)
 
-				if _, ok := serviceWithProjectNumber[metadata.Service]; ok {
-					caiAssetName = strings.Replace(caiAssetName, projectId, projectNumber, 1)
-				}
+					if _, ok := serviceWithProjectNumber[metadata.Service]; ok {
+						caiAssetName = strings.Replace(caiAssetName, projectId, projectNumber, 1)
+					}
 
-				metadata.CaiAssetNames = []string{caiAssetName}
+					metadata.CaiAssetNames = []string{caiAssetName}
+				}
 			}
 
 			log.Printf("[DEBUG] CaiAssetNames %#v", metadata.CaiAssetNames)
@@ -273,6 +277,11 @@ func determineImportMetadata(steps []resource.TestStep, currentStepIndex int, re
 func extendWithTGCData(t *testing.T, c resource.TestCase) resource.TestCase {
 	var updatedSteps []resource.TestStep
 
+	populateErr := GlobalMetadataCache.Populate()
+	if populateErr != nil {
+		log.Printf("[DEBUG] TGC Terraform error: couldn't populate metadata cache: %v", populateErr)
+	}
+
 	// Process all steps
 	for i, step := range c.Steps {
 		// If this is a non-plan config step, add our TGC check
@@ -313,12 +322,19 @@ func extendWithTGCData(t *testing.T, c resource.TestCase) resource.TestCase {
 					// Determine import metadata if the next step is an import step
 					importMeta := determineImportMetadata(c.Steps, i, res)
 
+					service := "unknown"
+					if populateErr != nil {
+						service = "failed_to_populate_metadata_cache"
+					} else if yamlMetadata, ok := GlobalMetadataCache.Get(resourceType); ok {
+						service = yamlMetadata.ServicePackage
+					}
+
 					// Create metadata for this resource
 					resourceMetadata[res] = ResourceMetadata{
 						ResourceType:    resourceType,
 						ResourceAddress: res,
 						ImportMetadata:  importMeta,
-						Service:         ServicePackageCache.Get(resourceType),
+						Service:         service,
 						// CaiAssetNames will be populated at runtime in the check function
 					}
 				}
@@ -360,6 +376,13 @@ func getIamResourceId(resourceType, id string) string {
 	return id
 }
 
+var iamSuffixes = []string{
+	"_iam_member",
+	"_iam_policy",
+	"_iam_binding",
+	"_iam_audit_config",
+}
+
 // Checks if a resource is an IAM resource
 func IsIamResource(resourceType string) bool {
 	for _, suffix := range iamSuffixes {
diff --git a/mmv1/third_party/terraform/services/container/resource_container_cluster_meta.yaml.tmpl b/mmv1/third_party/terraform/services/container/resource_container_cluster_meta.yaml.tmpl
index 7b909c8137f8..d89b0bfaca5b 100644
--- a/mmv1/third_party/terraform/services/container/resource_container_cluster_meta.yaml.tmpl
+++ b/mmv1/third_party/terraform/services/container/resource_container_cluster_meta.yaml.tmpl
@@ -254,14 +254,14 @@ fields:
   - api_field: 'nodeConfig.containerdConfig.writableCgroups.enabled'
   - api_field: 'nodeConfig.containerdConfig.registryHosts.server'
   - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.host'
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.capabilities
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.overridePath 
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.header.key
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.header.value
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.ca.gcpSecretManagerSecretUri
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.client.cert.gcpSecretManagerSecretUri
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.client.key.gcpSecretManagerSecretUri
-  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.dialTimeout
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.capabilities'
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.overridePath'
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.header.key'
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.header.value'
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.ca.gcpSecretManagerSecretUri'
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.client.cert.gcpSecretManagerSecretUri'
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.client.key.gcpSecretManagerSecretUri'
+  - api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.dialTimeout'
   - api_field: 'nodeConfig.diskSizeGb'
   - api_field: 'nodeConfig.diskType'
   - field: 'node_config.effective_taints.effect'
@@ -463,6 +463,26 @@ fields:
     api_field: 'nodePools.config.containerdConfig.privateRegistryAccessConfig.certificateAuthorityDomainConfig.gcpSecretManagerCertificateConfig.secretUri'
   - field: 'node_pool.node_config.containerd_config.private_registry_access_config.enabled'
     api_field: 'nodePools.config.containerdConfig.privateRegistryAccessConfig.enabled'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.ca.gcp_secret_manager_secret_uri'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.ca.gcpSecretManagerSecretUri'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.capabilities'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.capabilities'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.client.cert.gcp_secret_manager_secret_uri'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.client.cert.gcpSecretManagerSecretUri'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.client.key.gcp_secret_manager_secret_uri'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.client.key.gcpSecretManagerSecretUri'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.dial_timeout'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.dialTimeout'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.header.key'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.header.key'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.header.value'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.header.value'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.host'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.host'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.hosts.override_path'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.hosts.overridePath'
+  - field: 'node_pool.node_config.containerd_config.registry_hosts.server'
+    api_field: 'nodePools.config.containerdConfig.registryHosts.server'
   - field: 'node_pool.node_config.containerd_config.writable_cgroups.enabled'
     api_field: 'nodePools.config.containerdConfig.writableCgroups.enabled'
   - field: 'node_pool.node_config.disk_size_gb'
@@ -711,14 +731,14 @@ fields:
   - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.writableCgroups.enabled'
   - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.server'
   - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.host'
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.capabilities
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.overridePath 
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.header.key
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.header.value
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.ca.gcpSecretManagerSecretUri
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.client.cert.gcpSecretManagerSecretUri
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.client.key.gcpSecretManagerSecretUri
-  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.dialTimeout
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.capabilities'
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.overridePath'
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.header.key'
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.header.value'
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.ca.gcpSecretManagerSecretUri'
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.client.cert.gcpSecretManagerSecretUri'
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.client.key.gcpSecretManagerSecretUri'
+  - api_field: 'nodePoolDefaults.nodeConfigDefaults.containerdConfig.registryHosts.hosts.dialTimeout'
   - api_field: 'nodePoolDefaults.nodeConfigDefaults.gcfsConfig.enabled'
   - field: 'node_pool_defaults.node_config_defaults.insecure_kubelet_readonly_port_enabled'
     api_field: 'nodePoolDefaults.nodeConfigDefaults.nodeKubeletConfig.insecureKubeletReadonlyPortEnabled'
diff --git a/mmv1/third_party/terraform/services/container/resource_container_node_pool_meta.yaml.tmpl b/mmv1/third_party/terraform/services/container/resource_container_node_pool_meta.yaml.tmpl
index d040d7dc9ffe..94666bdc9821 100644
--- a/mmv1/third_party/terraform/services/container/resource_container_node_pool_meta.yaml.tmpl
+++ b/mmv1/third_party/terraform/services/container/resource_container_node_pool_meta.yaml.tmpl
@@ -69,25 +69,25 @@ fields:
   - field: 'node_config.containerd_config.writable_cgroups.enabled'
     api_field: 'config.containerdConfig.writableCgroups.enabled'
   - field: 'node_config.containerd_config.registry_hosts.server'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.server'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.host'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.host'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.capabilities'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.capabilities'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.override_path'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.overridePath'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.header.key'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.header.key'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.header.value'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.header.value'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.ca.gcp_secret_manager_secret_uri'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.ca.gcpSecretManagerSecretUri'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.client.cert.gcp_secret_manager_secret_uri'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.client.cert.gcpSecretManagerSecretUri'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts.client.key.gcp_secret_manager_secret_uri'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.client.key.gcpSecretManagerSecretUri'
-  - field: 'nodeConfig.containerd_config.registry_hosts.hosts..dial_timeout'
-    api_field: 'nodeConfig.containerdConfig.registryHosts.hosts.dialTimeout'
+    api_field: 'config.containerdConfig.registryHosts.server'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.host'
+    api_field: 'config.containerdConfig.registryHosts.hosts.host'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.capabilities'
+    api_field: 'config.containerdConfig.registryHosts.hosts.capabilities'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.override_path'
+    api_field: 'config.containerdConfig.registryHosts.hosts.overridePath'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.header.key'
+    api_field: 'config.containerdConfig.registryHosts.hosts.header.key'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.header.value'
+    api_field: 'config.containerdConfig.registryHosts.hosts.header.value'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.ca.gcp_secret_manager_secret_uri'
+    api_field: 'config.containerdConfig.registryHosts.hosts.ca.gcpSecretManagerSecretUri'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.client.cert.gcp_secret_manager_secret_uri'
+    api_field: 'config.containerdConfig.registryHosts.hosts.client.cert.gcpSecretManagerSecretUri'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.client.key.gcp_secret_manager_secret_uri'
+    api_field: 'config.containerdConfig.registryHosts.hosts.client.key.gcpSecretManagerSecretUri'
+  - field: 'node_config.containerd_config.registry_hosts.hosts.dial_timeout'
+    api_field: 'config.containerdConfig.registryHosts.hosts.dialTimeout'
   - field: 'node_config.disk_size_gb'
     api_field: 'config.diskSizeGb'
   - field: 'node_config.disk_type'