discovery: Remove EKS audit logs from EKS fetcher

Remove checking EKS clusters for audit log fetching from the EKS fetcher
and move it up into the AWS resource sync. This has slightly more
overhead in that we need to reconstruct a map to check if it should have
audit logs fetched, but the code is much cleaner as a result.

Get rid of the clwClient interface as it embedded only a single other
interface - its a bit redundant even if it does follow the pattern of
the other AWS service clients.

Author: camscale

lib/srv/discovery/access_graph_aws.go

@@ -130,14 +130,27 @@ func (s *Server) reconcileAccessGraph(
 	// Each fetcher can return an error and a result.
 	for range allFetchers {
 		fetcherResult := <-resultsC
-		if fetcherResult.err != nil {
-			errs = append(errs, fetcherResult.err)
+		fetcher, result, err := fetcherResult.fetcher, fetcherResult.result, fetcherResult.err
+		if err != nil {
+			errs = append(errs, err)
+		}
+		if result == nil {
+			continue
 		}
-		if fetcherResult.result != nil {
-			results = append(results, fetcherResult.result)
-			for _, cluster := range fetcherResult.result.EKSAuditLogClusters {
-				fetcherCluster := eksAuditLogCluster{fetcherResult.fetcher, cluster}
-				auditLogClusters = append(auditLogClusters, fetcherCluster)
+		results = append(results, result)
+		// If the fetcher is configured for EKS audit logs, see if any
+		// EKS clusters match the configured tags.
+		if fetcher.EKSAuditLogs == nil {
+			continue
+		}
+		for _, cluster := range result.EKSClusters {
+			clusterTags := make(map[string]string, len(cluster.Tags))
+			for _, tag := range cluster.Tags {
+				clusterTags[tag.GetKey()] = tag.GetValue().GetValue()
+			}
+			match, _, _ := services.MatchLabels(fetcher.EKSAuditLogs.Tags, clusterTags)
+			if match {
+				auditLogClusters = append(auditLogClusters, eksAuditLogCluster{fetcher, cluster})
 			}
 		}
 	}

lib/srv/discovery/fetchers/aws-sync/aws-sync.go

@@ -133,8 +133,8 @@ type awsClientProvider interface {
 	getSTSClient(cfg aws.Config, optFns ...func(*sts.Options)) stsClient
 	// getKMSClient provides a [kmsClient].
 	getKMSClient(cfg aws.Config, optFns ...func(*kms.Options)) kmsClient
-	// getCloudWatchLogsClient provides a [cwlClient].
-	getCloudWatchLogsClient(cfg aws.Config, optFns ...func(*cloudwatchlogs.Options)) cwlClient
+	// getCloudWatchLogsClient provides a [cloudwatchlogs.FilterLogEventsAPIClient].
+	getCloudWatchLogsClient(cfg aws.Config, optFns ...func(*cloudwatchlogs.Options)) cloudwatchlogs.FilterLogEventsAPIClient
 }
 
 type defaultAWSClients struct{}
@@ -159,7 +159,7 @@ func (defaultAWSClients) getKMSClient(cfg aws.Config, optFns ...func(*kms.Option
 	return kms.NewFromConfig(cfg, optFns...)
 }
 
-func (defaultAWSClients) getCloudWatchLogsClient(cfg aws.Config, optFns ...func(*cloudwatchlogs.Options)) cwlClient {
+func (defaultAWSClients) getCloudWatchLogsClient(cfg aws.Config, optFns ...func(*cloudwatchlogs.Options)) cloudwatchlogs.FilterLogEventsAPIClient {
 	return cloudwatchlogs.NewFromConfig(cfg, optFns...)
 }
 
@@ -234,11 +234,6 @@ type Resources struct {
 	OIDCProviders []*accessgraphv1alpha.AWSOIDCProviderV1
 	// KMSKeys is a list of KMS keys.
 	KMSKeys []*accessgraphv1alpha.AWSKMSKeyV1
-
-	// EKSAuditLogClusters is a subset of the clusters in the field EKSClusters.
-	// These are the clusters for which apiserver audit logs should be fetched.
-	// These are not sent to access graph with the other resources.
-	EKSAuditLogClusters []*accessgraphv1alpha.AWSEKSClusterV1
 }
 
 func (r *Resources) count() int {

lib/srv/discovery/fetchers/aws-sync/cloudwatchlogs.go

@@ -30,12 +30,6 @@ import (
 	accessgraphv1alpha "github.com/gravitational/teleport/gen/proto/go/accessgraph/v1alpha"
 )
 
-// cwlClient has the AWS interfaces required by this package to fetch
-// CloudWatch logs.
-type cwlClient interface {
-	cloudwatchlogs.FilterLogEventsAPIClient
-}
-
 // FetchEKSAuditLogs returns a slice of audit log events for the given cluster
 // starting from the given cursor.
 func (a *Fetcher) FetchEKSAuditLogs(ctx context.Context, cluster *accessgraphv1alpha.AWSEKSClusterV1, cursor *accessgraphv1alpha.KubeAuditLogCursor) ([]cwltypes.FilteredLogEvent, error) {

lib/srv/discovery/fetchers/aws-sync/eks.go

@@ -20,7 +20,6 @@ package aws_sync
 
 import (
 	"context"
-	"slices"
 	"sync"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -32,7 +31,6 @@ import (
 
 	accessgraphv1alpha "github.com/gravitational/teleport/gen/proto/go/accessgraph/v1alpha"
 	"github.com/gravitational/teleport/lib/cloud/awsconfig"
-	"github.com/gravitational/teleport/lib/services"
 )
 
 // EKSClientGetter returns an EKS client for aws-sync.
@@ -60,7 +58,6 @@ func (a *Fetcher) pollAWSEKSClusters(ctx context.Context, result *Resources, col
 		result.EKSClusters = output.clusters
 		result.AssociatedAccessPolicies = output.associatedPolicies
 		result.AccessEntries = output.accessEntry
-		result.EKSAuditLogClusters = output.auditLogClusters
 		return nil
 	}
 }
@@ -70,7 +67,6 @@ type fetchAWSEKSClustersOutput struct {
 	clusters           []*accessgraphv1alpha.AWSEKSClusterV1
 	associatedPolicies []*accessgraphv1alpha.AWSEKSAssociatedAccessPolicyV1
 	accessEntry        []*accessgraphv1alpha.AWSEKSClusterAccessEntryV1
-	auditLogClusters   []*accessgraphv1alpha.AWSEKSClusterV1
 }
 
 // fetchAWSSEKSClusters fetches eks instances from all regions.
@@ -89,7 +85,6 @@ func (a *Fetcher) fetchAWSSEKSClusters(ctx context.Context) (fetchAWSEKSClusters
 	collectClusters := func(cluster *accessgraphv1alpha.AWSEKSClusterV1,
 		clusterAssociatedPolicies []*accessgraphv1alpha.AWSEKSAssociatedAccessPolicyV1,
 		clusterAccessEntries []*accessgraphv1alpha.AWSEKSClusterAccessEntryV1,
-		fetchAuditLogs bool,
 		err error,
 	) {
 		hostsMu.Lock()
@@ -99,9 +94,6 @@ func (a *Fetcher) fetchAWSSEKSClusters(ctx context.Context) (fetchAWSEKSClusters
 		}
 		if cluster != nil {
 			output.clusters = append(output.clusters, cluster)
-			if fetchAuditLogs {
-				output.auditLogClusters = append(output.auditLogClusters, cluster)
-			}
 		}
 		output.associatedPolicies = append(output.associatedPolicies, clusterAssociatedPolicies...)
 		output.accessEntry = append(output.accessEntry, clusterAccessEntries...)
@@ -111,8 +103,7 @@ func (a *Fetcher) fetchAWSSEKSClusters(ctx context.Context) (fetchAWSEKSClusters
 		eG.Go(func() error {
 			eksClient, err := a.GetEKSClient(ctx, region, a.getAWSOptions()...)
 			if err != nil {
-				fetchAuditLogs := false
-				collectClusters(nil, nil, nil, fetchAuditLogs, trace.Wrap(err))
+				collectClusters(nil, nil, nil, trace.Wrap(err))
 				return nil
 			}
 
@@ -130,14 +121,10 @@ func (a *Fetcher) fetchAWSSEKSClusters(ctx context.Context) (fetchAWSEKSClusters
 					oldAssociatedPolicies := sliceFilter(existing.AssociatedAccessPolicies, func(ap *accessgraphv1alpha.AWSEKSAssociatedAccessPolicyV1) bool {
 						return ap.Cluster.Region == region && ap.AccountId == a.AccountID
 					})
-					oldAuditLogClusters := sliceFilter(existing.EKSAuditLogClusters, func(cluster *accessgraphv1alpha.AWSEKSClusterV1) bool {
-						return cluster.Region == region && cluster.AccountId == a.AccountID
-					})
 					hostsMu.Lock()
 					output.clusters = append(output.clusters, oldEKSClusters...)
 					output.associatedPolicies = append(output.associatedPolicies, oldAssociatedPolicies...)
 					output.accessEntry = append(output.accessEntry, oldAccessEntries...)
-					output.auditLogClusters = append(output.auditLogClusters, oldAuditLogClusters...)
 					hostsMu.Unlock()
 					break
 				}
@@ -160,27 +147,22 @@ func (a *Fetcher) fetchAWSSEKSClusters(ctx context.Context) (fetchAWSEKSClusters
 				},
 				)
 				if err != nil {
-					fetchAuditLogs := slices.Contains(existing.EKSAuditLogClusters, oldCluster)
-					collectClusters(oldCluster, oldAssociatedPolicies, oldAccessEntries, fetchAuditLogs, trace.Wrap(err))
+					collectClusters(oldCluster, oldAssociatedPolicies, oldAccessEntries, trace.Wrap(err))
 					return nil
 				}
 				protoCluster := awsEKSClusterToProtoCluster(cluster.Cluster, region, a.AccountID)
-				fetchAuditLogs := false
-				if a.EKSAuditLogs != nil {
-					fetchAuditLogs, _, _ = services.MatchLabels(a.EKSAuditLogs.Tags, cluster.Cluster.Tags)
-				}
 
 				// if eks cluster only allows CONFIGMAP auth, skip polling of access entries and
 				// associated policies.
 				if cluster.Cluster != nil && cluster.Cluster.AccessConfig != nil &&
 					cluster.Cluster.AccessConfig.AuthenticationMode == ekstypes.AuthenticationModeConfigMap {
-					collectClusters(protoCluster, nil, nil, fetchAuditLogs, nil)
+					collectClusters(protoCluster, nil, nil, nil)
 					continue
 				}
 				// fetchAccessEntries retries the list of configured access entries
 				accessEntries, err := a.fetchAccessEntries(ctx, eksClient, protoCluster)
 				if err != nil {
-					collectClusters(protoCluster, oldAssociatedPolicies, oldAccessEntries, fetchAuditLogs, trace.Wrap(err))
+					collectClusters(protoCluster, oldAssociatedPolicies, oldAccessEntries, trace.Wrap(err))
 				}
 
 				accessEntryARNs := make([]string, 0, len(accessEntries))
@@ -193,9 +175,9 @@ func (a *Fetcher) fetchAWSSEKSClusters(ctx context.Context) (fetchAWSEKSClusters
 
 				associatedPolicies, err := a.fetchAssociatedPolicies(ctx, eksClient, protoCluster, accessEntryARNs)
 				if err != nil {
-					collectClusters(protoCluster, oldAssociatedPolicies, accessEntries, fetchAuditLogs, trace.Wrap(err))
+					collectClusters(protoCluster, oldAssociatedPolicies, accessEntries, trace.Wrap(err))
 				}
-				collectClusters(protoCluster, associatedPolicies, accessEntries, fetchAuditLogs, nil)
+				collectClusters(protoCluster, associatedPolicies, accessEntries, nil)
 			}
 			return nil
 		})

lib/srv/discovery/fetchers/aws-sync/eks_test.go

@@ -33,7 +33,6 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
-	"github.com/gravitational/teleport/api/utils"
 	accessgraphv1alpha "github.com/gravitational/teleport/gen/proto/go/accessgraph/v1alpha"
 	"github.com/gravitational/teleport/lib/cloud/awsconfig"
 )
@@ -48,7 +47,7 @@ const (
 type mockedEKSClient struct {
 	clusters                 []*ekstypes.Cluster
 	accessEntries            []*ekstypes.AccessEntry
-	associatedAccessPolicies map[string][]ekstypes.AssociatedAccessPolicy
+	associatedAccessPolicies []ekstypes.AssociatedAccessPolicy
 }
 
 func (m *mockedEKSClient) DescribeCluster(ctx context.Context, input *eks.DescribeClusterInput, optFns ...func(*eks.Options)) (*eks.DescribeClusterOutput, error) {
@@ -75,9 +74,7 @@ func (m *mockedEKSClient) ListClusters(ctx context.Context, input *eks.ListClust
 func (m *mockedEKSClient) ListAccessEntries(ctx context.Context, input *eks.ListAccessEntriesInput, optFns ...func(*eks.Options)) (*eks.ListAccessEntriesOutput, error) {
 	accessEntries := make([]string, 0, len(m.accessEntries))
 	for _, accessEntry := range m.accessEntries {
-		if *input.ClusterName == *accessEntry.ClusterName {
-			accessEntries = append(accessEntries, aws.ToString(accessEntry.AccessEntryArn))
-		}
+		accessEntries = append(accessEntries, aws.ToString(accessEntry.AccessEntryArn))
 	}
 	return &eks.ListAccessEntriesOutput{
 		AccessEntries: accessEntries,
@@ -86,7 +83,7 @@ func (m *mockedEKSClient) ListAccessEntries(ctx context.Context, input *eks.List
 
 func (m *mockedEKSClient) ListAssociatedAccessPolicies(ctx context.Context, input *eks.ListAssociatedAccessPoliciesInput, optFns ...func(*eks.Options)) (*eks.ListAssociatedAccessPoliciesOutput, error) {
 	return &eks.ListAssociatedAccessPoliciesOutput{
-		AssociatedAccessPolicies: m.associatedAccessPolicies[*input.ClusterName],
+		AssociatedAccessPolicies: m.associatedAccessPolicies,
 	}, nil
 }
 
@@ -113,7 +110,7 @@ func TestPollAWSEKSClusters(t *testing.T) {
 		accountID = "12345678"
 	)
 	regions := []string{"eu-west-1"}
-	cluster1 := &accessgraphv1alpha.AWSEKSClusterV1{
+	cluster := &accessgraphv1alpha.AWSEKSClusterV1{
 		Name:      "cluster1",
 		Arn:       "arn:us-west1:eks:cluster1",
 		CreatedAt: timestamppb.New(date),
@@ -131,24 +128,6 @@ func TestPollAWSEKSClusters(t *testing.T) {
 		Region:    "eu-west-1",
 		AccountId: "12345678",
 	}
-	cluster2 := &accessgraphv1alpha.AWSEKSClusterV1{
-		Name:      "cluster2",
-		Arn:       "arn:us-west1:eks:cluster2",
-		CreatedAt: timestamppb.New(date),
-		Status:    "ACTIVE",
-		Tags: []*accessgraphv1alpha.AWSTag{
-			{
-				Key:   "tag1",
-				Value: wrapperspb.String("hello"),
-			},
-			{
-				Key:   "tag2",
-				Value: wrapperspb.String("val2"),
-			},
-		},
-		Region:    "eu-west-1",
-		AccountId: "12345678",
-	}
 	tests := []struct {
 		name string
 		want *Resources
@@ -157,14 +136,11 @@ func TestPollAWSEKSClusters(t *testing.T) {
 			name: "poll eks clusters",
 			want: &Resources{
 				EKSClusters: []*accessgraphv1alpha.AWSEKSClusterV1{
-					cluster1, cluster2,
-				},
-				EKSAuditLogClusters: []*accessgraphv1alpha.AWSEKSClusterV1{
-					cluster2,
+					cluster,
 				},
 				AccessEntries: []*accessgraphv1alpha.AWSEKSClusterAccessEntryV1{
 					{
-						Cluster:          cluster1,
+						Cluster:          cluster,
 						AccessEntryArn:   "arn:iam:access_entry",
 						CreatedAt:        timestamppb.New(date),
 						KubernetesGroups: []string{"teleport"},
@@ -183,7 +159,7 @@ func TestPollAWSEKSClusters(t *testing.T) {
 				},
 				AssociatedAccessPolicies: []*accessgraphv1alpha.AWSEKSAssociatedAccessPolicyV1{
 					{
-						Cluster:      cluster1,
+						Cluster:      cluster,
 						PrincipalArn: principalARN,
 						Scope: &accessgraphv1alpha.AWSEKSAccessScopeV1{
 							Type:       string(ekstypes.AccessScopeTypeCluster),
@@ -226,12 +202,6 @@ func TestPollAWSEKSClusters(t *testing.T) {
 					Regions:      regions,
 					Integration:  accountID,
 					GetEKSClient: getEKSClient,
-					EKSAuditLogs: &EKSAuditLogs{
-						Tags: map[string]utils.Strings{
-							"tag1": utils.Strings{"hello", "world"},
-							"tag2": utils.Strings{"val*"},
-						},
-					},
 				},
 				lastResult: &Resources{},
 			}
@@ -266,16 +236,6 @@ func eksClusters() []*ekstypes.Cluster {
 				"tag2": "val2",
 			},
 		},
-		{
-			Name:      aws.String("cluster2"),
-			Arn:       aws.String("arn:us-west1:eks:cluster2"),
-			CreatedAt: aws.Time(date),
-			Status:    ekstypes.ClusterStatusActive,
-			Tags: map[string]string{
-				"tag1": "hello",
-				"tag2": "val2",
-			},
-		},
 	}
 }
 
@@ -297,18 +257,16 @@ func accessEntries() []*ekstypes.AccessEntry {
 	}
 }
 
-func associatedPolicies() map[string][]ekstypes.AssociatedAccessPolicy {
-	return map[string][]ekstypes.AssociatedAccessPolicy{
-		"cluster1": {
-			{
-				AccessScope: &ekstypes.AccessScope{
-					Namespaces: []string{"ns1"},
-					Type:       ekstypes.AccessScopeTypeCluster,
-				},
-				ModifiedAt:   aws.Time(date),
-				AssociatedAt: aws.Time(date),
-				PolicyArn:    aws.String("policy_arn"),
+func associatedPolicies() []ekstypes.AssociatedAccessPolicy {
+	return []ekstypes.AssociatedAccessPolicy{
+		{
+			AccessScope: &ekstypes.AccessScope{
+				Namespaces: []string{"ns1"},
+				Type:       ekstypes.AccessScopeTypeCluster,
 			},
+			ModifiedAt:   aws.Time(date),
+			AssociatedAt: aws.Time(date),
+			PolicyArn:    aws.String("policy_arn"),
 		},
 	}
 }

lib/srv/discovery/fetchers/aws-sync/rds_test.go

@@ -225,7 +225,7 @@ type fakeAWSClients struct {
 	s3Client  s3Client
 	stsClient stsClient
 	kmsClient kmsClient
-	cwlClient cwlClient
+	cwlClient cloudwatchlogs.FilterLogEventsAPIClient
 }
 
 func (f fakeAWSClients) getIAMClient(cfg aws.Config, optFns ...func(*iam.Options)) iamClient {
@@ -248,6 +248,6 @@ func (f fakeAWSClients) getKMSClient(cfg aws.Config, optFns ...func(*kms.Options
 	return f.kmsClient
 }
 
-func (f fakeAWSClients) getCloudWatchLogsClient(cfg aws.Config, optFns ...func(*cloudwatchlogs.Options)) cwlClient {
+func (f fakeAWSClients) getCloudWatchLogsClient(cfg aws.Config, optFns ...func(*cloudwatchlogs.Options)) cloudwatchlogs.FilterLogEventsAPIClient {
 	return f.cwlClient
 }