Wire kopia exclude list up to BackupOp (#2244)

## Description

Expose the global exclude list mechanism in kopia BackupCollections to other components. Add tests for BackupCollections testing new exclude list functionality

Currently wired to always be passed nil

## Does this PR need a docs update or release note?

- [ ] ✅ Yes, it's included
- [ ] 🕐 Yes, but in a later PR
- [x] ⛔ No 

## Type of change

- [x] 🌻 Feature
- [ ] 🐛 Bugfix
- [ ] 🗺️ Documentation
- [ ] 🤖 Test
- [ ] 💻 CI/Deployment
- [ ] 🧹 Tech Debt/Cleanup

## Issue(s)

* #2243

merge after:
* #2143 

## Test Plan

- [ ] 💪 Manual
- [x] ⚡ Unit test
- [ ] 💚 E2E

src/internal/kopia/wrapper.go

@@ -119,6 +119,7 @@ func (w Wrapper) BackupCollections(
 	ctx context.Context,
 	previousSnapshots []IncrementalBase,
 	collections []data.Collection,
+	globalExcludeSet map[string]struct{},
 	tags map[string]string,
 	buildTreeWithBase bool,
 ) (*BackupStats, *details.Builder, map[string]path.Path, error) {
@@ -129,10 +130,6 @@ func (w Wrapper) BackupCollections(
 	ctx, end := D.Span(ctx, "kopia:backupCollections")
 	defer end()
 
-	// TODO(ashmrtn): Make this a parameter when actually enabling the global
-	// exclude set.
-	var globalExcludeSet map[string]struct{}
-
 	if len(collections) == 0 && len(globalExcludeSet) == 0 {
 		return &BackupStats{}, &details.Builder{}, nil, nil
 	}

src/internal/kopia/wrapper_test.go

@@ -266,6 +266,7 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections() {
 				suite.ctx,
 				prevSnaps,
 				collections,
+				nil,
 				tags,
 				true,
 			)
@@ -353,6 +354,7 @@ func (suite *KopiaIntegrationSuite) TestRestoreAfterCompressionChange() {
 		ctx,
 		nil,
 		[]data.Collection{dc1, dc2},
+		nil,
 		tags,
 		true,
 	)
@@ -435,6 +437,7 @@ func (suite *KopiaIntegrationSuite) TestBackupCollections_ReaderError() {
 		suite.ctx,
 		nil,
 		collections,
+		nil,
 		tags,
 		true,
 	)
@@ -496,6 +499,7 @@ func (suite *KopiaIntegrationSuite) TestBackupCollectionsHandlesNoCollections()
 				nil,
 				test.collections,
 				nil,
+				nil,
 				true,
 			)
 			require.NoError(t, err)
@@ -653,6 +657,7 @@ func (suite *KopiaSimpleRepoIntegrationSuite) SetupTest() {
 		suite.ctx,
 		nil,
 		collections,
+		nil,
 		tags,
 		false,
 	)
@@ -681,6 +686,136 @@ func (c *i64counter) Count(i int64) {
 	c.i += i
 }
 
+func (suite *KopiaSimpleRepoIntegrationSuite) TestBackupExcludeItem() {
+	reason := Reason{
+		ResourceOwner: testUser,
+		Service:       path.ExchangeService,
+		Category:      path.EmailCategory,
+	}
+
+	subtreePathTmp, err := path.Builder{}.Append("tmp").ToDataLayerExchangePathForCategory(
+		testTenant,
+		testUser,
+		path.EmailCategory,
+		false,
+	)
+	require.NoError(suite.T(), err)
+
+	subtreePath := subtreePathTmp.ToBuilder().Dir()
+
+	manifests, err := suite.w.FetchPrevSnapshotManifests(
+		suite.ctx,
+		[]Reason{reason},
+		nil,
+	)
+	require.NoError(suite.T(), err)
+	require.Len(suite.T(), manifests, 1)
+	require.Equal(suite.T(), suite.snapshotID, manifests[0].ID)
+
+	tags := map[string]string{}
+
+	for _, k := range reason.TagKeys() {
+		tags[k] = ""
+	}
+
+	table := []struct {
+		name                  string
+		excludeItem           bool
+		expectedCachedItems   int
+		expectedUncachedItems int
+		cols                  func() []data.Collection
+		backupIDCheck         require.ValueAssertionFunc
+		restoreCheck          assert.ErrorAssertionFunc
+	}{
+		{
+			name:                  "ExcludeItem",
+			excludeItem:           true,
+			expectedCachedItems:   len(suite.filesByPath) - 1,
+			expectedUncachedItems: 0,
+			cols: func() []data.Collection {
+				return nil
+			},
+			backupIDCheck: require.NotEmpty,
+			restoreCheck:  assert.Error,
+		},
+		{
+			name: "NoExcludeItemNoChanges",
+			// No snapshot should be made since there were no changes.
+			expectedCachedItems:   0,
+			expectedUncachedItems: 0,
+			cols: func() []data.Collection {
+				return nil
+			},
+			// Backup doesn't run.
+			backupIDCheck: require.Empty,
+		},
+		{
+			name:                  "NoExcludeItemWithChanges",
+			expectedCachedItems:   len(suite.filesByPath),
+			expectedUncachedItems: 1,
+			cols: func() []data.Collection {
+				c := mockconnector.NewMockExchangeCollection(
+					suite.testPath1,
+					1,
+				)
+				c.ColState = data.NotMovedState
+
+				return []data.Collection{c}
+			},
+			backupIDCheck: require.NotEmpty,
+			restoreCheck:  assert.NoError,
+		},
+	}
+
+	for _, test := range table {
+		suite.T().Run(test.name, func(t *testing.T) {
+			var excluded map[string]struct{}
+			if test.excludeItem {
+				excluded = map[string]struct{}{
+					suite.files[suite.testPath1.String()][0].itemPath.Item(): {},
+				}
+			}
+
+			stats, _, _, err := suite.w.BackupCollections(
+				suite.ctx,
+				[]IncrementalBase{
+					{
+						Manifest: manifests[0].Manifest,
+						SubtreePaths: []*path.Builder{
+							subtreePath,
+						},
+					},
+				},
+				test.cols(),
+				excluded,
+				tags,
+				true,
+			)
+			require.NoError(t, err)
+			assert.Equal(t, test.expectedCachedItems, stats.CachedFileCount)
+			assert.Equal(t, test.expectedUncachedItems, stats.UncachedFileCount)
+
+			test.backupIDCheck(t, stats.SnapshotID)
+
+			if len(stats.SnapshotID) == 0 {
+				return
+			}
+
+			ic := i64counter{}
+
+			_, err = suite.w.RestoreMultipleItems(
+				suite.ctx,
+				string(stats.SnapshotID),
+				[]path.Path{
+					suite.files[suite.testPath1.String()][0].itemPath,
+				},
+				&ic,
+			)
+			test.restoreCheck(t, err)
+		})
+	}
+}
+
 func (suite *KopiaSimpleRepoIntegrationSuite) TestRestoreMultipleItems() {
 	doesntExist, err := path.Builder{}.Append("subdir", "foo").ToDataLayerExchangePathForCategory(
 		testTenant,

src/internal/operations/backup.go

@@ -275,6 +275,7 @@ type backuper interface {
 		ctx context.Context,
 		bases []kopia.IncrementalBase,
 		cs []data.Collection,
+		excluded map[string]struct{},
 		tags map[string]string,
 		buildTreeWithBase bool,
 	) (*kopia.BackupStats, *details.Builder, map[string]path.Path, error)
@@ -404,6 +405,7 @@ func consumeBackupDataCollections(
 		ctx,
 		bases,
 		cs,
+		nil,
 		tags,
 		isIncremental,
 	)

src/internal/operations/backup_test.go

@@ -95,6 +95,7 @@ func (mbu mockBackuper) BackupCollections(
 	ctx context.Context,
 	bases []kopia.IncrementalBase,
 	cs []data.Collection,
+	excluded map[string]struct{},
 	tags map[string]string,
 	buildTreeWithBase bool,
 ) (*kopia.BackupStats, *details.Builder, map[string]path.Path, error) {

src/internal/streamstore/streamstore.go

@@ -78,6 +78,7 @@ func (ss *streamStore) WriteBackupDetails(
 		nil,
 		[]data.Collection{dc},
 		nil,
+		nil,
 		false,
 	)
 	if err != nil {