This repository has been archived by the owner on Aug 28, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
430 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
301 changes: 301 additions & 0 deletions
301
patches/com_github_buildbarn_bb_storage/expose_find_missing_queue.diff
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,301 @@ | ||
| diff --git pkg/blobstore/completenesschecking/BUILD.bazel pkg/blobstore/completenesschecking/BUILD.bazel | ||
| index 0c0e849..3b46bf7 100644 | ||
| --- pkg/blobstore/completenesschecking/BUILD.bazel | ||
| +++ pkg/blobstore/completenesschecking/BUILD.bazel | ||
| @@ -2,7 +2,10 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test") | ||
|
|
||
| go_library( | ||
| name = "go_default_library", | ||
| - srcs = ["completeness_checking_blob_access.go"], | ||
| + srcs = [ | ||
| + "completeness_checking_blob_access.go", | ||
| + "find_missing_queue.go" | ||
| + ], | ||
| importpath = "github.com/buildbarn/bb-storage/pkg/blobstore/completenesschecking", | ||
| visibility = ["//visibility:public"], | ||
| deps = [ | ||
| diff --git pkg/blobstore/completenesschecking/completeness_checking_blob_access.go pkg/blobstore/completenesschecking/completeness_checking_blob_access.go | ||
| index 93db060..b437e58 100644 | ||
| --- pkg/blobstore/completenesschecking/completeness_checking_blob_access.go | ||
| +++ pkg/blobstore/completenesschecking/completeness_checking_blob_access.go | ||
| @@ -8,80 +8,8 @@ import ( | ||
| "github.com/buildbarn/bb-storage/pkg/blobstore/buffer" | ||
| "github.com/buildbarn/bb-storage/pkg/digest" | ||
| "github.com/buildbarn/bb-storage/pkg/util" | ||
| - | ||
| - "google.golang.org/grpc/codes" | ||
| - "google.golang.org/grpc/status" | ||
| ) | ||
|
|
||
| -// findMissingQueue is a helper for calling BlobAccess.FindMissing() in | ||
| -// batches, as opposed to calling it for individual digests. | ||
| -type findMissingQueue struct { | ||
| - context context.Context | ||
| - instanceName digest.InstanceName | ||
| - contentAddressableStorage blobstore.BlobAccess | ||
| - batchSize int | ||
| - | ||
| - pending digest.SetBuilder | ||
| -} | ||
| - | ||
| -// deriveDigest converts a digest embedded into an action result from | ||
| -// the wire format to an in-memory representation. If that fails, we | ||
| -// assume that some data corruption has occurred. In that case, we | ||
| -// should destroy the action result. | ||
| -func (q *findMissingQueue) deriveDigest(blobDigest *remoteexecution.Digest) (digest.Digest, error) { | ||
| - derivedDigest, err := q.instanceName.NewDigestFromProto(blobDigest) | ||
| - if err != nil { | ||
| - return digest.BadDigest, util.StatusWrapWithCode(err, codes.NotFound, "Action result contained malformed digest") | ||
| - } | ||
| - return derivedDigest, err | ||
| -} | ||
| - | ||
| -// Add a digest to the list of digests that are pending to be checked | ||
| -// for existence in the Content Addressable Storage. | ||
| -func (q *findMissingQueue) add(blobDigest *remoteexecution.Digest) error { | ||
| - if blobDigest != nil { | ||
| - derivedDigest, err := q.deriveDigest(blobDigest) | ||
| - if err != nil { | ||
| - return err | ||
| - } | ||
| - | ||
| - if q.pending.Length() >= q.batchSize { | ||
| - if err := q.finalize(); err != nil { | ||
| - return err | ||
| - } | ||
| - q.pending = digest.NewSetBuilder() | ||
| - } | ||
| - q.pending.Add(derivedDigest) | ||
| - } | ||
| - return nil | ||
| -} | ||
| - | ||
| -// AddDirectory adds all digests contained with a directory to the list | ||
| -// of digests pending to be checked for existence. | ||
| -func (q *findMissingQueue) addDirectory(directory *remoteexecution.Directory) error { | ||
| - if directory == nil { | ||
| - return nil | ||
| - } | ||
| - for _, child := range directory.Files { | ||
| - if err := q.add(child.Digest); err != nil { | ||
| - return err | ||
| - } | ||
| - } | ||
| - return nil | ||
| -} | ||
| - | ||
| -// Finalize by checking the last batch of digests for existence. | ||
| -func (q *findMissingQueue) finalize() error { | ||
| - missing, err := q.contentAddressableStorage.FindMissing(q.context, q.pending.Build()) | ||
| - if err != nil { | ||
| - return util.StatusWrap(err, "Failed to determine existence of child objects") | ||
| - } | ||
| - if digest, ok := missing.First(); ok { | ||
| - return status.Errorf(codes.NotFound, "Object %s referenced by the action result is not present in the Content Addressable Storage", digest) | ||
| - } | ||
| - return nil | ||
| -} | ||
| - | ||
| type completenessCheckingBlobAccess struct { | ||
| blobstore.BlobAccess | ||
| contentAddressableStorage blobstore.BlobAccess | ||
| @@ -113,13 +41,12 @@ func NewCompletenessCheckingBlobAccess(actionCache blobstore.BlobAccess, content | ||
| } | ||
|
|
||
| func (ba *completenessCheckingBlobAccess) checkCompleteness(ctx context.Context, instanceName digest.InstanceName, actionResult *remoteexecution.ActionResult) error { | ||
| - findMissingQueue := findMissingQueue{ | ||
| - context: ctx, | ||
| - instanceName: instanceName, | ||
| - contentAddressableStorage: ba.contentAddressableStorage, | ||
| - batchSize: ba.batchSize, | ||
| - pending: digest.NewSetBuilder(), | ||
| - } | ||
| + findMissingQueue := NewFindMissingQueue( | ||
| + ctx, | ||
| + instanceName, | ||
| + ba.contentAddressableStorage, | ||
| + ba.batchSize, | ||
| + ) | ||
|
|
||
| // Iterate over all remoteexecution.Digest fields contained | ||
| // within the ActionResult. Check the existence of output | ||
| @@ -127,19 +54,19 @@ func (ba *completenessCheckingBlobAccess) checkCompleteness(ctx context.Context, | ||
| // later on. GetTree() may not necessarily cause those objects | ||
| // to be touched. | ||
| for _, outputFile := range actionResult.OutputFiles { | ||
| - if err := findMissingQueue.add(outputFile.Digest); err != nil { | ||
| + if err := findMissingQueue.Add(outputFile.Digest); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| for _, outputDirectory := range actionResult.OutputDirectories { | ||
| - if err := findMissingQueue.add(outputDirectory.TreeDigest); err != nil { | ||
| + if err := findMissingQueue.Add(outputDirectory.TreeDigest); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| - if err := findMissingQueue.add(actionResult.StdoutDigest); err != nil { | ||
| + if err := findMissingQueue.Add(actionResult.StdoutDigest); err != nil { | ||
| return err | ||
| } | ||
| - if err := findMissingQueue.add(actionResult.StderrDigest); err != nil { | ||
| + if err := findMissingQueue.Add(actionResult.StderrDigest); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| @@ -147,7 +74,7 @@ func (ba *completenessCheckingBlobAccess) checkCompleteness(ctx context.Context, | ||
| // within output directories (remoteexecution.Tree objects) | ||
| // referenced by the ActionResult. | ||
| for _, outputDirectory := range actionResult.OutputDirectories { | ||
| - treeDigest, err := findMissingQueue.deriveDigest(outputDirectory.TreeDigest) | ||
| + treeDigest, err := findMissingQueue.DeriveDigest(outputDirectory.TreeDigest) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| @@ -156,16 +83,16 @@ func (ba *completenessCheckingBlobAccess) checkCompleteness(ctx context.Context, | ||
| return util.StatusWrapf(err, "Failed to fetch output directory %#v", outputDirectory.Path) | ||
| } | ||
| tree := treeMessage.(*remoteexecution.Tree) | ||
| - if err := findMissingQueue.addDirectory(tree.Root); err != nil { | ||
| + if err := findMissingQueue.AddDirectory(tree.Root); err != nil { | ||
| return err | ||
| } | ||
| for _, child := range tree.Children { | ||
| - if err := findMissingQueue.addDirectory(child); err != nil { | ||
| + if err := findMissingQueue.AddDirectory(child); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| - return findMissingQueue.finalize() | ||
| + return findMissingQueue.Finalize() | ||
| } | ||
|
|
||
| func (ba *completenessCheckingBlobAccess) Get(ctx context.Context, digest digest.Digest) buffer.Buffer { | ||
| diff --git pkg/blobstore/completenesschecking/completeness_checking_blob_access_test.go pkg/blobstore/completenesschecking/completeness_checking_blob_access_test.go | ||
| index 8f3d5fc..69f4938 100644 | ||
| --- pkg/blobstore/completenesschecking/completeness_checking_blob_access_test.go | ||
| +++ pkg/blobstore/completenesschecking/completeness_checking_blob_access_test.go | ||
| @@ -56,7 +56,7 @@ func TestCompletenessCheckingBlobAccess(t *testing.T) { | ||
| buffer.Reparable(actionDigest, repairFunc.Call))) | ||
|
|
||
| _, err := completenessCheckingBlobAccess.Get(ctx, actionDigest).ToProto(&remoteexecution.ActionResult{}, 1000) | ||
| - require.Equal(t, err, status.Error(codes.NotFound, "Action result contained malformed digest: Unknown digest hash length: 24 characters")) | ||
| + require.Equal(t, err, status.Error(codes.NotFound, "Malformed digest found during FindMissing process: Unknown digest hash length: 24 characters")) | ||
| }) | ||
|
|
||
| t.Run("MissingInput", func(t *testing.T) { | ||
| @@ -92,7 +92,7 @@ func TestCompletenessCheckingBlobAccess(t *testing.T) { | ||
| nil) | ||
|
|
||
| _, err := completenessCheckingBlobAccess.Get(ctx, actionDigest).ToProto(&remoteexecution.ActionResult{}, 1000) | ||
| - require.Equal(t, err, status.Error(codes.NotFound, "Object 8b1a9953c4611296a827abf8c47804d7-5-hello referenced by the action result is not present in the Content Addressable Storage")) | ||
| + require.Equal(t, err, status.Error(codes.NotFound, "Referenced Object 8b1a9953c4611296a827abf8c47804d7-5-hello is not present in the Content Addressable Storage")) | ||
| }) | ||
|
|
||
| t.Run("FindMissingError", func(t *testing.T) { | ||
| diff --git pkg/blobstore/completenesschecking/find_missing_queue.go pkg/blobstore/completenesschecking/find_missing_queue.go | ||
| new file mode 100644 | ||
| index 0000000..be884c2 | ||
| --- /dev/null | ||
| +++ pkg/blobstore/completenesschecking/find_missing_queue.go | ||
| @@ -0,0 +1,95 @@ | ||
| +package completenesschecking | ||
| + | ||
| +import ( | ||
| + "context" | ||
| + | ||
| + remoteexecution "github.com/bazelbuild/remote-apis/build/bazel/remote/execution/v2" | ||
| + "github.com/buildbarn/bb-storage/pkg/blobstore" | ||
| + "github.com/buildbarn/bb-storage/pkg/digest" | ||
| + "github.com/buildbarn/bb-storage/pkg/util" | ||
| + | ||
| + "google.golang.org/grpc/codes" | ||
| + "google.golang.org/grpc/status" | ||
| +) | ||
| + | ||
| +// findMissingQueue is a helper for calling BlobAccess.FindMissing() in | ||
| +// batches, as opposed to calling it for individual digests. | ||
| +type findMissingQueue struct { | ||
| + context context.Context | ||
| + instanceName digest.InstanceName | ||
| + contentAddressableStorage blobstore.BlobAccess | ||
| + batchSize int | ||
| + | ||
| + pending digest.SetBuilder | ||
| +} | ||
| + | ||
| +func NewFindMissingQueue(context context.Context, instanceName digest.InstanceName, | ||
| + contentAddressableStorage blobstore.BlobAccess, | ||
| + batchSize int) findMissingQueue { | ||
| + return findMissingQueue { | ||
| + context: context, | ||
| + instanceName: instanceName, | ||
| + contentAddressableStorage: contentAddressableStorage, | ||
| + batchSize: batchSize, | ||
| + pending: digest.NewSetBuilder(), | ||
| + } | ||
| + | ||
| +} | ||
| + | ||
| +// DeriveDigest converts a digest embedded into an action result from | ||
| +// the wire format to an in-memory representation. If that fails, we | ||
| +// assume that some data corruption has occurred. In that case, we | ||
| +// should destroy the action result. | ||
| +func (q *findMissingQueue) DeriveDigest(blobDigest *remoteexecution.Digest) (digest.Digest, error) { | ||
| + derivedDigest, err := q.instanceName.NewDigestFromProto(blobDigest) | ||
| + if err != nil { | ||
| + return digest.BadDigest, util.StatusWrapWithCode(err, codes.NotFound, "Malformed digest found during FindMissing process") | ||
| + } | ||
| + return derivedDigest, err | ||
| +} | ||
| + | ||
| +// Add a digest to the list of digests that are pending to be checked | ||
| +// for existence in the Content Addressable Storage. | ||
| +func (q *findMissingQueue) Add(blobDigest *remoteexecution.Digest) error { | ||
| + if blobDigest != nil { | ||
| + derivedDigest, err := q.DeriveDigest(blobDigest) | ||
| + if err != nil { | ||
| + return err | ||
| + } | ||
| + | ||
| + if q.pending.Length() >= q.batchSize { | ||
| + if err := q.Finalize(); err != nil { | ||
| + return err | ||
| + } | ||
| + q.pending = digest.NewSetBuilder() | ||
| + } | ||
| + q.pending.Add(derivedDigest) | ||
| + } | ||
| + return nil | ||
| +} | ||
| + | ||
| +// AddDirectory adds all digests contained with a directory to the list | ||
| +// of digests pending to be checked for existence. | ||
| +func (q *findMissingQueue) AddDirectory(directory *remoteexecution.Directory) error { | ||
| + if directory == nil { | ||
| + return nil | ||
| + } | ||
| + for _, child := range directory.Files { | ||
| + if err := q.Add(child.Digest); err != nil { | ||
| + return err | ||
| + } | ||
| + } | ||
| + return nil | ||
| +} | ||
| + | ||
| +// Finalize by checking the last batch of digests for existence. | ||
| +func (q *findMissingQueue) Finalize() error { | ||
| + missing, err := q.contentAddressableStorage.FindMissing(q.context, q.pending.Build()) | ||
| + if err != nil { | ||
| + return util.StatusWrap(err, "Failed to determine existence of child objects") | ||
| + } | ||
| + if digest, ok := missing.First(); ok { | ||
| + return status.Errorf(codes.NotFound, "Referenced Object %s is not present in the Content Addressable Storage", digest) | ||
| + } | ||
| + return nil | ||
| +} | ||
| \ No newline at end of file | ||
| -- | ||
| 2.23.0 | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.