Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Conditionally write tiles. #70

Merged
merged 6 commits into from
Jan 9, 2024
Merged
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 50 additions & 3 deletions experimental/gcp-log/internal/storage/storage.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
package storage

import (
"bytes"
"context"
"errors"
"fmt"
Expand Down Expand Up @@ -376,14 +377,38 @@ func (c *Client) Sequence(ctx context.Context, leafhash []byte, leaf []byte) (ui
return 0, fmt.Errorf("couldn't create leafhash object: %w", err)
}
if err := wLeaf.Close(); err != nil {
return 0, fmt.Errorf("couldn't close writer for object %q", leafPath)
return 0, fmt.Errorf("couldn't close writer for object %q, %w", leafPath, err)
}

// All done!
return seq, nil
}
}

// assertContent checks that the content at `gcsPath` matches the passed in `data`.
func (c *Client) assertContent(ctx context.Context, gcsPath string, data []byte) (equal bool, err error) {
bkt := c.gcsClient.Bucket(c.bucket)

obj := bkt.Object(gcsPath)
r, err := obj.NewReader(ctx)
if err != nil {
klog.V(2).Infof("assertContent: failed to create reader for object %q in bucket %q: %v",
gcsPath, c.bucket, err)
return false, err
}
defer r.Close()

gcsData, err := io.ReadAll(r)
if err != nil {
return false, err
}

if bytes.Equal(gcsData, data) {
return true, nil
}
return false, nil
}

// StoreTile writes a tile out to GCS.
// Fully populated tiles are stored at the path corresponding to the level &
// index parameters, partially populated (i.e. right-hand edge) tiles are
Expand All @@ -405,12 +430,34 @@ func (c *Client) StoreTile(ctx context.Context, level, index uint64, tile *api.T
tPath := filepath.Join(layout.TilePath("", level, index, tileSize%256))
obj := bkt.Object(tPath)

w := obj.NewWriter(ctx)
// Tiles, partial or full, should only be written once.
w := obj.If(gcs.Conditions{DoesNotExist: true}).NewWriter(ctx)
jiggoha marked this conversation as resolved.
Show resolved Hide resolved
if c.otherCacheControl != "" {
w.ObjectAttrs.CacheControl = c.otherCacheControl
}
if _, err := w.Write(t); err != nil {
return fmt.Errorf("failed to write tile object %q to bucket %q: %w", tPath, c.bucket, err)
}
return w.Close()

if err := w.Close(); err != nil {
switch ee := err.(type) {
case *googleapi.Error:
// If we run into a precondition failure error, check that the object
// which exists contains the same content that we want to write.
if ee.Code == http.StatusPreconditionFailed {
if equal, err := c.assertContent(ctx, tPath, t); err != nil {
return fmt.Errorf("failed to read content of %q: %w", tPath, err)
} else if !equal {
return fmt.Errorf("assertion that tile content for %q has not changed failed", tPath)
} else if equal {
jiggoha marked this conversation as resolved.
Show resolved Hide resolved
klog.V(2).Infof("StoreTile: Tile already exists for level %d index %x ts: %x", level, index, tileSize)
jiggoha marked this conversation as resolved.
Show resolved Hide resolved
return nil
}
}
default:
return err
}
}

return nil
}
Loading