Skip to content

Commit a2a7e99

Browse files
samtkaplansamtkaplan
committed
use copy api
1 parent 5ab726f commit a2a7e99

File tree

4 files changed

+201
-18
lines changed

4 files changed

+201
-18
lines changed

‎.github/workflows/ci.yml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ jobs:
3737
run: |
3838
az group create -l southcentralus -n "AzureBackupRG-azstorage-${{ matrix.os }}-${{ matrix.version }}-${{ github.run_id }}"
3939
az storage account create --min-tls-version TLS1_2 -n "s${{ steps.uuid.outputs.uuid }}" -g "AzureBackupRG-azstorage-${{ matrix.os }}-${{ matrix.version }}-${{ github.run_id }}" -l southcentralus
40+
az storage account create --min-tls-version TLS1_2 -n "s${{ steps.uuid.outputs.uuid }}2" -g "AzureBackupRG-azstorage-${{ matrix.os }}-${{ matrix.version }}-${{ github.run_id }}" -l southcentralus
4041
- uses: julia-actions/cache@v1
4142
- uses: julia-actions/julia-buildpkg@v1
4243
- uses: julia-actions/julia-runtest@v1
@@ -45,6 +46,7 @@ jobs:
4546
CLIENT_SECRET: ${{ secrets.CLIENT_SECRET }}
4647
TENANT: ${{ secrets.TENANT_ID }}
4748
STORAGE_ACCOUNT: "s${{ steps.uuid.outputs.uuid }}"
49+
STORAGE_ACCOUNT_TOO: "s${{ steps.uuid.outputs.uuid }}-2"
4850
- uses: julia-actions/julia-processcoverage@v1
4951
- uses: codecov/codecov-action@v5
5052
with:

‎Project.toml‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
name = "AzStorage"
22
uuid = "c6697862-1611-5eae-9ef8-48803c85c8d6"
3-
version = "2.7.2"
3+
version = "2.7.3"
44

55
[deps]
66
AbstractStorage = "14dbef02-f468-5f15-853e-5ec8dee7b899"
@@ -12,6 +12,7 @@ DelimitedFiles = "8bb1440f-4735-579b-a4ab-409b98df4dab"
1212
HTTP = "cd3eb016-35fb-5094-929b-558a96fad6f3"
1313
Printf = "de0858da-6303-5e67-8744-51eddeeeb8d7"
1414
ProgressMeter = "92933f4c-e287-5a05-a399-4b506db050ca"
15+
SHA = "ea8e919c-243c-51af-8825-aaa63cd721ce"
1516
Serialization = "9e88b42a-f829-5b0c-bbe9-9e923198166b"
1617
Sockets = "6462fe0b-24de-5631-8697-dd941f90decc"
1718
XML = "72c71f33-b9b6-44de-8c94-c961784809e2"
@@ -20,8 +21,10 @@ XML = "72c71f33-b9b6-44de-8c94-c961784809e2"
2021
AbstractStorage = "^1.3"
2122
AzSessions = "2"
2223
AzStorage_jll = "0.9"
24+
Base64 = "1"
2325
DelimitedFiles = "1"
2426
HTTP = "1"
2527
ProgressMeter = "1"
28+
SHA = "0.7"
2629
XML = "0.3"
2730
julia = "^1.6"

‎src/AzStorage.jl‎

Lines changed: 179 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
module AzStorage
22

3-
using AbstractStorage, AzSessions, AzStorage_jll, Base64, Dates, DelimitedFiles, XML, HTTP, Printf, ProgressMeter, Serialization, Sockets
3+
using AbstractStorage, AzSessions, AzStorage_jll, Base64, Dates, DelimitedFiles, XML, HTTP, Printf, ProgressMeter, SHA, Serialization, Sockets
44

55
# https://docs.microsoft.com/en-us/rest/api/storageservices/common-rest-api-error-codes
66
# https://learn.microsoft.com/en-us/azure/azure-resource-manager/management/request-limits-and-throttling
@@ -854,9 +854,182 @@ function Base.cp(inc::AzContainer, inb::AbstractString, out::AbstractString; buf
854854
close(io)
855855
end
856856

857-
function Base.cp(inc::AzContainer, inb::AbstractString, outc::AzContainer, outb::AbstractString)
858-
bytes = read!(inc, inb, Vector{UInt8}(undef, filesize(inc, inb)))
859-
write(outc, outb, bytes)
857+
#=
858+
If the source and destination storage accounts for a blob copy are different, then the Azure storage API does not allow us
859+
to use OAuth2/RBAC directly for the source blob. But, we can use a user delegation SAS token which is build in the following
860+
two methods (generate_user_delegation_key and get_user_delegation_sas).
861+
=#
862+
function get_user_delegation_key(c::AzContainer; start=now(UTC), expiry=now(UTC)+Hour(6))
863+
start_str = Dates.format(start, "yyyy-mm-ddTHH:MM:SSZ")
864+
expiry_str = Dates.format(expiry, "yyyy-mm-ddTHH:MM:SSZ")
865+
866+
r = @retry c.nretry HTTP.request(
867+
"POST",
868+
"https://$(c.storageaccount).blob.core.windows.net/?restype=service&comp=userdelegationkey",
869+
[
870+
"Authorization" => "Bearer $(token(c.session))",
871+
"x-ms-version" => API_VERSION,
872+
"Content-Type" => "application/xml"
873+
],
874+
"""
875+
<?xml version="1.0" encoding="utf-8"?>
876+
<KeyInfo>
877+
<Start>$start_str</Start>
878+
<Expiry>$expiry_str</Expiry>
879+
</KeyInfo>
880+
""",
881+
retry = false,
882+
verbose = c.verbose)
883+
884+
b = XML.parse(String(r.body), LazyNode)
885+
delegation_key = Dict{String,String}()
886+
for child in children(b)
887+
if tag(child) == "UserDelegationKey"
888+
for grandchild in children(child)
889+
if tag(grandchild) in ("SignedOid", "SignedTid", "SignedStart", "SignedExpiry", "SignedService", "SignedVersion", "Value")
890+
delegation_key[string(tag(grandchild))] = value(first(children(grandchild)))
891+
end
892+
end
893+
end
894+
end
895+
896+
delegation_key
897+
end
898+
899+
function generate_user_delegation_sas(c::AzContainer, b::AbstractString; permissions="r", start=now(UTC), expiry=now(UTC)+Hour(6))
900+
delegation_key = get_user_delegation_key(c; start, expiry)
901+
902+
signedPermissions = permissions
903+
signedStart = Dates.format(start, "yyyy-mm-ddTHH:MM:SSZ")
904+
signedExpiry = Dates.format(expiry, "yyyy-mm-ddTHH:MM:SSZ")
905+
canonicalizedResource = "/blob/$(c.storageaccount)/$(c.containername)/$(addprefix(c,b))"
906+
signedKeyObjectId = delegation_key["SignedOid"]
907+
signedKeyTenantId = delegation_key["SignedTid"]
908+
signedKeyStart = delegation_key["SignedStart"]
909+
signedKeyExpiry = delegation_key["SignedExpiry"]
910+
signedKeyService = delegation_key["SignedService"]
911+
signedKeyVersion = delegation_key["SignedVersion"]
912+
signedAuthorizedUserObjectId = ""
913+
signedUnauthorizedUserObjectId = ""
914+
signedCorrelationId = ""
915+
signedIP = ""
916+
signedProtocol = "https"
917+
signedVersion = API_VERSION
918+
signedResource = "b"
919+
signedSnapshotTime = ""
920+
signedEncryptionScope = ""
921+
rscc = ""
922+
rscd = ""
923+
rsce = ""
924+
rscl = ""
925+
rsct = ""
926+
927+
string_to_sign =
928+
signedPermissions * "\n" *
929+
signedStart * "\n" *
930+
signedExpiry * "\n" *
931+
canonicalizedResource * "\n" *
932+
signedKeyObjectId * "\n" *
933+
signedKeyTenantId * "\n" *
934+
signedKeyStart * "\n" *
935+
signedKeyExpiry * "\n" *
936+
signedKeyService * "\n" *
937+
signedKeyVersion * "\n" *
938+
signedAuthorizedUserObjectId * "\n" *
939+
signedUnauthorizedUserObjectId * "\n" *
940+
signedCorrelationId * "\n" *
941+
"\n" *
942+
"\n" *
943+
signedIP * "\n" *
944+
signedProtocol * "\n" *
945+
signedVersion * "\n" *
946+
signedResource * "\n" *
947+
signedSnapshotTime * "\n" *
948+
signedEncryptionScope * "\n" *
949+
rscc * "\n" *
950+
rscd * "\n" *
951+
rsce * "\n" *
952+
rscl * "\n" *
953+
rsct
954+
955+
# sign the string using the delegation key
956+
key = base64decode(delegation_key["Value"])
957+
message = collect(codeunits(string_to_sign))
958+
signed_string = HTTP.escapeuri(base64encode(hmac_sha256(key, message)))
959+
960+
# sas token
961+
"sp=$signedPermissions&" *
962+
"st=$signedStart&" *
963+
"se=$signedExpiry&" *
964+
"skoid=$signedKeyObjectId&" *
965+
"sktid=$signedKeyTenantId&" *
966+
"skt=$signedKeyStart&" *
967+
"ske=$signedKeyExpiry&" *
968+
"sks=$signedKeyService&" *
969+
"skv=$signedKeyVersion&" *
970+
(isempty(signedIP) ? "" : "sip=$signedIP&") *
971+
"spr=$signedProtocol&" *
972+
"sv=$signedVersion&" *
973+
"sr=$signedResource&" *
974+
"sig=$signed_string"
975+
end
976+
977+
function Base.cp(inc::AzContainer, inb::AbstractString, outc::AzContainer, outb::AbstractString; showprogress=true)
978+
source_url = "https://$(inc.storageaccount).blob.core.windows.net/$(inc.containername)/$(addprefix(inc,inb))"
979+
980+
if inc.storageaccount != outc.storageaccount
981+
sas = generate_user_delegation_sas(inc, inb; permissions="r", start=now(UTC), expiry=now(UTC)+Hour(1))
982+
source_url *= "?$sas"
983+
end
984+
985+
headers = [
986+
"Authorization" => "Bearer $(token(outc.session))",
987+
"x-ms-version" => API_VERSION,
988+
"x-ms-copy-source" => source_url
989+
]
990+
991+
r = @retry inc.nretry HTTP.request(
992+
"PUT",
993+
"https://$(outc.storageaccount).blob.core.windows.net/$(outc.containername)/$(addprefix(outc,outb))",
994+
headers;
995+
retry = false,
996+
verbose = inc.verbose,
997+
connect_timeout = inc.connect_timeout,
998+
readtimeout = inc.read_timeout
999+
)
1000+
1001+
if r.status == 202
1002+
local copy_progress
1003+
while true
1004+
r_status = HTTP.request(
1005+
"GET",
1006+
"https://$(outc.storageaccount).blob.core.windows.net/$(outc.containername)/$(addprefix(outc,outb))",
1007+
[
1008+
"Authorization" => "Bearer $(token(outc.session))",
1009+
"x-ms-version" => API_VERSION
1010+
];
1011+
retry = false,
1012+
verbose = inc.verbose,
1013+
connect_timeout = inc.connect_timeout,
1014+
readtimeout = inc.read_timeout
1015+
)
1016+
1017+
copy_status = HTTP.header(r_status, "x-ms-copy-status")
1018+
copy_progress = HTTP.header(r_status, "x-ms-copy-progress")
1019+
if copy_status == "success"
1020+
break
1021+
end
1022+
if copy_status == "aborted"
1023+
reason = HTTP.header(r_status, "x-ms-copy-status-description")
1024+
error("blob copy aborted, src=$(inc.storageaccount): $(inc.containername)/$(addprefix(inc,inb)), dest=$(outc.storageaccount): $(outc.containername)/$(addprefix(outc,outb)), reason=$reason")
1025+
break
1026+
end
1027+
1028+
showprogress && print("copy progress: $copy_progress bytes\r")
1029+
sleep(.1)
1030+
end
1031+
showprogress && print("copy progress: $copy_progress bytes\r\n")
1032+
end
8601033
end
8611034

8621035
"""
@@ -1193,19 +1366,8 @@ function Base.cp(src::AzContainer, dst::AzContainer)
11931366
mkpath(dst)
11941367

11951368
blobs = readdir(src)
1196-
for blob in blobs
1197-
@retry dst.nretry HTTP.request(
1198-
"PUT",
1199-
"https://$(dst.storageaccount).blob.core.windows.net/$(dst.containername)/$(addprefix(dst,blob))",
1200-
[
1201-
"Authorization" => "Bearer $(token(dst.session))",
1202-
"x-ms-version" => API_VERSION,
1203-
"x-ms-copy-source" => "https://$(src.storageaccount).blob.core.windows.net/$(src.containername)/$(addprefix(src,blob))"
1204-
],
1205-
retry = false,
1206-
verbose = src.verbose,
1207-
connect_timeout = src.connect_timeout,
1208-
readtimeout = src.read_timeout)
1369+
@sync for blob in blobs
1370+
@async cp(src, blob, dst, blob; showprogress=false)
12091371
end
12101372
nothing
12111373
end

‎test/runtests.jl‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,9 @@ end
4848
storageaccount = ENV["STORAGE_ACCOUNT"]
4949
@info "storageaccount=$storageaccount"
5050

51+
storageaccount_too = get(ENV, "STORAGE_ACCOUNT_TOO", "")
52+
@info "storageaccount_too=$storageaccount_too"
53+
5154
for container in containers(;storageaccount=storageaccount,session=session)
5255
rm(AzContainer(container;storageaccount=storageaccount,session=session))
5356
end
@@ -548,6 +551,19 @@ end
548551
rm(c)
549552
end
550553

554+
@testset "Container, copy blob to blob, different storage accounts" begin
555+
r = uuid4()
556+
c1 = AzContainer("foo-$r-o", storageaccount=storageaccount, session=session, nthreads=2, nretry=10)
557+
c1 = robust_mkpath(c1)
558+
write(c1, "foo.txt", "Hello world")
559+
c2 = AzContainer("foo-$r-o", storageaccount=storageaccount_too, session=session, nthreads=2, nretry=10)
560+
c2 = robust_mkpath(c2)
561+
cp(c1, "foo.txt", c2, "bar.txt")
562+
@test read(c2, "bar.txt", String) == "Hello world"
563+
rm(c1)
564+
rm(c2)
565+
end
566+
551567
@testset "Object, copy blob to local file" begin
552568
r = uuid4()
553569
c = AzContainer("foo-$r-o", storageaccount=storageaccount, session=session, nthreads=2, nretry=10)

0 commit comments

Comments
 (0)