bug: fix CSM's access log query was wrong for GDC clusters

pkg/task/inspection/googlecloudclustergdcbaremetal/impl/gdcvforbaremetalclusternameprefix_task.go

@@ -20,10 +20,17 @@ import (
 	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 	googlecloudclustergdcbaremetal_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustergdcbaremetal/contract"
+	googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
 )
 
-// GDCVForBaremetalClusterNamePrefixTask is a task that returns an empty string as the cluster name prefix for GDCV for Baremetal.
-// This task is necessary to satisfy the dependency of the log source profile, but GDCV for Baremetal does not require a prefix.
-var GDCVForBaremetalClusterNamePrefixTask = coretask.NewTask(googlecloudclustergdcbaremetal_contract.ClusterNamePrefixTaskIDForGDCVForBaremetal, []taskid.UntypedTaskReference{}, func(_ context.Context) (string, error) {
-	return "", nil
+// GDCVForBaremetalClusterNamePrefixTask is a task that returns a prefix policy as the cluster name prefix for GDCV for Baremetal.
+// This task applies "baremetalClusters/" prefix only for platform audit and CSM logs.
+var GDCVForBaremetalClusterNamePrefixTask = coretask.NewTask(googlecloudclustergdcbaremetal_contract.ClusterNamePrefixTaskIDForGDCVForBaremetal, []taskid.UntypedTaskReference{}, func(_ context.Context) (googlecloudk8scommon_contract.ClusterPrefixPolicy, error) {
+	return googlecloudk8scommon_contract.ClusterPrefixPolicy{
+		Prefix: "baremetalClusters/",
+		RequiredUsages: []googlecloudk8scommon_contract.ClusterNameUsage{
+			googlecloudk8scommon_contract.ClusterNameUsageK8sPlatformAudit,
+			googlecloudk8scommon_contract.ClusterNameUsageCSM,
+		},
+	}, nil
 })

pkg/task/inspection/googlecloudclustergdcvmware/impl/gdcvforvmwareclusternameprefix_task.go

@@ -20,10 +20,17 @@ import (
 	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 	googlecloudclustergdcvmware_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustergdcvmware/contract"
+	googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
 )
 
-// GDCVForVMWareClusterNamePrefixTask is a task that returns an empty string as the cluster name prefix for GDCV for VMWare.
-// This task is necessary to satisfy the dependency of the log source profile, but GDCV for VMWare does not require a prefix.
-var GDCVForVMWareClusterNamePrefixTask = coretask.NewTask(googlecloudclustergdcvmware_contract.ClusterNamePrefixTaskIDForGDCVForVMWare, []taskid.UntypedTaskReference{}, func(_ context.Context) (string, error) {
-	return "", nil
+// GDCVForVMWareClusterNamePrefixTask is a task that returns a prefix policy as the cluster name prefix for GDCV for VMWare.
+// This task applies "vmwareClusters/" prefix only for platform audit and CSM logs.
+var GDCVForVMWareClusterNamePrefixTask = coretask.NewTask(googlecloudclustergdcvmware_contract.ClusterNamePrefixTaskIDForGDCVForVMWare, []taskid.UntypedTaskReference{}, func(_ context.Context) (googlecloudk8scommon_contract.ClusterPrefixPolicy, error) {
+	return googlecloudk8scommon_contract.ClusterPrefixPolicy{
+		Prefix: "vmwareClusters/",
+		RequiredUsages: []googlecloudk8scommon_contract.ClusterNameUsage{
+			googlecloudk8scommon_contract.ClusterNameUsageK8sPlatformAudit,
+			googlecloudk8scommon_contract.ClusterNameUsageCSM,
+		},
+	}, nil
 })

pkg/task/inspection/googlecloudclustergke/impl/gkeclusternameprefix_task.go

@@ -20,10 +20,14 @@ import (
 	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 	googlecloudclustergke_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustergke/contract"
+	googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
 )
 
-// GKEClusterNamePrefixTask is a task that returns an empty string as the cluster name prefix for GKE.
+// GKEClusterNamePrefixTask is a task that returns an empty prefix policy as the cluster name prefix for GKE.
 // This task is necessary to satisfy the dependency of the log source profile, but GKE does not require a prefix.
-var GKEClusterNamePrefixTask = coretask.NewTask(googlecloudclustergke_contract.ClusterNamePrefixTaskIDForGKE, []taskid.UntypedTaskReference{}, func(ctx context.Context) (string, error) {
-	return "", nil
+var GKEClusterNamePrefixTask = coretask.NewTask(googlecloudclustergke_contract.ClusterNamePrefixTaskIDForGKE, []taskid.UntypedTaskReference{}, func(ctx context.Context) (googlecloudk8scommon_contract.ClusterPrefixPolicy, error) {
+	return googlecloudk8scommon_contract.ClusterPrefixPolicy{
+		Prefix:         "",
+		RequiredUsages: nil,
+	}, nil
 })

pkg/task/inspection/googlecloudclustergkeonaws/impl/prefix.go

@@ -20,9 +20,18 @@ import (
 	common_task "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 	googlecloudclustergkeonaws_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustergkeonaws/contract"
+	googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
 )
 
-// AnthosOnAWSClusterNamePrefixTask is a task that provides the cluster name prefix for GKE on AWS.
-var AnthosOnAWSClusterNamePrefixTask = common_task.NewTask(googlecloudclustergkeonaws_contract.ClusterNamePrefixTaskID, []taskid.UntypedTaskReference{}, func(_ context.Context) (string, error) {
-	return "awsClusters/", nil
+// AnthosOnAWSClusterNamePrefixTask is a task that provides the cluster name prefix policy for GKE on AWS.
+// This task applies "awsClusters/" prefix across all usage layers.
+var AnthosOnAWSClusterNamePrefixTask = common_task.NewTask(googlecloudclustergkeonaws_contract.ClusterNamePrefixTaskID, []taskid.UntypedTaskReference{}, func(_ context.Context) (googlecloudk8scommon_contract.ClusterPrefixPolicy, error) {
+	return googlecloudk8scommon_contract.ClusterPrefixPolicy{
+		Prefix: "awsClusters/",
+		RequiredUsages: []googlecloudk8scommon_contract.ClusterNameUsage{
+			googlecloudk8scommon_contract.ClusterNameUsageK8sCluster,
+			googlecloudk8scommon_contract.ClusterNameUsageK8sPlatformAudit,
+			googlecloudk8scommon_contract.ClusterNameUsageCSM,
+		},
+	}, nil
 })

pkg/task/inspection/googlecloudclustergkeonazure/impl/prefix.go

@@ -20,9 +20,18 @@ import (
 	coretask "github.com/GoogleCloudPlatform/khi/pkg/core/task"
 	"github.com/GoogleCloudPlatform/khi/pkg/core/task/taskid"
 	googlecloudclustergkeonazure_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudclustergkeonazure/contract"
+	googlecloudk8scommon_contract "github.com/GoogleCloudPlatform/khi/pkg/task/inspection/googlecloudk8scommon/contract"
 )
 
-// AnthosOnAzureClusterNamePrefixTask is a task that provides the cluster name prefix for GKE on Azure.
-var AnthosOnAzureClusterNamePrefixTask = coretask.NewTask(googlecloudclustergkeonazure_contract.ClusterNamePrefixTaskID, []taskid.UntypedTaskReference{}, func(_ context.Context) (string, error) {
-	return "azureClusters/", nil
+// AnthosOnAzureClusterNamePrefixTask is a task that provides the cluster name prefix policy for GKE on Azure.
+// This task applies "azureClusters/" prefix across all usage layers.
+var AnthosOnAzureClusterNamePrefixTask = coretask.NewTask(googlecloudclustergkeonazure_contract.ClusterNamePrefixTaskID, []taskid.UntypedTaskReference{}, func(_ context.Context) (googlecloudk8scommon_contract.ClusterPrefixPolicy, error) {
+	return googlecloudk8scommon_contract.ClusterPrefixPolicy{
+		Prefix: "azureClusters/",
+		RequiredUsages: []googlecloudk8scommon_contract.ClusterNameUsage{
+			googlecloudk8scommon_contract.ClusterNameUsageK8sCluster,
+			googlecloudk8scommon_contract.ClusterNameUsageK8sPlatformAudit,
+			googlecloudk8scommon_contract.ClusterNameUsageCSM,
+		},
+	}, nil
 })

pkg/task/inspection/googlecloudk8scommon/contract/cluster.go

@@ -16,23 +16,64 @@ package googlecloudk8scommon_contract
 
 import "fmt"
 
+// ClusterNameUsage represents a usage context layer of a cluster name.
+type ClusterNameUsage string
+
+const (
+	// ClusterNameUsageK8sCluster is the usage context for Kubernetes internal common logs.
+	ClusterNameUsageK8sCluster ClusterNameUsage = "k8s_cluster"
+	// ClusterNameUsageK8sPlatformAudit is the usage context for Kubernetes platform management audit logs.
+	ClusterNameUsageK8sPlatformAudit ClusterNameUsage = "k8s_platform_audit"
+	// ClusterNameUsageCSM is the usage context for Cloud Service Mesh logs.
+	ClusterNameUsageCSM ClusterNameUsage = "csm"
+)
+
+// ClusterPrefixPolicy defines a policy to apply a prefix to a cluster name based on the usage context.
+type ClusterPrefixPolicy struct {
+	// Prefix is the single cluster prefix string.
+	Prefix string
+	// RequiredUsages represents the list of usages that require the prefix.
+	RequiredUsages []ClusterNameUsage
+}
+
+// PrefixFor returns the prefix for the given usage context if the usage is listed in RequiredUsages.
+func (c ClusterPrefixPolicy) PrefixFor(usage ClusterNameUsage) string {
+	for _, u := range c.RequiredUsages {
+		if u == usage {
+			return c.Prefix
+		}
+	}
+	return ""
+}
+
+// Apply applies the prefix to the given cluster name if the usage is listed in RequiredUsages.
+func (c ClusterPrefixPolicy) Apply(usage ClusterNameUsage, clusterName string) string {
+	return fmt.Sprintf("%s%s", c.PrefixFor(usage), clusterName)
+}
+
 // GoogleCloudClusterIdentity is the tuple identify a cluster in Google Cloud.
 type GoogleCloudClusterIdentity struct {
 	// ProjectID is the project ID of the cluster.
 	ProjectID string
-	// ClusterTypePrefix is an empty string for GKE & GDC, "awsClusters/" for GKE on AWS and "azureClusters/" for GKE on Azure.
-	ClusterTypePrefix string
+	// PrefixPolicy is the prefix policy applied to the cluster name.
+	PrefixPolicy ClusterPrefixPolicy
 	// ClusterName is the name of the cluster.
 	ClusterName string
 	// Location is the location of the cluster.
 	Location string
 }
 
-func (g *GoogleCloudClusterIdentity) NameWithClusterTypePrefix() string {
-	return fmt.Sprintf("%s%s", g.ClusterTypePrefix, g.ClusterName)
+// NameFor returns the cluster name representation for the given usage context.
+func (g *GoogleCloudClusterIdentity) NameFor(usage ClusterNameUsage) string {
+	return g.PrefixPolicy.Apply(usage, g.ClusterName)
+}
+
+// PrefixFor returns the cluster prefix for the given usage context.
+func (g *GoogleCloudClusterIdentity) PrefixFor(usage ClusterNameUsage) string {
+	return g.PrefixPolicy.PrefixFor(usage)
 }
 
 // UniqueDigest returns an unique string for the cluster identity. This can be used as the cache key depending on a cluster.
 func (g *GoogleCloudClusterIdentity) UniqueDigest() string {
-	return fmt.Sprintf("%s|%s|%s|%s", g.ProjectID, g.ClusterTypePrefix, g.ClusterName, g.Location)
+	return fmt.Sprintf("%s|%s|%s|%s", g.ProjectID, g.PrefixPolicy.Prefix, g.ClusterName, g.Location)
 }

pkg/task/inspection/googlecloudk8scommon/contract/taskid.go

@@ -49,10 +49,10 @@ var AutocompletePodNamesTaskID = taskid.NewDefaultImplementationID[*inspectionco
 var HeaderSuggestedFileNameTaskID = taskid.NewDefaultImplementationID[struct{}](GoogleCloudCommonK8STaskIDPrefix + "header-suggested-file-name")
 
 // ClusterNamePrefixTaskRef is the task reference ID for generating the cluster name prefix used in query.
-// For GKE, it's just a task to return "" always.
-// For Anthos on AWS, it should return "awsClusters/" because the `resource.labels.cluster_name` field would be `awsClusters/<cluster-name>`
-// For Anthos on Azure, it will be "azureClusters/"
-var ClusterNamePrefixTaskRef = taskid.NewTaskReference[string](GoogleCloudCommonK8STaskIDPrefix + "cluster-name-prefix")
+// For GKE, it's just a task to return empty prefix policy always.
+// For Anthos on AWS, it should return "awsClusters/".
+// For Anthos on Azure, it will be "azureClusters/".
+var ClusterNamePrefixTaskRef = taskid.NewTaskReference[ClusterPrefixPolicy](GoogleCloudCommonK8STaskIDPrefix + "cluster-name-prefix")
 
 // InputClusterNameTaskID is the task ID for the cluster name.
 var InputClusterNameTaskID = taskid.NewDefaultImplementationID[string](GoogleCloudCommonK8STaskIDPrefix + "input-cluster-name")

pkg/task/inspection/googlecloudk8scommon/impl/autocomplete.go

@@ -58,7 +58,7 @@ var AutocompleteClusterIdentityTask = inspectiontaskbase.NewCachedTask(googleclo
 	cf := coretask.GetTaskResult(ctx, googlecloudcommon_contract.APIClientFactoryTaskID.Ref())
 	optionInjector := coretask.GetTaskResult(ctx, googlecloudcommon_contract.APIClientCallOptionsInjectorTaskID.Ref())
 
-	currentDigest := fmt.Sprintf("%s-%s-%d-%d", clusterNamePrefix, projectID, startTime.Unix(), endTime.Unix())
+	currentDigest := fmt.Sprintf("%s-%s-%d-%d", clusterNamePrefix.PrefixFor(googlecloudk8scommon_contract.ClusterNameUsageK8sCluster), projectID, startTime.Unix(), endTime.Unix())
 	if currentDigest == prevValue.DependencyDigest {
 		return prevValue, nil
 	}
@@ -91,18 +91,18 @@ var AutocompleteClusterIdentityTask = inspectiontaskbase.NewCachedTask(googleclo
 	if err != nil {
 		errorString = err.Error()
 	}
-	metricsLabels = filterAndTrimPrefixFromClusterNames(metricsLabels, clusterNamePrefix)
+	metricsLabels = filterAndTrimPrefixFromClusterNames(metricsLabels, clusterNamePrefix.PrefixFor(googlecloudk8scommon_contract.ClusterNameUsageK8sCluster))
 	if hintString == "" && errorString == "" && len(metricsLabels) == 0 {
 		hintString = fmt.Sprintf("No cluster names found between %s and %s. It is highly likely that the time range is incorrect. Please verify the time range, or proceed by manually entering the cluster name.", startTime.Format(time.RFC3339), endTime.Format(time.RFC3339))
 	}
 
 	identities := make([]googlecloudk8scommon_contract.GoogleCloudClusterIdentity, len(metricsLabels))
 	for i, labels := range metricsLabels {
 		identities[i] = googlecloudk8scommon_contract.GoogleCloudClusterIdentity{
-			ProjectID:         projectID,
-			ClusterTypePrefix: clusterNamePrefix,
-			ClusterName:       labels["cluster_name"],
-			Location:          labels["location"],
+			ProjectID:    projectID,
+			PrefixPolicy: clusterNamePrefix,
+			ClusterName:  labels["cluster_name"],
+			Location:     labels["location"],
 		}
 	}
 

pkg/task/inspection/googlecloudk8scommon/impl/clusteridentity.go

@@ -34,11 +34,12 @@ var ClusterIdentityTask = inspectiontaskbase.NewInspectionTask(googlecloudk8scom
 	projectID := coretask.GetTaskResult(ctx, googlecloudcommon_contract.InputProjectIdTaskID.Ref())
 	clusterName := coretask.GetTaskResult(ctx, googlecloudk8scommon_contract.InputClusterNameTaskID.Ref())
 	location := coretask.GetTaskResult(ctx, googlecloudcommon_contract.InputLocationsTaskID.Ref())
-	clusterTypePrefix := coretask.GetTaskResult(ctx, googlecloudk8scommon_contract.ClusterNamePrefixTaskRef)
+	prefixPolicy := coretask.GetTaskResult(ctx, googlecloudk8scommon_contract.ClusterNamePrefixTaskRef)
 	return googlecloudk8scommon_contract.GoogleCloudClusterIdentity{
-		ProjectID:         projectID,
-		ClusterTypePrefix: clusterTypePrefix,
-		ClusterName:       clusterName,
-		Location:          location,
+		ProjectID:    projectID,
+		PrefixPolicy: prefixPolicy,
+		ClusterName:  clusterName,
+		Location:     location,
 	}, nil
+
 })

pkg/task/inspection/googlecloudk8scommon/impl/inputclustername_task.go

@@ -63,10 +63,11 @@ var InputClusterNameTask = formtask.NewTextFormTaskBuilder(googlecloudk8scommon_
 			return clusters.Hint, inspectionmetadata.Info, nil
 		}
 		for _, suggestedCluster := range clusters.Values {
-			if suggestedCluster.NameWithClusterTypePrefix() == convertedValue.(string) {
+			if suggestedCluster.ClusterName == convertedValue.(string) {
 				return "", inspectionmetadata.Info, nil
 			}
 		}
+
 		availableClusterNameStr := ""
 		for _, cluster := range dedupeClusterName(clusters.Values) {
 			availableClusterNameStr += fmt.Sprintf("* %s\n", cluster)