Merge pull request dkataskin#42 from thalesmg/tmg-fixes-2-upstream

Implement `put_append_blob` and `append_block` ops, bump MS API version, handle `httpc` errors, pin `jsx` version

include/erlazure.hrl

@@ -31,15 +31,15 @@
 -define(http_created, 201).
 -define(http_accepted, 202).
 -define(http_no_content, 204).
--define(http_partial_content, 206). 
+-define(http_partial_content, 206).
 
 -define(blob_service, blob).
 -define(table_service, table).
 -define(queue_service, queue).
 -define(file_service, file).
 
 -define(queue_service_ver, "2014-02-14").
--define(blob_service_ver, "2014-02-14").
+-define(blob_service_ver, "2024-05-04").
 -define(table_service_ver, "2014-02-14").
 -define(file_service_ver, "2014-02-14").
 

rebar.config

@@ -1,5 +1,5 @@
 {deps, [
-  {jsx, ".*", {git, "https://github.com/talentdeficit/jsx.git", "main"}}
+  {jsx, {git, "https://github.com/talentdeficit/jsx.git", {tag, "v3.1.0"}}}
 ]}.
 {eunit_opts, [verbose]}.
 {cover_enabled, true}.

src/erlazure.erl

@@ -62,6 +62,8 @@
 -export([lease_container/3, lease_container/4, lease_container/5]).
 -export([list_blobs/2, list_blobs/3, list_blobs/4]).
 -export([put_block_blob/4, put_block_blob/5, put_block_blob/6]).
+-export([put_append_blob/3, put_append_blob/4, put_append_blob/5]).
+-export([append_block/4, append_block/5, append_block/6]).
 -export([put_page_blob/4, put_page_blob/5, put_page_blob/6]).
 -export([get_blob/3, get_blob/4, get_blob/5]).
 -export([snapshot_blob/3, snapshot_blob/4, snapshot_blob/5]).
@@ -242,6 +244,20 @@ put_page_blob(Pid, Container, Name, ContentLength, Options) ->
 put_page_blob(Pid, Container, Name, ContentLength, Options, Timeout) when is_list(Options); is_integer(Timeout) ->
         gen_server:call(Pid, {put_blob, Container, Name, page_blob, ContentLength, Options}, Timeout).
 
+put_append_blob(Pid, Container, Name) ->
+        put_append_blob(Pid, Container, Name, []).
+put_append_blob(Pid, Container, Name, Options) ->
+        put_append_blob(Pid, Container, Name, Options, ?gen_server_call_default_timeout).
+put_append_blob(Pid, Container, Name, Options, Timeout) when is_list(Options); is_integer(Timeout) ->
+        gen_server:call(Pid, {put_blob, Container, Name, append_blob, Options}, Timeout).
+
+append_block(Pid, Container, Name, Data) ->
+        append_block(Pid, Container, Name, Data, []).
+append_block(Pid, Container, Name, Data, Options) ->
+        append_block(Pid, Container, Name, Data, Options, ?gen_server_call_default_timeout).
+append_block(Pid, Container, Name, Data, Options, Timeout) when is_list(Options); is_integer(Timeout) ->
+        gen_server:call(Pid, {append_block, Container, Name, Data, Options}, Timeout).
+
 list_blobs(Pid, Container) ->
         list_blobs(Pid, Container, []).
 list_blobs(Pid, Container, Options) ->
@@ -481,9 +497,13 @@ handle_call({list_containers, Options}, _From, State) ->
         ReqOptions = [{params, [{comp, list}] ++ Options}],
         ReqContext = new_req_context(?blob_service, ReqOptions, State),
 
-        {?http_ok, Body} = execute_request(ServiceContext, ReqContext),
-        {ok, Containers} = erlazure_blob:parse_container_list(Body),
-        {reply, Containers, State};
+        case execute_request(ServiceContext, ReqContext) of
+            {?http_ok, Body} ->
+                {ok, Containers} = erlazure_blob:parse_container_list(Body),
+                {reply, Containers, State};
+            {error, _} = Error ->
+                {reply, Error, State}
+        end;
 
 % Create a container
 handle_call({create_container, Name, Options}, _From, State) ->
@@ -565,6 +585,35 @@ handle_call({put_blob, Container, Name, Type = page_blob, ContentLength, Options
         {Code, Body} = execute_request(ServiceContext, ReqContext),
         return_response(Code, Body, State, ?http_created, created);
 
+% Put append blob
+handle_call({put_blob, Container, Name, Type = append_blob, Options}, _From, State) ->
+        ServiceContext = new_service_context(?blob_service, State),
+        Params = [{blob_type, Type}],
+        ReqOptions = [{method, put},
+                      {path, lists:concat([Container, "/", Name])},
+                      {params, Params ++ Options}],
+        ReqContext1 = new_req_context(?blob_service, ReqOptions, State),
+        ReqContext = case proplists:get_value(content_type, Options) of
+                       undefined    -> ReqContext1#req_context{ content_type = "application/octet-stream" };
+                       ContentType  -> ReqContext1#req_context{ content_type = ContentType }
+                     end,
+
+        {Code, Body} = execute_request(ServiceContext, ReqContext),
+        return_response(Code, Body, State, ?http_created, created);
+
+% Append block
+handle_call({append_block, Container, Name, Data, Options}, _From, State) ->
+        ServiceContext = new_service_context(?blob_service, State),
+        Params = [{comp, "appendblock"}],
+        ReqOptions = [{method, put},
+                      {path, lists:concat([Container, "/", Name])},
+                      {body, Data},
+                      {params, Params ++ Options}],
+        ReqContext = new_req_context(?blob_service, ReqOptions, State),
+
+        {Code, Body} = execute_request(ServiceContext, ReqContext),
+        return_response(Code, Body, State, ?http_created, appended);
+
 % Get blob
 handle_call({get_blob, Container, Blob, Options}, _From, State) ->
         ServiceContext = new_service_context(?blob_service, State),
@@ -618,7 +667,7 @@ handle_call({delete_blob, Container, Blob, Options}, _From, State) ->
 handle_call({put_block, Container, Blob, BlockId, Content, Options}, _From, State) ->
         ServiceContext = new_service_context(?blob_service, State),
         Params = [{comp, block},
-                  {blob_block_id, base64:encode_to_string(BlockId)}],
+                  {blob_block_id, uri_string:quote(base64:encode_to_string(BlockId))}],
         ReqOptions = [{method, put},
                       {path, lists:concat([Container, "/", Blob])},
                       {body, Content},
@@ -629,12 +678,13 @@ handle_call({put_block, Container, Blob, BlockId, Content, Options}, _From, Stat
         return_response(Code, Body, State, ?http_created, created);
 
 % Put block list
-handle_call({put_block_list, Container, Blob, BlockRefs, Options}, _From, State) ->
+handle_call({put_block_list, Container, Blob, BlockRefs, Options0}, _From, State) ->
         ServiceContext = new_service_context(?blob_service, State),
+        {ExtraReqOpts, Options} = proplist_take(req_opts, Options0, []),
         ReqOptions = [{method, put},
                       {path, lists:concat([Container, "/", Blob])},
                       {body, erlazure_blob:get_request_body(BlockRefs)},
-                      {params, [{comp, "blocklist"}] ++ Options}],
+                      {params, [{comp, "blocklist"}] ++ Options} | ExtraReqOpts],
         ReqContext = new_req_context(?blob_service, ReqOptions, State),
 
         {Code, Body} = execute_request(ServiceContext, ReqContext),
@@ -764,7 +814,10 @@ execute_request(ServiceContext = #service_context{}, ReqContext = #req_context{}
             {Code, Body};
 
           {ok, {{_, _, _}, _, Body}} ->
-            get_error_code(Body)
+            get_error_code(Body);
+
+          {error, Error} ->
+            {error, Error}
         end.
 
 get_error_code(Body) ->
@@ -886,6 +939,11 @@ combine_canonical_param({Param, Value}, _PreviousParam, Acc, ParamList) ->
         [H | T] = ParamList,
         combine_canonical_param(H, Param, add_param_value(Param, Value, Acc), T).
 
+add_param_value(Param = "blockid", Value, Acc) ->
+        %% special case: `blockid' must be URL-encoded when sending the request, but not
+        %% when signing it.  At this point, we've already encoded it.
+        Acc ++ "\n" ++ string:to_lower(Param) ++ ":" ++ uri_string:unquote(Value);
+
 add_param_value(Param, Value, Acc) ->
         Acc ++ "\n" ++ string:to_lower(Param) ++ ":" ++ Value.
 
@@ -1028,6 +1086,14 @@ ensure_wrapped_key(#{key := Key} = InitOpts) ->
             InitOpts#{key := wrap(Key)}
     end.
 
+proplist_take(Key, Proplist, Default) ->
+    case lists:keytake(Key, 1, Proplist) of
+        false ->
+            {Default, Proplist};
+        {value, {Key, Value}, NewProplist} ->
+            {Value, NewProplist}
+    end.
+
 %%====================================================================
 %% Tests
 %%====================================================================

src/erlazure_blob.erl

@@ -166,9 +166,11 @@ parse_block(#xmlElement{ content = Content }, Type) ->
         lists:foldl(FoldFun, #blob_block{ type = Type }, Nodes).
 
 str_to_blob_type("BlockBlob") -> block_blob;
+str_to_blob_type("AppendBlob") -> append_blob;
 str_to_blob_type("PageBlob") -> page_blob.
 
 blob_type_to_str(block_blob) -> "BlockBlob";
+blob_type_to_str(append_blob) -> "AppendBlob";
 blob_type_to_str(page_blob) -> "PageBlob".
 
 block_type_to_node(uncommitted) -> 'Uncommitted';
@@ -186,7 +188,11 @@ get_request_body(BlockRefs) ->
         FoldFun = fun(BlockRef=#blob_block{}, Acc) ->
                       [{block_type_to_node(BlockRef#blob_block.type),
                         [],
-                        [base64:encode_to_string(BlockRef#blob_block.id)]} | Acc]
+                        [base64:encode_to_string(BlockRef#blob_block.id)]} | Acc];
+                     ({BlockId, BlockType}, Acc) ->
+                      [{block_type_to_node(BlockType),
+                        [],
+                        [base64:encode_to_string(BlockId)]} | Acc]
                   end,
         Data = {'BlockList', [], lists:reverse(lists:foldl(FoldFun, [], BlockRefs))},
         lists:flatten(xmerl:export_simple([Data], xmerl_xml)).

test/erlazure_SUITE.erl

@@ -125,3 +125,91 @@ t_blob_storage_wrapped_key(Config) ->
     {ok, Pid} = erlazure:start(#{account => ?ACCOUNT, key => ?KEY, endpoint => Endpoint}),
     ?assertMatch({[], _}, erlazure:list_containers(Pid)),
     ok.
+
+%% Basic smoke test for append blob storage operations.
+t_append_blob_smoke_test(Config) ->
+    Endpoint = ?config(endpoint, Config),
+    {ok, Pid} = erlazure:start(#{account => ?ACCOUNT, key => ?KEY, endpoint => Endpoint}),
+    %% Create a container
+    Container = container_name(?FUNCTION_NAME),
+    ?assertMatch({[], _}, erlazure:list_containers(Pid)),
+    ?assertMatch({ok, created}, erlazure:create_container(Pid, Container)),
+    %% Upload some blobs
+    Opts = [{content_type, "text/csv"}],
+    ?assertMatch({ok, created}, erlazure:put_append_blob(Pid, Container, "blob1", Opts)),
+    ?assertMatch({ok, appended}, erlazure:append_block(Pid, Container, "blob1", <<"1">>)),
+    ?assertMatch({ok, appended}, erlazure:append_block(Pid, Container, "blob1", <<"\n">>)),
+    ?assertMatch({ok, appended}, erlazure:append_block(Pid, Container, "blob1", <<"2">>)),
+    ListedBlobs = erlazure:list_blobs(Pid, Container),
+    ?assertMatch({[#cloud_blob{name = "blob1"}], _},
+                 ListedBlobs),
+    {[#cloud_blob{name = "blob1", properties = BlobProps}], _} = ListedBlobs,
+    ?assertMatch(#{content_type := "text/csv"}, maps:from_list(BlobProps)),
+    %% Read back data
+    ?assertMatch({ok, <<"1\n2">>}, erlazure:get_blob(Pid, Container, "blob1")),
+    %% Delete blob
+    ?assertMatch({ok, deleted}, erlazure:delete_blob(Pid, Container, "blob1")),
+    ?assertMatch({[], _}, erlazure:list_blobs(Pid, Container)),
+    %% Delete container
+    ?assertMatch({ok, deleted}, erlazure:delete_container(Pid, Container)),
+    ok.
+
+%% Test error handling when endpoint is unavailable
+t_blob_failure_to_connect(_Config) ->
+    BadEndpoint = "http://127.0.0.2:65535/",
+    {ok, Pid} = erlazure:start(#{account => ?ACCOUNT, key => ?KEY, endpoint => BadEndpoint}),
+    ?assertMatch({error, {failed_connect, _}}, erlazure:list_containers(Pid)),
+    ?assertMatch({error, {failed_connect, _}}, erlazure:create_container(Pid, "c")),
+    ?assertMatch({error, {failed_connect, _}}, erlazure:delete_container(Pid, "c")),
+    ?assertMatch({error, {failed_connect, _}}, erlazure:put_append_blob(Pid, "c", "b1")),
+    ?assertMatch({error, {failed_connect, _}}, erlazure:put_block_blob(Pid, "c", "b1", <<"a">>)),
+    ?assertMatch({error, {failed_connect, _}}, erlazure:append_block(Pid, "c", "b1", <<"a">>)),
+    ?assertMatch({error, {failed_connect, _}}, erlazure:get_blob(Pid, "c", "b1")),
+    ok.
+
+%% Basic smoke test for block blob storage operations.
+t_put_block(Config) ->
+    Endpoint = ?config(endpoint, Config),
+    {ok, Pid} = erlazure:start(#{account => ?ACCOUNT, key => ?KEY, endpoint => Endpoint}),
+    %% Create a container
+    Container = container_name(?FUNCTION_NAME),
+    ?assertMatch({[], _}, erlazure:list_containers(Pid)),
+    ?assertMatch({ok, created}, erlazure:create_container(Pid, Container)),
+    %% Upload some blocks.  Note: this content-type will be overwritten later by `put_block_list'.
+    Opts1 = [{content_type, "application/json"}],
+    BlobName = "blob1",
+    ?assertMatch({ok, created}, erlazure:put_block_blob(Pid, Container, BlobName, <<"0">>, Opts1)),
+    %% Note: this short name is important for this test.  It'll produce a base64 string
+    %% that's padded.  That padding must be URL-encoded when sending the request, but not
+    %% when generating the string to sign.
+    BlockId1 = <<"blo1">>,
+    ?assertMatch({ok, created}, erlazure:put_block(Pid, Container, BlobName, BlockId1, <<"a">>)),
+    %% Testing iolists
+    BlockId2 = <<"blo2">>,
+    ?assertMatch({ok, created}, erlazure:put_block(Pid, Container, BlobName, BlockId2, [<<"\n">>, ["b", [$\n]]])),
+    %% Not yet committed.  Contains only the data from the blob creation.
+    ?assertMatch({ok, <<"0">>}, erlazure:get_blob(Pid, Container, BlobName)),
+    %% Committing
+    BlockList1 = [{BlockId1, latest}],
+    ?assertMatch({ok, created}, erlazure:put_block_list(Pid, Container, BlobName, BlockList1)),
+    %% Committed only first block.  Initial data was lost, as it was not in the block list.
+    ?assertMatch({ok, <<"a">>}, erlazure:get_blob(Pid, Container, BlobName)),
+    %% Block 2 was dropped after committing.
+    ?assertMatch({[#blob_block{id = "blo1"}], _}, erlazure:get_block_list(Pid, Container, BlobName)),
+    BlockId3 = <<"blo3">>,
+    ?assertMatch({ok, created}, erlazure:put_block(Pid, Container, BlobName, BlockId3, [<<"\n">>, ["b", [$\n]]])),
+    %% Commit both blocks
+    Opts2 = [{req_opts, [{headers, [{"x-ms-blob-content-type", "text/csv"}]}]}],
+    BlockList2 = [{BlockId1, committed}, {BlockId3, uncommitted}],
+    ?assertMatch({ok, created}, erlazure:put_block_list(Pid, Container, BlobName, BlockList2, Opts2)),
+    ?assertMatch({ok, <<"a\nb\n">>}, erlazure:get_blob(Pid, Container, BlobName)),
+    %% Check content type.
+    ListedBlobs = erlazure:list_blobs(Pid, Container),
+    ?assertMatch({[#cloud_blob{name = "blob1"}], _},
+                 ListedBlobs),
+    {[#cloud_blob{name = "blob1", properties = Props}], _} = ListedBlobs,
+    %% Content-type from `put_block_list' prevails.
+    ?assertMatch(#{content_type := "text/csv"}, maps:from_list(Props)),
+    %% Delete container
+    ?assertMatch({ok, deleted}, erlazure:delete_container(Pid, Container)),
+    ok.