Skip to content

Commit 513b937

Browse files
tritonetritone
authored
feat(storage): takeover appendable object (#11977)
Adds a new method `NewWriterFromAppendableObject` which allows the caller to start writing to an appendable-but-unfinalized object in GCS. Conformance/emulator tests waiting on full support in the testbench for takeover. Integration tests pass locally, but cannot run in kokoro since zonal buckets are not yet available in the region. Also fixes a bug where Writer.Attrs is not set appropriately for unfinalized uploads in some cases.
1 parent 0facde6 commit 513b937

6 files changed

Lines changed: 360 additions & 34 deletions

File tree

‎storage/client.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -257,6 +257,9 @@ type openWriterParams struct {
257257
// conds - see `Writer.o.conds`.
258258
// Optional.
259259
conds *Conditions
260+
// appendGen -- object generation to write to.
261+
// Optional; required for taking over appendable objects only
262+
appendGen int64
260263
// encryptionKey - see `Writer.o.encryptionKey`
261264
// Optional.
262265
encryptionKey []byte
@@ -290,6 +293,8 @@ type openWriterParams struct {
290293
setFlush func(func() (int64, error))
291294
// setPipeWriter callback for reseting `Writer.pw` if needed.
292295
setPipeWriter func(*io.PipeWriter)
296+
// setTakeoverOffset callback for returning offset to start writing from to Writer.
297+
setTakeoverOffset func(int64)
293298
}
294299

295300
type newMultiRangeDownloaderParams struct {

‎storage/grpc_client.go‎

Lines changed: 38 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1767,6 +1767,21 @@ func (c *grpcStorageClient) OpenWriter(params *openWriterParams, opts ...storage
17671767
return nil, err
17681768
}
17691769

1770+
// If we are taking over an appendable object, send the first message here
1771+
// to get the append offset.
1772+
if params.appendGen > 0 {
1773+
// Create the buffer sender. This opens a stream and blocks until we
1774+
// get a response that tells us what offset to write from.
1775+
wbs, err := gw.newGRPCAppendTakeoverWriteBufferSender(params.ctx)
1776+
if err != nil {
1777+
return nil, fmt.Errorf("storage: creating buffer sender: %w", err)
1778+
}
1779+
// Propagate append offset to caller and buffer sending logic below.
1780+
params.setTakeoverOffset(wbs.takeoverOffset)
1781+
offset = wbs.takeoverOffset
1782+
gw.streamSender = wbs
1783+
}
1784+
17701785
// This function reads the data sent to the pipe and sends sets of messages
17711786
// on the gRPC client-stream as the buffer is filled.
17721787
go func() {
@@ -1778,17 +1793,19 @@ func (c *grpcStorageClient) OpenWriter(params *openWriterParams, opts ...storage
17781793
}
17791794

17801795
// Loop until there is an error or the Object has been finalized.
1796+
var o *storagepb.Object
17811797
for {
17821798
// Note: This blocks until either the buffer is full or EOF is read.
17831799
recvd, doneReading, err := gw.read()
17841800
if err != nil {
17851801
return err
17861802
}
17871803

1788-
var o *storagepb.Object
17891804
uploadBuff := func(ctx context.Context) error {
17901805
obj, err := gw.uploadBuffer(ctx, recvd, offset, doneReading)
1791-
o = obj
1806+
if obj != nil {
1807+
o = obj
1808+
}
17921809
return err
17931810
}
17941811

@@ -2620,6 +2637,14 @@ func newGRPCWriter(c *grpcStorageClient, s *settings, params *openWriterParams,
26202637
Resource: params.attrs.toProtoObject(params.bucket),
26212638
Appendable: proto.Bool(params.append),
26222639
}
2640+
var appendSpec *storagepb.AppendObjectSpec
2641+
if params.appendGen > 0 {
2642+
appendSpec = &storagepb.AppendObjectSpec{
2643+
Bucket: bucketResourceName(globalProjectAlias, params.bucket),
2644+
Object: params.attrs.Name,
2645+
Generation: params.appendGen,
2646+
}
2647+
}
26232648
// WriteObject doesn't support the generation condition, so use default.
26242649
if err := applyCondsProto("WriteObject", defaultGen, params.conds, spec); err != nil {
26252650
return nil, err
@@ -2635,6 +2660,7 @@ func newGRPCWriter(c *grpcStorageClient, s *settings, params *openWriterParams,
26352660
attrs: params.attrs,
26362661
conds: params.conds,
26372662
spec: spec,
2663+
appendSpec: appendSpec,
26382664
encryptionKey: params.encryptionKey,
26392665
settings: s,
26402666
progress: params.progress,
@@ -2663,6 +2689,7 @@ type gRPCWriter struct {
26632689
attrs *ObjectAttrs
26642690
conds *Conditions
26652691
spec *storagepb.WriteObjectSpec
2692+
appendSpec *storagepb.AppendObjectSpec
26662693
encryptionKey []byte
26672694
settings *settings
26682695
progress func(int64)
@@ -2700,17 +2727,22 @@ func drainInboundStream(stream storagepb.Storage_BidiWriteObjectClient) (object
27002727
}
27012728

27022729
func bidiWriteObjectRequest(buf []byte, offset int64, flush, finishWrite bool) *storagepb.BidiWriteObjectRequest {
2703-
return &storagepb.BidiWriteObjectRequest{
2704-
Data: &storagepb.BidiWriteObjectRequest_ChecksummedData{
2730+
var data *storagepb.BidiWriteObjectRequest_ChecksummedData
2731+
if buf != nil {
2732+
data = &storagepb.BidiWriteObjectRequest_ChecksummedData{
27052733
ChecksummedData: &storagepb.ChecksummedData{
27062734
Content: buf,
27072735
},
2708-
},
2736+
}
2737+
}
2738+
req := &storagepb.BidiWriteObjectRequest{
2739+
Data: data,
27092740
WriteOffset: offset,
27102741
FinishWrite: finishWrite,
27112742
Flush: flush,
27122743
StateLookup: flush,
27132744
}
2745+
return req
27142746
}
27152747

27162748
type gRPCBidiWriteBufferSender interface {
@@ -2938,7 +2970,7 @@ func (w *gRPCWriter) uploadBuffer(ctx context.Context, recvd int, start int64, d
29382970
if w.streamSender == nil {
29392971
if w.append {
29402972
// Appendable object semantics
2941-
w.streamSender, err = w.newGRPCAppendBidiWriteBufferSender()
2973+
w.streamSender, err = w.newGRPCAppendableObjectBufferSender()
29422974
} else if doneReading || w.forceOneShot {
29432975
// One shot semantics
29442976
w.streamSender, err = w.newGRPCOneshotBidiWriteBufferSender()

‎storage/grpc_writer.go‎

Lines changed: 40 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ type gRPCAppendBidiWriteBufferSender struct {
4545
forceFirstMessage bool
4646
progress func(int64)
4747
flushOffset int64
48+
takeoverOffset int64
4849

4950
// Fields used to report responses from the receive side of the stream
5051
// recvs is closed when the current recv goroutine is complete. recvErr is set
@@ -53,7 +54,8 @@ type gRPCAppendBidiWriteBufferSender struct {
5354
recvErr error
5455
}
5556

56-
func (w *gRPCWriter) newGRPCAppendBidiWriteBufferSender() (*gRPCAppendBidiWriteBufferSender, error) {
57+
// Use for a newly created appendable object.
58+
func (w *gRPCWriter) newGRPCAppendableObjectBufferSender() (*gRPCAppendBidiWriteBufferSender, error) {
5759
s := &gRPCAppendBidiWriteBufferSender{
5860
bucket: w.spec.GetResource().GetBucket(),
5961
raw: w.c.raw,
@@ -72,6 +74,36 @@ func (w *gRPCWriter) newGRPCAppendBidiWriteBufferSender() (*gRPCAppendBidiWriteB
7274
return s, nil
7375
}
7476

77+
// Use for a takeover of an appendable object.
78+
// Unlike newGRPCAppendableObjectBufferSender, this blocks until the stream is
79+
// open because it needs to get the append offset from the server.
80+
func (w *gRPCWriter) newGRPCAppendTakeoverWriteBufferSender(ctx context.Context) (*gRPCAppendBidiWriteBufferSender, error) {
81+
s := &gRPCAppendBidiWriteBufferSender{
82+
bucket: w.spec.GetResource().GetBucket(),
83+
raw: w.c.raw,
84+
settings: w.c.settings,
85+
firstMessage: &storagepb.BidiWriteObjectRequest{
86+
FirstMessage: &storagepb.BidiWriteObjectRequest_AppendObjectSpec{
87+
AppendObjectSpec: w.appendSpec,
88+
},
89+
},
90+
objectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs),
91+
finalizeOnClose: w.finalizeOnClose,
92+
forceFirstMessage: true,
93+
progress: w.progress,
94+
}
95+
if err := s.connect(ctx); err != nil {
96+
return nil, fmt.Errorf("storage: opening appendable write stream: %w", err)
97+
}
98+
_, err := s.sendOnConnectedStream(nil, 0, true, false, true)
99+
if err != nil {
100+
return nil, err
101+
}
102+
firstResp := <-s.recvs
103+
s.takeoverOffset = firstResp.GetResource().GetSize()
104+
return s, nil
105+
}
106+
75107
func (s *gRPCAppendBidiWriteBufferSender) connect(ctx context.Context) (err error) {
76108
err = func() error {
77109
// If this is a forced first message, we've already determined it's safe to
@@ -86,6 +118,10 @@ func (s *gRPCAppendBidiWriteBufferSender) connect(ctx context.Context) (err erro
86118
if s.firstMessage.GetAppendObjectSpec().GetWriteHandle() != nil {
87119
return nil
88120
}
121+
// Also always okay to reconnect if there is a generation.
122+
if s.firstMessage.GetAppendObjectSpec().GetGeneration() != 0 {
123+
return nil
124+
}
89125

90126
// We can also reconnect if the first message has an if_generation_match or
91127
// if_metageneration_match condition. Note that negative conditions like
@@ -283,13 +319,15 @@ func (s *gRPCAppendBidiWriteBufferSender) sendOnConnectedStream(buf []byte, offs
283319
if flushOffset < rSize {
284320
flushOffset = rSize
285321
}
322+
if resp.GetResource() != nil {
323+
obj = resp.GetResource()
324+
}
286325
}
287326
if s.flushOffset < flushOffset {
288327
s.flushOffset = flushOffset
289328
s.progress(s.flushOffset)
290329
}
291330
}
292-
293331
return
294332
}
295333

@@ -321,7 +359,6 @@ func (s *gRPCAppendBidiWriteBufferSender) sendBuffer(ctx context.Context, buf []
321359
s.forceFirstMessage = true
322360
continue
323361
}
324-
325362
return
326363
}
327364
}

0 commit comments

Comments
 (0)