Skip to content

Commit 757507a

Browse files
tasherif-msftjhendrixMSFTealsurrichardpark-msftazure-sdk
authored
[AzDatalake] Directory Client Implementation (#21283)
* 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 dir methods * small fixes * added rescursive set acl methods * recursive support * added sas and tests --------- 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]>
1 parent ad288fb commit 757507a

File tree

15 files changed

+2582
-221
lines changed

15 files changed

+2582
-221
lines changed

sdk/storage/azdatalake/assets.json

+1-1
Original file line numberDiff line numberDiff line change
@@ -2,5 +2,5 @@
22
"AssetsRepo": "Azure/azure-sdk-assets",
33
"AssetsRepoPrefixPath": "go",
44
"TagPrefix": "go/storage/azdatalake",
5-
"Tag": "go/storage/azdatalake_db1de4a48b"
5+
"Tag": "go/storage/azdatalake_9dd1cc3e0e"
66
}

sdk/storage/azdatalake/directory/client.go

+169-34
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,17 @@ import (
1212
"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
1313
"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
1414
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/blockblob"
15+
"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/datalakeerror"
1516
"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/internal/base"
1617
"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/internal/exported"
1718
"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/internal/generated"
19+
"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/internal/path"
1820
"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/internal/shared"
21+
"github.com/Azure/azure-sdk-for-go/sdk/storage/azdatalake/sas"
22+
"net/http"
23+
"net/url"
24+
"strings"
25+
"time"
1926
)
2027

2128
// ClientOptions contains the optional parameters when creating a Client.
@@ -38,7 +45,7 @@ func NewClient(directoryURL string, cred azcore.TokenCredential, options *Client
3845
}
3946
base.SetPipelineOptions((*base.ClientOptions)(conOptions), &plOpts)
4047

41-
azClient, err := azcore.NewClient(shared.FileClient, exported.ModuleVersion, plOpts, &conOptions.ClientOptions)
48+
azClient, err := azcore.NewClient(shared.DirectoryClient, exported.ModuleVersion, plOpts, &conOptions.ClientOptions)
4249
if err != nil {
4350
return nil, err
4451
}
@@ -154,6 +161,10 @@ func (d *Client) blobClient() *blockblob.Client {
154161
return blobClient
155162
}
156163

164+
func (d *Client) getClientOptions() *base.ClientOptions {
165+
return base.GetCompositeClientOptions((*base.CompositeClient[generated.PathClient, generated.PathClient, blockblob.Client])(d))
166+
}
167+
157168
func (d *Client) sharedKey() *exported.SharedKeyCredential {
158169
return base.SharedKeyComposite((*base.CompositeClient[generated.PathClient, generated.PathClient, blockblob.Client])(d))
159170
}
@@ -176,63 +187,187 @@ func (d *Client) BlobURL() string {
176187

177188
// Create creates a new directory (dfs1).
178189
func (d *Client) Create(ctx context.Context, options *CreateOptions) (CreateResponse, error) {
179-
return CreateResponse{}, nil
190+
lac, mac, httpHeaders, createOpts, cpkOpts := options.format()
191+
resp, err := d.generatedDirClientWithDFS().Create(ctx, createOpts, httpHeaders, lac, mac, nil, cpkOpts)
192+
err = exported.ConvertToDFSError(err)
193+
return resp, err
180194
}
181195

182-
// Delete removes the directory (dfs1).
196+
// Delete deletes directory and any path under it (dfs1).
183197
func (d *Client) Delete(ctx context.Context, options *DeleteOptions) (DeleteResponse, error) {
184-
//TODO: pass recursive = true
185-
return DeleteResponse{}, nil
198+
lac, mac, deleteOpts := path.FormatDeleteOptions(options, true)
199+
resp, err := d.generatedDirClientWithDFS().Delete(ctx, deleteOpts, lac, mac)
200+
err = exported.ConvertToDFSError(err)
201+
return resp, err
186202
}
187203

188-
// GetProperties returns the properties of the directory (blob3). #TODO: we may just move this method to path client
204+
// GetProperties gets the properties of a directory (blob3)
189205
func (d *Client) GetProperties(ctx context.Context, options *GetPropertiesOptions) (GetPropertiesResponse, error) {
190-
// TODO: format blob response to path response
191-
return GetPropertiesResponse{}, nil
206+
opts := path.FormatGetPropertiesOptions(options)
207+
var respFromCtx *http.Response
208+
ctxWithResp := runtime.WithCaptureResponse(ctx, &respFromCtx)
209+
resp, err := d.blobClient().GetProperties(ctxWithResp, opts)
210+
newResp := path.FormatGetPropertiesResponse(&resp, respFromCtx)
211+
err = exported.ConvertToDFSError(err)
212+
return newResp, err
192213
}
193214

194-
// Rename renames the directory (dfs1).
215+
func (d *Client) renamePathInURL(newName string) (string, string, string) {
216+
endpoint := d.DFSURL()
217+
separator := "/"
218+
// Find the index of the last occurrence of the separator
219+
lastIndex := strings.LastIndex(endpoint, separator)
220+
// Split the string based on the last occurrence of the separator
221+
firstPart := endpoint[:lastIndex] // From the beginning of the string to the last occurrence of the separator
222+
newPathURL, newBlobURL := shared.GetURLs(runtime.JoinPaths(firstPart, newName))
223+
parsedNewURL, _ := url.Parse(d.DFSURL())
224+
return parsedNewURL.Path, newPathURL, newBlobURL
225+
}
226+
227+
// Rename renames a directory (dfs1)
195228
func (d *Client) Rename(ctx context.Context, newName string, options *RenameOptions) (RenameResponse, error) {
196-
return RenameResponse{}, nil
229+
newPathWithoutURL, newBlobURL, newPathURL := d.renamePathInURL(newName)
230+
lac, mac, smac, createOpts := path.FormatRenameOptions(options, newPathWithoutURL)
231+
var newBlobClient *blockblob.Client
232+
var err error
233+
if d.identityCredential() != nil {
234+
newBlobClient, err = blockblob.NewClient(newBlobURL, *d.identityCredential(), nil)
235+
} else if d.sharedKey() != nil {
236+
blobSharedKey, _ := d.sharedKey().ConvertToBlobSharedKey()
237+
newBlobClient, err = blockblob.NewClientWithSharedKeyCredential(newBlobURL, blobSharedKey, nil)
238+
} else {
239+
newBlobClient, err = blockblob.NewClientWithNoCredential(newBlobURL, nil)
240+
}
241+
if err != nil {
242+
return RenameResponse{}, err
243+
}
244+
newDirClient := (*Client)(base.NewPathClient(newPathURL, newBlobURL, newBlobClient, d.generatedDirClientWithDFS().InternalClient().WithClientName(shared.DirectoryClient), d.sharedKey(), d.identityCredential(), d.getClientOptions()))
245+
resp, err := newDirClient.generatedDirClientWithDFS().Create(ctx, createOpts, nil, lac, mac, smac, nil)
246+
return RenameResponse{
247+
Response: resp,
248+
NewDirectoryClient: newDirClient,
249+
}, exported.ConvertToDFSError(err)
197250
}
198251

199-
// SetAccessControl sets the owner, owning group, and permissions for a file or directory (dfs1).
252+
// SetAccessControl sets the owner, owning group, and permissions for a directory (dfs1).
200253
func (d *Client) SetAccessControl(ctx context.Context, options *SetAccessControlOptions) (SetAccessControlResponse, error) {
201-
return SetAccessControlResponse{}, nil
254+
opts, lac, mac, err := path.FormatSetAccessControlOptions(options)
255+
if err != nil {
256+
return SetAccessControlResponse{}, err
257+
}
258+
resp, err := d.generatedDirClientWithDFS().SetAccessControl(ctx, opts, lac, mac)
259+
err = exported.ConvertToDFSError(err)
260+
return resp, err
202261
}
203262

204-
// SetAccessControlRecursive sets the owner, owning group, and permissions for a file or directory (dfs1).
205-
func (d *Client) SetAccessControlRecursive(ctx context.Context, options *SetAccessControlRecursiveOptions) (SetAccessControlRecursiveResponse, error) {
206-
// TODO explicitly pass SetAccessControlRecursiveMode
207-
return SetAccessControlRecursiveResponse{}, nil
263+
func (d *Client) setAccessControlHelper(mode generated.PathSetAccessControlRecursiveMode, listOptions *generated.PathClientSetAccessControlRecursiveOptions) *runtime.Pager[SetAccessControlRecursiveResponse] {
264+
return runtime.NewPager(runtime.PagingHandler[SetAccessControlRecursiveResponse]{
265+
More: func(page SetAccessControlRecursiveResponse) bool {
266+
return page.Continuation != nil && len(*page.Continuation) > 0
267+
},
268+
Fetcher: func(ctx context.Context, page *SetAccessControlRecursiveResponse) (SetAccessControlRecursiveResponse, error) {
269+
var req *policy.Request
270+
var err error
271+
if page == nil {
272+
req, err = d.generatedDirClientWithDFS().SetAccessControlRecursiveCreateRequest(ctx, mode, listOptions)
273+
err = exported.ConvertToDFSError(err)
274+
} else {
275+
listOptions.Continuation = page.Continuation
276+
req, err = d.generatedDirClientWithDFS().SetAccessControlRecursiveCreateRequest(ctx, mode, listOptions)
277+
err = exported.ConvertToDFSError(err)
278+
}
279+
if err != nil {
280+
return SetAccessControlRecursiveResponse{}, err
281+
}
282+
resp, err := d.generatedDirClientWithDFS().InternalClient().Pipeline().Do(req)
283+
err = exported.ConvertToDFSError(err)
284+
if err != nil {
285+
return SetAccessControlRecursiveResponse{}, err
286+
}
287+
if !runtime.HasStatusCode(resp, http.StatusOK) {
288+
return SetAccessControlRecursiveResponse{}, runtime.NewResponseError(resp)
289+
}
290+
newResp, err := d.generatedDirClientWithDFS().SetAccessControlRecursiveHandleResponse(resp)
291+
return newResp, exported.ConvertToDFSError(err)
292+
},
293+
})
294+
208295
}
209296

210-
// UpdateAccessControlRecursive updates the owner, owning group, and permissions for a file or directory (dfs1).
211-
func (d *Client) UpdateAccessControlRecursive(ctx context.Context, options *UpdateAccessControlRecursiveOptions) (UpdateAccessControlRecursiveResponse, error) {
212-
// TODO explicitly pass SetAccessControlRecursiveMode
213-
return SetAccessControlRecursiveResponse{}, nil
297+
// NewSetAccessControlRecursivePager sets the owner, owning group, and permissions for a directory (dfs1).
298+
func (d *Client) NewSetAccessControlRecursivePager(ACL string, options *SetAccessControlRecursiveOptions) *runtime.Pager[SetAccessControlRecursiveResponse] {
299+
mode, listOptions := options.format(ACL, "set")
300+
return d.setAccessControlHelper(mode, listOptions)
214301
}
215302

216-
// GetAccessControl gets the owner, owning group, and permissions for a file or directory (dfs1).
217-
func (d *Client) GetAccessControl(ctx context.Context, options *GetAccessControlOptions) (GetAccessControlResponse, error) {
218-
return GetAccessControlResponse{}, nil
303+
// NewUpdateAccessControlRecursivePager updates the owner, owning group, and permissions for a directory (dfs1).
304+
func (d *Client) NewUpdateAccessControlRecursivePager(ACL string, options *UpdateAccessControlRecursiveOptions) *runtime.Pager[UpdateAccessControlRecursiveResponse] {
305+
mode, listOptions := options.format(ACL, "modify")
306+
return d.setAccessControlHelper(mode, listOptions)
307+
}
308+
309+
// NewRemoveAccessControlRecursivePager removes the owner, owning group, and permissions for a directory (dfs1).
310+
func (d *Client) NewRemoveAccessControlRecursivePager(ACL string, options *RemoveAccessControlRecursiveOptions) *runtime.Pager[RemoveAccessControlRecursiveResponse] {
311+
mode, listOptions := options.format(ACL, "remove")
312+
return d.setAccessControlHelper(mode, listOptions)
219313
}
220314

221-
// RemoveAccessControlRecursive removes the owner, owning group, and permissions for a file or directory (dfs1).
222-
func (d *Client) RemoveAccessControlRecursive(ctx context.Context, options *RemoveAccessControlRecursiveOptions) (RemoveAccessControlRecursiveResponse, error) {
223-
// TODO explicitly pass SetAccessControlRecursiveMode
224-
return SetAccessControlRecursiveResponse{}, nil
315+
// GetAccessControl gets the owner, owning group, and permissions for a directory (dfs1).
316+
func (d *Client) GetAccessControl(ctx context.Context, options *GetAccessControlOptions) (GetAccessControlResponse, error) {
317+
opts, lac, mac := path.FormatGetAccessControlOptions(options)
318+
resp, err := d.generatedDirClientWithDFS().GetProperties(ctx, opts, lac, mac)
319+
err = exported.ConvertToDFSError(err)
320+
return resp, err
225321
}
226322

227-
// SetMetadata sets the metadata for a file or directory (blob3).
323+
// SetMetadata sets the metadata for a directory (blob3).
228324
func (d *Client) SetMetadata(ctx context.Context, options *SetMetadataOptions) (SetMetadataResponse, error) {
229-
// TODO: call directly into blob
230-
return SetMetadataResponse{}, nil
325+
opts, metadata := path.FormatSetMetadataOptions(options)
326+
resp, err := d.blobClient().SetMetadata(ctx, metadata, opts)
327+
err = exported.ConvertToDFSError(err)
328+
return resp, err
231329
}
232330

233-
// SetHTTPHeaders sets the HTTP headers for a file or directory (blob3).
331+
// SetHTTPHeaders sets the HTTP headers for a directory (blob3).
234332
func (d *Client) SetHTTPHeaders(ctx context.Context, httpHeaders HTTPHeaders, options *SetHTTPHeadersOptions) (SetHTTPHeadersResponse, error) {
235-
// TODO: call formatBlobHTTPHeaders() since we want to add the blob prefix to our options before calling into blob
236-
// TODO: call into blob
237-
return SetHTTPHeadersResponse{}, nil
333+
opts, blobHTTPHeaders := path.FormatSetHTTPHeadersOptions(options, httpHeaders)
334+
resp, err := d.blobClient().SetHTTPHeaders(ctx, blobHTTPHeaders, opts)
335+
newResp := SetHTTPHeadersResponse{}
336+
path.FormatSetHTTPHeadersResponse(&newResp, &resp)
337+
err = exported.ConvertToDFSError(err)
338+
return newResp, err
339+
}
340+
341+
// GetSASURL is a convenience method for generating a SAS token for the currently pointed at blob.
342+
// It can only be used if the credential supplied during creation was a SharedKeyCredential.
343+
func (f *Client) GetSASURL(permissions sas.DirectoryPermissions, expiry time.Time, o *GetSASURLOptions) (string, error) {
344+
if f.sharedKey() == nil {
345+
return "", datalakeerror.MissingSharedKeyCredential
346+
}
347+
348+
urlParts, err := sas.ParseURL(f.BlobURL())
349+
err = exported.ConvertToDFSError(err)
350+
if err != nil {
351+
return "", err
352+
}
353+
354+
st := path.FormatGetSASURLOptions(o)
355+
356+
qps, err := sas.DatalakeSignatureValues{
357+
DirectoryPath: urlParts.PathName,
358+
FilesystemName: urlParts.FilesystemName,
359+
Version: sas.Version,
360+
Permissions: permissions.String(),
361+
StartTime: st,
362+
ExpiryTime: expiry.UTC(),
363+
}.SignWithSharedKey(f.sharedKey())
364+
365+
err = exported.ConvertToDFSError(err)
366+
if err != nil {
367+
return "", err
368+
}
369+
370+
endpoint := f.BlobURL() + "?" + qps.Encode()
371+
372+
return endpoint, nil
238373
}

0 commit comments

Comments
 (0)