[AzDatalake] File Client Upload/Download Support (#21261)

* Enable gocritic during linting (#20715) Enabled gocritic's evalOrder to catch dependencies on undefined behavior on return statements. Updated to latest version of golangci-lint. Fixed issue in azblob flagged by latest linter. * Cosmos DB: Enable merge support (#20716) * Adding header and value * Wiring and tests * format * Fixing value * change log * [azservicebus, azeventhubs] Stress test and logging improvement (#20710) Logging improvements: * Updating the logging to print more tracing information (per-link) in prep for the bigger release coming up. * Trimming out some of the verbose logging, seeing if I can get it a bit more reasonable. Stress tests: * Add a timestamp to the log name we generate and also default to append, not overwrite. * Use 0.5 cores, 0.5GB as our baseline. Some pods use more and I'll tune them more later. * update proxy version (#20712) Co-authored-by: Scott Beddall <[email protected]> * Return an error when you try to send a message that's too large. (#20721) This now works just like the message batch - you'll get an ErrMessageTooLarge if you attempt to send a message that's too large for the link's configured size. NOTE: there's a patch to `internal/go-amqp/Sender.go` to match what's in go-amqp's main so it returns a programmatically useful error when the message is too large. Fixes #20647 * Changes in test that is failing in pipeline (#20693) * [azservicebus, azeventhubs] Treat 'entity full' as a fatal error (#20722) When the remote entity is full we get a resource-limit-exceeded condition. This isn't something we should keep retrying on and it's best to just abort and let the user know immediately, rather than hoping it might eventually clear out. This affected both Event Hubs and Service Bus. Fixes #20647 * [azservicebus/azeventhubs] Redirect stderr and stdout to tee (#20726) * Update changelog with latest features (#20730) * Update changelog with latest features Prepare for upcoming release. * bump minor version * pass along the artifact name so we can override it later (#20732) Co-authored-by: scbedd <[email protected]> * [azeventhubs] Fixing checkpoint store race condition (#20727) The checkpoint store wasn't guarding against multiple owners claiming for the first time - fixing this by using IfNoneMatch Fixes #20717 * Fix azidentity troubleshooting guide link (#20736) * [Release] sdk/resourcemanager/paloaltonetworksngfw/armpanngfw/0.1.0 (#20437) * [Release] sdk/resourcemanager/paloaltonetworksngfw/armpanngfw/0.1.0 generation from spec commit: 85fb4ac6f8bfefd179e6c2632976a154b5c9ff04 * client factory * fix * fix * update * add sdk/resourcemanager/postgresql/armpostgresql live test (#20685) * add sdk/resourcemanager/postgresql/armpostgresql live test * update assets.json * set subscriptionId default value * format * add sdk/resourcemanager/eventhub/armeventhub live test (#20686) * add sdk/resourcemanager/eventhub/armeventhub live test * update assets * add sdk/resourcemanager/compute/armcompute live test (#20048) * add sdk/resourcemanager/compute/armcompute live test * skus filter * fix subscriptionId default value * fix * gofmt * update recording * sdk/resourcemanager/network/armnetwork live test (#20331) * sdk/resourcemanager/network/armnetwork live test * update subscriptionId default value * update recording * add sdk/resourcemanager/cosmos/armcosmos live test (#20705) * add sdk/resourcemanager/cosmos/armcosmos live test * update assets.json * update assets.json * update assets.json * update assets.json * Increment package version after release of azcore (#20740) * [azeventhubs] Improperly resetting etag in the checkpoint store (#20737) We shouldn't be resetting the etag to nil - it's what we use to enforce a "single winner" when doing ownership claims. The bug here was two-fold: I had bad logic in my previous claim ownership, which I fixed in a previous PR, but we need to reflect that same constraint properly in our in-memory checkpoint store for these tests. * Eng workflows sync and branch cleanup additions (#20743) Co-authored-by: James Suplizio <[email protected]> * [azeventhubs] Latest start position can also be inclusive (ie, get the latest message) (#20744) * Update GitHubEventProcessor version and remove pull_request_review procesing (#20751) Co-authored-by: James Suplizio <[email protected]> * Rename DisableAuthorityValidationAndInstanceDiscovery (#20746) * fix (#20707) * AzFile (#20739) * azfile: Fixing connection string parsing logic (#20798) * Fixing connection string parse logic * Update README * [azadmin] fix flaky test (#20758) * fix flaky test * charles suggestion * Prepare azidentity v1.3.0 for release (#20756) * Fix broken podman link (#20801) Co-authored-by: Wes Haggard <[email protected]> * [azquery] update doc comments (#20755) * update doc comments * update statistics and visualization generation * prep-for-release * Fixed contribution section (#20752) Co-authored-by: Bob Tabor <[email protected]> * [azeventhubs,azservicebus] Some API cleanup, renames (#20754) * Adding options to UpdateCheckpoint(), just for future potential expansion * Make Offset an int64, not a *int64 (it's not optional, it'll always come back with ReceivedEvents) * Adding more logging into the checkpoint store. * Point all imports at the production go-amqp * Add supporting features to enable distributed tracing (#20301) (#20708) * Add supporting features to enable distributed tracing This includes new internal pipeline policies and other supporting types. See the changelog for a full description. Added some missing doc comments. * fix linter issue * add net.peer.name trace attribute sequence custom HTTP header policy before logging policy. sequence logging policy after HTTP trace policy. keep body download policy at the end. * add span for iterating over pages * Restore ARM CAE support for azcore beta (#20657) This reverts commit 9020972. * Upgrade to stable azcore (#20808) * Increment package version after release of data/azcosmos (#20807) * Updating changelog (#20810) * Add fake package to azcore (#20711) * Add fake package to azcore This is the supporting infrastructure for the generated SDK fakes. * fix doc comment * Updating CHANGELOG.md (#20809) * changelog (#20811) * Increment package version after release of storage/azfile (#20813) * Update changelog (azblob) (#20815) * Updating CHANGELOG.md * Update the changelog with correct version * [azquery] migration guide (#20742) * migration guide * Charles feedback * Richard feedback --------- Co-authored-by: Charles Lowell <[email protected]> * Increment package version after release of monitor/azquery (#20820) * [keyvault] prep for release (#20819) * prep for release * perf tests * update date * added all upload methods * added more tests for upload stream * added more tests * added downloaders * added more tests * cleanup * feedback --------- Co-authored-by: Joel Hendrix <[email protected]> Co-authored-by: Matias Quaranta <[email protected]> Co-authored-by: Richard Park <[email protected]> Co-authored-by: Azure SDK Bot <[email protected]> Co-authored-by: Scott Beddall <[email protected]> Co-authored-by: siminsavani-msft <[email protected]> Co-authored-by: scbedd <[email protected]> Co-authored-by: Charles Lowell <[email protected]> Co-authored-by: Peng Jiahui <[email protected]> Co-authored-by: James Suplizio <[email protected]> Co-authored-by: Sourav Gupta <[email protected]> Co-authored-by: gracewilcox <[email protected]> Co-authored-by: Wes Haggard <[email protected]> Co-authored-by: Bob Tabor <[email protected]> Co-authored-by: Bob Tabor <[email protected]>
Azure · Jul 27, 2023 · ad288fb · ad288fb
1 parent 68d465f
commit ad288fb
Show file tree

Hide file tree

Showing 18 changed files with 2,277 additions and 46 deletions.
diff --git a/sdk/storage/azdatalake/file/chunkwriting.go b/sdk/storage/azdatalake/file/chunkwriting.go
@@ -0,0 +1,193 @@
+//go:build go1.18
+// +build go1.18
+
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+package file
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/streaming"
+	"io"
+	"sync"
+)
+
+// chunkWriter provides methods to upload chunks that represent a file to a server.
+// This allows us to provide a local implementation that fakes the server for hermetic testing.
+type chunkWriter interface {
+	AppendData(context.Context, int64, io.ReadSeekCloser, *AppendDataOptions) (AppendDataResponse, error)
+	FlushData(context.Context, int64, *FlushDataOptions) (FlushDataResponse, error)
+}
+
+// bufferManager provides an abstraction for the management of buffers.
+// this is mostly for testing purposes, but does allow for different implementations without changing the algorithm.
+type bufferManager[T ~[]byte] interface {
+	// Acquire returns the channel that contains the pool of buffers.
+	Acquire() <-chan T
+
+	// Release releases the buffer back to the pool for reuse/cleanup.
+	Release(T)
+
+	// Grow grows the number of buffers, up to the predefined max.
+	// It returns the total number of buffers or an error.
+	// No error is returned if the number of buffers has reached max.
+	// This is called only from the reading goroutine.
+	Grow() (int, error)
+
+	// Free cleans up all buffers.
+	Free()
+}
+
+// copyFromReader copies a source io.Reader to file storage using concurrent uploads.
+func copyFromReader[T ~[]byte](ctx context.Context, src io.Reader, dst chunkWriter, options UploadStreamOptions, getBufferManager func(maxBuffers int, bufferSize int64) bufferManager[T]) error {
+	options.setDefaults()
+	actualSize := int64(0)
+	wg := sync.WaitGroup{}       // Used to know when all outgoing chunks have finished processing
+	errCh := make(chan error, 1) // contains the first error encountered during processing
+	var err error
+
+	buffers := getBufferManager(int(options.Concurrency), options.ChunkSize)
+	defer buffers.Free()
+
+	// this controls the lifetime of the uploading goroutines.
+	// if an error is encountered, cancel() is called which will terminate all uploads.
+	// NOTE: the ordering is important here.  cancel MUST execute before
+	// cleaning up the buffers so that any uploading goroutines exit first,
+	// releasing their buffers back to the pool for cleanup.
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// This goroutine grabs a buffer, reads from the stream into the buffer,
+	// then creates a goroutine to upload/stage the chunk.
+	for chunkNum := uint32(0); true; chunkNum++ {
+		var buffer T
+		select {
+		case buffer = <-buffers.Acquire():
+			// got a buffer
+		default:
+			// no buffer available; allocate a new buffer if possible
+			if _, err := buffers.Grow(); err != nil {
+				return err
+			}
+
+			// either grab the newly allocated buffer or wait for one to become available
+			buffer = <-buffers.Acquire()
+		}
+
+		var n int
+		n, err = io.ReadFull(src, buffer)
+
+		if n > 0 {
+			// some data was read, upload it
+			wg.Add(1) // We're posting a buffer to be sent
+
+			// NOTE: we must pass chunkNum as an arg to our goroutine else
+			// it's captured by reference and can change underneath us!
+			go func(chunkNum uint32) {
+				// Upload the outgoing chunk, matching the number of bytes read
+				offset := int64(chunkNum) * options.ChunkSize
+				appendDataOpts := options.getAppendDataOptions()
+				actualSize += int64(len(buffer[:n]))
+				_, err := dst.AppendData(ctx, offset, streaming.NopCloser(bytes.NewReader(buffer[:n])), appendDataOpts)
+				if err != nil {
+					select {
+					case errCh <- err:
+						// error was set
+					default:
+						// some other error is already set
+					}
+					cancel()
+				}
+				buffers.Release(buffer) // The goroutine reading from the stream can reuse this buffer now
+
+				// signal that the chunk has been staged.
+				// we MUST do this after attempting to write to errCh
+				// to avoid it racing with the reading goroutine.
+				wg.Done()
+			}(chunkNum)
+		} else {
+			// nothing was read so the buffer is empty, send it back for reuse/clean-up.
+			buffers.Release(buffer)
+		}
+
+		if err != nil { // The reader is done, no more outgoing buffers
+			if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+				// these are expected errors, we don't surface those
+				err = nil
+			} else {
+				// some other error happened, terminate any outstanding uploads
+				cancel()
+			}
+			break
+		}
+	}
+
+	wg.Wait() // Wait for all outgoing chunks to complete
+
+	if err != nil {
+		// there was an error reading from src, favor this error over any error during staging
+		return err
+	}
+
+	select {
+	case err = <-errCh:
+		// there was an error during staging
+		return err
+	default:
+		// no error was encountered
+	}
+
+	// All chunks uploaded, return nil error
+	flushOpts := options.getFlushDataOptions()
+	_, err = dst.FlushData(ctx, actualSize, flushOpts)
+	return err
+}
+
+// mmbPool implements the bufferManager interface.
+// it uses anonymous memory mapped files for buffers.
+// don't use this type directly, use newMMBPool() instead.
+type mmbPool struct {
+	buffers chan mmb
+	count   int
+	max     int
+	size    int64
+}
+
+func newMMBPool(maxBuffers int, bufferSize int64) bufferManager[mmb] {
+	return &mmbPool{
+		buffers: make(chan mmb, maxBuffers),
+		max:     maxBuffers,
+		size:    bufferSize,
+	}
+}
+
+func (pool *mmbPool) Acquire() <-chan mmb {
+	return pool.buffers
+}
+
+func (pool *mmbPool) Grow() (int, error) {
+	if pool.count < pool.max {
+		buffer, err := newMMB(pool.size)
+		if err != nil {
+			return 0, err
+		}
+		pool.buffers <- buffer
+		pool.count++
+	}
+	return pool.count, nil
+}
+
+func (pool *mmbPool) Release(buffer mmb) {
+	pool.buffers <- buffer
+}
+
+func (pool *mmbPool) Free() {
+	for i := 0; i < pool.count; i++ {
+		buffer := <-pool.buffers
+		buffer.delete()
+	}
+	pool.count = 0
+}