Repository: kopia/kopia
Pull request #2835 (2023-03-16 21:19:31)

feat(snapshots): Enable compression for StreamingFile entries
Also take fs.Entry for helper that gets compression policy instead of fs.File.

Hooks into existing logic for parsing policy trees to get the policy for the
current path. Users expecting compression for StreamingFiles do need to be
cognizant of the fact that StreamingFiles return a size of 0 when `f.Size()`
is called. This means that setting a minimum compression size may interfere
with enabling compression on StreamingFiles

codecov[bot]: ## [Codecov](https://codecov.io/gh/kopia/kopia/pull/2835?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) Report
Patch coverage: **`100.00`**% and project coverage change: **`-0.05`** :warning:
> Comparison is base [(`90b606f`)](https://codecov.io/gh/kopia/kopia/commit/90b606f5e51cfcd38e8951e983df3dddaee62ea6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) 75.30% compared to head [(`cc81751`)](https://codecov.io/gh/kopia/kopia/pull/2835?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) 75.25%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #2835      +/-   ##
==========================================
- Coverage   75.30%   75.25%   -0.05%     
==========================================
  Files         457      457              
  Lines       35865    35868       +3     
==========================================
- Hits        27008    26993      -15     
- Misses       6965     6980      +15     
- Partials     1892     1895       +3     
```


| [Impacted Files](https://codecov.io/gh/kopia/kopia/pull/2835?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia) | Coverage Δ | |
|---|---|---|
| [snapshot/policy/compression\_policy.go](https://codecov.io/gh/kopia/kopia/pull/2835?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-c25hcHNob3QvcG9saWN5L2NvbXByZXNzaW9uX3BvbGljeS5nbw==) | `60.00% <100.00%> (ø)` | |
| [snapshot/snapshotfs/upload.go](https://codecov.io/gh/kopia/kopia/pull/2835?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia#diff-c25hcHNob3Qvc25hcHNob3Rmcy91cGxvYWQuZ28=) | `84.85% <100.00%> (-0.73%)` | :arrow_down: |

... and [5 files with indirect coverage changes](https://codecov.io/gh/kopia/kopia/pull/2835/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia)

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia)

</details>

[:umbrella: View full report in Codecov by Sentry](https://codecov.io/gh/kopia/kopia/pull/2835?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=kopia).

Shrekster: Thank you for posting this change! I have a few questions:

1. Does this change make compression mandatory for StreamingFile(s) which were otherwise not respecting the compression policy so far on an exiting repository ? That is, would a restore from an older repository fail ? Can we add a compatibility test for this ?
2. Would adding compression to an existing repository bloat its next incremental run when a user will upgrade Kopia to a version that has this change ?
ashmrtn: Great questions! I've put answers to the questions inline with quotes for context

>1. Does this change make compression mandatory for StreamingFile(s) which were otherwise not respecting the compression policy so far on an exiting repository ?
>

If compression is configured for the kopia repository and there's no minimum size set for files to be compressed then yes, it will make compression mandatory for all StreamingFiles. Since StreamingFiles have return 0 when `StreamingFile.Size()` is called an easy way to disable compression for them is to make a policy that has a minimum size for compression > 0 (i.e. minimum file size = 1)

>That is, would a restore from an older repository fail ?
>

No, this reuses existing logic for compressing files. All the added code does is ensure that we allow compression to take place if the policy requests compression. The existing code for compressing data [adds a compression header](https://github.com/kopia/kopia/blob/cd5de60c87c2d497e5f6929a062d0eba96c14758/repo/compression/compressor_zstd.go#L41) to the blob when creating it. The header is then read back when fetching the data from kopia, ensuring kopia can distinguish compressed and uncompressed data as well as different compression types

>Can we add a compatibility test for this ?
>

I can add a test if desired, though I must admit I'm not sure precisely where to add it nor what existing test frameworks I may be able to reuse to help setup said test. If you have any pointers on this I'd really appreciate it.

Though it doesn't give the assurances of a regression test that's checked into the codebase, I can say I've run a manual test where I created a kopia backup without this patch, updated kopia, and restored the (uncompressed) data from kopia without issue

>2. Would adding compression to an existing repository bloat its next incremental run when a user will upgrade Kopia to a version that has this change ?
>

I've tried looking around for docs to answer this solidly and came away with conflicting answers. Because this ties into existing logic for compressing data, it should follow the regular compression/dedupe rules for kopia. Looking at the current [doc page](https://kopia.io/docs/advanced/compression/) for compression, it states

> When Kopia sees a file to backup, it first splits its data into “chunks” with the configured splitter. Then compute its hash with the configured hash algorithm, and compare it against the existing database. If a matching hash is found, this chunk is already stored, thus can be safely discarded. Otherwise, Kopia applies the compression algorithm on the chunk, encrypt it, and pack a few of such into a blob if fit.

The above seems to imply that no, changing compression shouldn't cause mass file reuploads because the (uncompressed) hash of the data will still match

On the other hand there's this [comment](https://github.com/kopia/kopia/issues/728#issuecomment-748270429) from an open question in the kopia repo which seems to imply that compression happens before hashing/dedupe, which would probably cause reuploads of data as kopia would compare uncompressed hashes to compressed hashes

>objects are written in the following stages:
>
>```text
>splitter -> compressor -> hash/dedup -> encrypt -> package
>```

Empirically testing this, the answer appears to be no, it wouldn't cause lots of files to be reuploaded on the next run due to a change in compression. For reference, the empirical test I ran was to backup some StreamingFiles with the unpatched version of kopia. The first run showed all bytes hashed and uploaded. The second run (with patched kopia) showed all bytes hashed (expected as modtime for the StreamingFiles wasn't explicitly set) and only a small amount of data uploaded (i.e. < 1MB for a backup that included 2 trivially compressible files of 512MB)
ashmrtn: re: the discussion on whether this change will cause mass file reuploads. On closer examination, it appears the answer to that may actually be "it depends." I was digging around the codebase a bit more and realized that compression may actually occur in one of two different places during a snapshot: the content manager or the object manager, with [preference for compressing in the content manager](https://github.com/kopia/kopia/blob/cc00388f6d2a2b78f03e2507b4b2f25519d1f7e4/repo/object/object_writer.go#L197)

If compression is supported by and done in the content manager ([index version](https://github.com/kopia/kopia/blob/cc00388f6d2a2b78f03e2507b4b2f25519d1f7e4/repo/content/content_manager.go#L767) >= 2) then this change won't cause file reuploads because the content is hashed before being compressed

If, on the hand, compression is _not_ supported by the content manager, then the data will be compressed by the object writer. That will cause reuploads of files because the data will be compressed and then hashed, causing hash mismatches

From what I've been able to see, kopia switched to v2 index on Sept 2, 2021. For repositories that haven't upgraded yet I'd assume it would reupload any `StreamingFile`s now that they'd be compressed. For repositories that started out with v1 but then upgraded, I'm not sure what the behavior would be
ashmrtn: >Sorry it took me longer to respond back here.

it gave me a chance to dig into the code a bit more so it's all good. Mostly me commenting here has been me trying to make sure I tie up all the loose ends and give the best information I can on how it will affect things moving forward 😅 

Patch:
diff --git a/snapshot/policy/compression_policy.go b/snapshot/policy/compression_policy.go
index 4b9e9374..4c782272 100644
--- a/snapshot/policy/compression_policy.go
+++ b/snapshot/policy/compression_policy.go
@@ -30,7 +30,7 @@ type CompressionPolicyDefinition struct {
 }
 
 // CompressorForFile returns compression name to be used for compressing a given file according to policy, using attributes such as name or size.
-func (p *CompressionPolicy) CompressorForFile(e fs.File) compression.Name {
+func (p *CompressionPolicy) CompressorForFile(e fs.Entry) compression.Name {
 	ext := filepath.Ext(e.Name())
 	size := e.Size()
 
diff --git a/snapshot/snapshotfs/upload.go b/snapshot/snapshotfs/upload.go
index d615eb00..f626a670 100644
--- a/snapshot/snapshotfs/upload.go
+++ b/snapshot/snapshotfs/upload.go
@@ -336,7 +336,7 @@ func (u *Uploader) uploadSymlinkInternal(ctx context.Context, relativePath strin
 	return de, nil
 }
 
-func (u *Uploader) uploadStreamingFileInternal(ctx context.Context, relativePath string, f fs.StreamingFile) (dirEntry *snapshot.DirEntry, ret error) {
+func (u *Uploader) uploadStreamingFileInternal(ctx context.Context, relativePath string, f fs.StreamingFile, pol *policy.Policy) (dirEntry *snapshot.DirEntry, ret error) {
 	reader, err := f.GetReader(ctx)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to get streaming file reader")
@@ -353,9 +353,12 @@ func (u *Uploader) uploadStreamingFileInternal(ctx context.Context, relativePath
 		u.Progress.FinishedFile(relativePath, ret)
 	}()
 
+	comp := pol.CompressionPolicy.CompressorForFile(f)
 	writer := u.repo.NewObjectWriter(ctx, object.WriterOptions{
 		Description: "STREAMFILE:" + f.Name(),
+		Compressor:  comp,
 	})
+
 	defer writer.Close() //nolint:errcheck
 
 	written, err := u.copyWithProgress(writer, reader)
@@ -919,7 +922,7 @@ func (u *Uploader) processSingle(
 	case fs.StreamingFile:
 		atomic.AddInt32(&u.stats.NonCachedFiles, 1)
 
-		de, err := u.uploadStreamingFileInternal(ctx, entryRelativePath, entry)
+		de, err := u.uploadStreamingFileInternal(ctx, entryRelativePath, entry, policyTree.Child(entry.Name()).EffectivePolicy())
 
 		return u.processEntryUploadResult(ctx, de, err, entryRelativePath, parentDirBuilder,
 			policyTree.EffectivePolicy().ErrorHandlingPolicy.IgnoreFileErrors.OrDefault(false),


Test patch:
diff --git a/snapshot/snapshotfs/upload_test.go b/snapshot/snapshotfs/upload_test.go
index fbb026c8..01a262b4 100644
--- a/snapshot/snapshotfs/upload_test.go
+++ b/snapshot/snapshotfs/upload_test.go
@@ -873,6 +873,42 @@ func TestUpload_VirtualDirectoryWithStreamingFile(t *testing.T) {
 	}
 }
 
+func TestUpload_VirtualDirectoryWithStreamingFile_WithCompression(t *testing.T) {
+	ctx := testlogging.Context(t)
+	th := newUploadTestHarness(ctx, t)
+
+	defer th.cleanup()
+
+	u := NewUploader(th.repo)
+
+	pol := *policy.DefaultPolicy
+	pol.CompressionPolicy.CompressorName = "pgzip"
+
+	policyTree := policy.BuildTree(nil, &pol)
+
+	// Create a temporary file with test data. Want something compressible but
+	// small so we don't trigger dedupe.
+	content := []byte(strings.Repeat("a", 4096))
+	r := io.NopCloser(bytes.NewReader(content))
+
+	staticRoot := virtualfs.NewStaticDirectory("rootdir", []fs.Entry{
+		virtualfs.StreamingFileFromReader("stream-file", r),
+	})
+
+	man, err := u.Upload(ctx, staticRoot, policyTree, snapshot.SourceInfo{})
+	require.NoError(t, err)
+
+	assert.Equal(t, int32(0), atomic.LoadInt32(&man.Stats.CachedFiles), "cached file count")
+	assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.NonCachedFiles), "non-cached file count")
+	assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.TotalDirectoryCount), "directory count")
+	assert.Equal(t, int32(1), atomic.LoadInt32(&man.Stats.TotalFileCount), "total file count")
+
+	// Write out pending data so the below size check compares properly.
+	require.NoError(t, th.repo.Flush(ctx), "flushing repo")
+
+	assert.Less(t, testutil.MustGetTotalDirSize(t, th.repoDir), int64(14000))
+}
+
 func TestUpload_VirtualDirectoryWithStreamingFileWithModTime(t *testing.T) {
 	content := []byte("Streaming Temporary file content")
 	mt := time.Date(2021, 1, 2, 3, 4, 5, 0, time.UTC)