[Multi_K8s-Plugin] Fix livestate drift detection and resource filtering (#6673)

* Implement livestate drift detection ignore and include/exclude resource filtering

Signed-off-by: Mohammed Firdous <124298708+mohammedfirdouss@users.noreply.github.com>

* Fix gofmt formatting in provider/kubernetes.go

Signed-off-by: Mohammed Firdous <124298708+mohammedfirdouss@users.noreply.github.com>

---------

Signed-off-by: Mohammed Firdous <124298708+mohammedfirdouss@users.noreply.github.com>
Co-authored-by: Khanh Tran <32532742+khanhtc1202@users.noreply.github.com>

Author: mohammedfirdouss

pkg/app/pipedv1/plugin/kubernetes_multicluster/config/application.go

@@ -138,6 +138,28 @@ type KubernetesDeployTargetConfig struct {
 	KubeConfigPath string `json:"kubeConfigPath,omitempty"`
 	// Version of kubectl will be used.
 	KubectlVersion string `json:"kubectlVersion"`
+	// Configuration for application resource informer.
+	AppStateInformer KubernetesAppStateInformer `json:"appStateInformer"`
+}
+
+// KubernetesAppStateInformer represents the configuration for application resource informer.
+type KubernetesAppStateInformer struct {
+	// Only watches the specified namespace.
+	// Empty means watching all namespaces.
+	Namespace string `json:"namespace,omitempty"`
+	// List of resources that should be added to the watching targets.
+	IncludeResources []KubernetesResourceMatcher `json:"includeResources,omitempty"`
+	// List of resources that should be ignored from the watching targets.
+	ExcludeResources []KubernetesResourceMatcher `json:"excludeResources,omitempty"`
+}
+
+// KubernetesResourceMatcher represents the matcher for a Kubernetes resource.
+type KubernetesResourceMatcher struct {
+	// The APIVersion of the kubernetes resource.
+	APIVersion string `json:"apiVersion,omitempty"`
+	// The kind name of the kubernetes resource.
+	// Empty means all kinds are matching.
+	Kind string `json:"kind,omitempty"`
 }
 
 // K8sSyncStageOptions contains all configurable values for a K8S_SYNC stage.

pkg/app/pipedv1/plugin/kubernetes_multicluster/livestate/plugin.go

@@ -95,14 +95,17 @@ func (p Plugin) GetLivestate(ctx context.Context, _ *sdk.ConfigNone, deployTarge
 		// Create the kubectl wrapper for the target cluster.
 		kubectl := provider.NewKubectl(kubectlPath)
 
-		// TODO: We need to implement including/excluding resources.
-		// ref; https://pipecd.dev/docs-v0.50.x/user-guide/managing-piped/configuration-reference/#kubernetesappstateinformer
 		namespacedLiveResources, clusterScopedLiveResources, err := provider.GetLiveResources(ctx, kubectl, tc.deployTarget.Config.KubeConfigPath, input.Request.ApplicationID)
 		if err != nil {
 			input.Logger.Error("Failed to get live resources", zap.Error(err))
 			return nil, err
 		}
 
+		// Filter live resources by include/exclude rules from the deploy target's AppStateInformer config.
+		informer := tc.deployTarget.Config.AppStateInformer
+		namespacedLiveResources = filterByAppStateInformer(namespacedLiveResources, informer)
+		clusterScopedLiveResources = filterByAppStateInformer(clusterScopedLiveResources, informer)
+
 		liveState := p.makeAppLivestate(namespacedLiveResources, clusterScopedLiveResources, tc.deployTarget)
 		liveStates = append(liveStates, liveState)
 
@@ -157,9 +160,11 @@ func (p Plugin) makeAppLivestate(namespacedLiveResources, clusterScopedLiveResou
 }
 
 func (p Plugin) makeAppSyncState(liveManifests, gitManifests []provider.Manifest, dt *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig], commit string, logger *zap.Logger) (sdk.ApplicationSyncState, error) {
-	// Calculate SyncState by comparing live manifests with desired manifests
-	// TODO: Implement drift detection ignore configs
-	diffResult, err := provider.DiffList(liveManifests, gitManifests, logger,
+	// Filter out resources annotated to be ignored from drift detection.
+	diffResult, err := provider.DiffList(
+		filterIgnoringManifests(liveManifests),
+		filterIgnoringManifests(gitManifests),
+		logger,
 		diff.WithEquateEmpty(),
 		diff.WithIgnoreAddingMapKeys(),
 		diff.WithCompareNumberAndNumericString(),
@@ -171,6 +176,72 @@ func (p Plugin) makeAppSyncState(liveManifests, gitManifests []provider.Manifest
 	return calculateSyncState(diffResult, commit, dt), nil
 }
 
+// filterIgnoringManifests removes manifests that are annotated to be excluded from drift detection.
+// Resources with the annotation pipecd.dev/ignore-drift-detection=true are skipped.
+func filterIgnoringManifests(manifests []provider.Manifest) []provider.Manifest {
+	out := make([]provider.Manifest, 0, len(manifests))
+	for _, m := range manifests {
+		if m.GetAnnotations()[provider.LabelIgnoreDriftDirection] == provider.IgnoreDriftDetectionTrue {
+			continue
+		}
+		out = append(out, m)
+	}
+	return out
+}
+
+// filterByAppStateInformer filters live resources based on include/exclude rules
+// from the deploy target's AppStateInformer config.
+// If IncludeResources is set, only resources matching at least one entry are kept.
+// If ExcludeResources is set, resources matching any entry are removed.
+// If neither is set, all resources are returned unchanged.
+func filterByAppStateInformer(manifests []provider.Manifest, informer kubeconfig.KubernetesAppStateInformer) []provider.Manifest {
+	if len(informer.IncludeResources) == 0 && len(informer.ExcludeResources) == 0 {
+		return manifests
+	}
+
+	out := make([]provider.Manifest, 0, len(manifests))
+	for _, m := range manifests {
+		if len(informer.IncludeResources) > 0 {
+			included := false
+			for _, matcher := range informer.IncludeResources {
+				if matchesResourceMatcher(m, matcher) {
+					included = true
+					break
+				}
+			}
+			if !included {
+				continue
+			}
+		}
+
+		excluded := false
+		for _, matcher := range informer.ExcludeResources {
+			if matchesResourceMatcher(m, matcher) {
+				excluded = true
+				break
+			}
+		}
+		if excluded {
+			continue
+		}
+
+		out = append(out, m)
+	}
+	return out
+}
+
+// matchesResourceMatcher returns true if the manifest matches the given resource matcher.
+// An empty APIVersion or Kind in the matcher acts as a wildcard for that field.
+func matchesResourceMatcher(m provider.Manifest, matcher kubeconfig.KubernetesResourceMatcher) bool {
+	if matcher.APIVersion != "" && m.APIVersion() != matcher.APIVersion {
+		return false
+	}
+	if matcher.Kind != "" && m.Kind() != matcher.Kind {
+		return false
+	}
+	return true
+}
+
 func calculateSyncState(diffResult *provider.DiffListResult, commit string, dt *sdk.DeployTarget[kubeconfig.KubernetesDeployTargetConfig]) sdk.ApplicationSyncState {
 	if diffResult.NoChanges() {
 		return sdk.ApplicationSyncState{

pkg/app/pipedv1/plugin/kubernetes_multicluster/provider/kubernetes.go

@@ -36,8 +36,11 @@ const (
 	AnnotationOrder      = "pipecd.dev/order"       // The order number of resource used to sort them before using.
 	AnnotationConfigHash = "pipecd.dev/config-hash" // The hash value of all mouting config resources.
 
+	LabelIgnoreDriftDirection = "pipecd.dev/ignore-drift-detection" // Whether the drift detection should ignore this resource.
+
 	// label/annotation values
-	ManagedByPiped     = "piped"
-	UseReplaceEnabled  = "enabled"
-	UseServerSideApply = "true"
+	ManagedByPiped           = "piped"
+	UseReplaceEnabled        = "enabled"
+	UseServerSideApply       = "true"
+	IgnoreDriftDetectionTrue = "true"
 )