%% -*- mode: erlang;erlang-indent-level: 4;indent-tabs-mode: nil -*- -module(erlcloud_emr). -include("erlcloud_aws.hrl"). %% Library initialization -export([configure/2, configure/3, configure/4, configure/5, new/2, new/3, new/4, new/5]). %% ElasticMapReduce API -export([add_job_flow_steps/2, add_job_flow_steps/3, add_job_flow_steps/4]). -export([describe_step/2, describe_step/3, describe_step/4]). -export([run_job_flow/1, run_job_flow/2, run_job_flow/3]). -export([terminate_job_flows/1, terminate_job_flows/2, terminate_job_flows/3]). -export([set_termination_protection/2, set_termination_protection/3, set_termination_protection/4]). -export_type([emr_opts/0, emr_return/0]). %%------------------------------------------------------------------------------ %% Library initialization %%------------------------------------------------------------------------------ -spec new(string(), string()) -> aws_config(). new(AccessKeyID, SecretAccessKey) -> #aws_config{access_key_id=AccessKeyID, secret_access_key=SecretAccessKey}. -spec new(string(), string(), string()) -> aws_config(). new(AccessKeyID, SecretAccessKey, Host) -> #aws_config{access_key_id=AccessKeyID, secret_access_key=SecretAccessKey, emr_host=Host}. -spec new(string(), string(), string(), non_neg_integer()) -> aws_config(). new(AccessKeyID, SecretAccessKey, Host, Port) -> #aws_config{access_key_id=AccessKeyID, secret_access_key=SecretAccessKey, emr_host=Host, emr_port=Port}. -spec new(string(), string(), string(), non_neg_integer(), string()) -> aws_config(). new(AccessKeyID, SecretAccessKey, Host, Port, Scheme) -> #aws_config{access_key_id=AccessKeyID, secret_access_key=SecretAccessKey, emr_host=Host, emr_port=Port, emr_scheme=Scheme}. -spec configure(string(), string()) -> ok. configure(AccessKeyID, SecretAccessKey) -> erlcloud_config:configure(AccessKeyID, SecretAccessKey, fun new/2). -spec configure(string(), string(), string()) -> ok. configure(AccessKeyID, SecretAccessKey, Host) -> erlcloud_config:configure(AccessKeyID, SecretAccessKey, Host, fun new/3). -spec configure(string(), string(), string(), non_neg_integer()) -> ok. configure(AccessKeyID, SecretAccessKey, Host, Port) -> erlcloud_config:configure(AccessKeyID, SecretAccessKey, Host, Port, fun new/4). -spec configure(string(), string(), string(), non_neg_integer(), string()) -> ok. configure(AccessKeyID, SecretAccessKey, Host, Port, Scheme) -> erlcloud_config:configure(AccessKeyID, SecretAccessKey, Host, Port, Scheme, fun new/5). %%------------------------------------------------------------------------------ %% ElasticMapReduce API %%------------------------------------------------------------------------------ -type emr_opts() :: [{out, json | raw}]. -type emr_return() :: {ok, jsx:json_term() | binary()} | {error, {aws_error, jsx:json_term()}} | {error, {socket_error, any()}} | {error, tuple()}. %% --- AddJobFlowSteps --- %% http://docs.aws.amazon.com/ElasticMapReduce/latest/API/API_AddJobFlowSteps.html -spec add_job_flow_steps(binary(), jsx:json_term()) -> emr_return(). add_job_flow_steps(JobFlowId, Steps) -> add_job_flow_steps(JobFlowId, Steps, [], erlcloud_aws:default_config()). -spec add_job_flow_steps(binary(), jsx:json_term(), aws_config() | emr_opts()) -> emr_return(). add_job_flow_steps(JobFlowId, Steps, Config = #aws_config{}) -> add_job_flow_steps(JobFlowId, Steps, [], Config); add_job_flow_steps(JobFlowId, Steps, Opts) when is_list(Opts) -> add_job_flow_steps(JobFlowId, Steps, Opts, erlcloud_aws:default_config()). -spec add_job_flow_steps(binary(), jsx:json_term(), emr_opts(), aws_config()) -> emr_return(). add_job_flow_steps(JobFlowId, Steps, Opts, Config) -> emr_request("ElasticMapReduce.AddJobFlowSteps", [{'JobFlowId', JobFlowId}, {'Steps', Steps}], Opts, Config). %% --- DescribeStep --- %% http://docs.aws.amazon.com/ElasticMapReduce/latest/API/API_DescribeStep.html -spec describe_step(binary(), binary()) -> emr_return(). describe_step(ClusterId, StepId) -> describe_step(ClusterId, StepId, [], erlcloud_aws:default_config()). -spec describe_step(binary(), binary(), aws_config() | emr_opts()) -> emr_return(). describe_step(ClusterId, StepId, Config = #aws_config{}) -> describe_step(ClusterId, StepId, [], Config); describe_step(ClusterId, StepId, Opts) when is_list(Opts) -> describe_step(ClusterId, StepId, Opts, erlcloud_aws:default_config()). -spec describe_step(binary(), binary(), emr_opts(), aws_config()) -> emr_return(). describe_step(ClusterId, StepId, Opts, Config) -> emr_request("ElasticMapReduce.DescribeStep", [{'ClusterId', ClusterId}, {'StepId', StepId}], Opts, Config). %% --- RunJobFlow --- %% http://docs.aws.amazon.com/ElasticMapReduce/latest/API/API_RunJobFlow.html -spec run_job_flow(jsx:json_term()) -> emr_return(). run_job_flow(Params) -> run_job_flow(Params, [], erlcloud_aws:default_config()). -spec run_job_flow(jsx:json_term(), aws_config() | emr_opts()) -> emr_return(). run_job_flow(Params, Config = #aws_config{}) -> run_job_flow(Params, [], Config); run_job_flow(Params, Opts) when is_list(Opts) -> run_job_flow(Params, Opts, erlcloud_aws:default_config()). -spec run_job_flow(jsx:json_term(), emr_opts(), aws_config()) -> emr_return(). run_job_flow(Params, Opts, Config) -> emr_request("ElasticMapReduce.RunJobFlow", Params, Opts, Config). %% --- TerminateJobFlows --- %% http://docs.aws.amazon.com/ElasticMapReduce/latest/API/API_TerminateJobFlows.html -spec terminate_job_flows([binary()]) -> emr_return(). terminate_job_flows(JobFlowIds) -> terminate_job_flows(JobFlowIds, [], erlcloud_aws:default_config()). -spec terminate_job_flows([binary()], aws_config() | emr_opts()) -> emr_return(). terminate_job_flows(JobFlowIds, Config = #aws_config{}) -> terminate_job_flows(JobFlowIds, [], Config); terminate_job_flows(JobFlowIds, Opts) when is_list(Opts) -> terminate_job_flows(JobFlowIds, Opts, erlcloud_aws:default_config()). -spec terminate_job_flows([binary()], emr_opts(), aws_config()) -> emr_return(). terminate_job_flows(JobFlowIds, Opts, Config) -> emr_request("ElasticMapReduce.TerminateJobFlows", [{'JobFlowIds', JobFlowIds}], Opts, Config). %% --- SetTerminationProtection --- %% http://docs.aws.amazon.com/ElasticMapReduce/latest/API/API_SetTerminationProtection.html -spec set_termination_protection([binary()], boolean()) -> emr_return(). set_termination_protection(JobFlowIds, TerminationProtected) -> set_termination_protection(JobFlowIds, TerminationProtected, [], erlcloud_aws:default_config()). -spec set_termination_protection([binary()], boolean(), aws_config() | emr_opts()) -> emr_return(). set_termination_protection(JobFlowIds, TerminationProtected, Config = #aws_config{}) -> set_termination_protection(JobFlowIds, TerminationProtected, [], Config); set_termination_protection(JobFlowIds, TerminationProtected, Opts) when is_list(Opts) -> set_termination_protection(JobFlowIds, TerminationProtected, Opts, erlcloud_aws:default_config()). -spec set_termination_protection([binary()], boolean(), emr_opts(), aws_config()) -> emr_return(). set_termination_protection(JobFlowIds, TerminationProtected, Opts, Config) -> emr_request("ElasticMapReduce.SetTerminationProtection", [{'JobFlowIds', JobFlowIds}, {'TerminationProtected', TerminationProtected}], Opts, Config). %%------------------------------------------------------------------------------ %% Internal functions %%------------------------------------------------------------------------------ emr_request(Action, Json, Opts, Cfg) -> {Scheme, Host, Port, Service} = emr_endpoint(Cfg), request(Action, Json, Scheme, Host, Port, Service, Opts, Cfg). emr_endpoint(Cfg) -> %% Scheme, Host, Port, Service {Cfg#aws_config.emr_scheme, Cfg#aws_config.emr_host, Cfg#aws_config.emr_port, "elasticmapreduce"}. request(Action, Json, Scheme, Host, Port, Service, Opts, Cfg0) -> case erlcloud_aws:update_config(Cfg0) of {ok, Cfg} -> request_no_update(Action, Json, Scheme, Host, Port, Service, Opts, Cfg); Err -> Err end. request_no_update(Action, Json, Scheme, Host, Port, Service, Opts, Cfg) -> ReqBody = jsx:encode(if Json == [] -> [{}]; true -> Json end), H1 = [{"host", Host}], H2 = [{"content-type", "application/x-amz-json-1.1"}, {"x-amz-target", Action}], Region = erlcloud_aws:aws_region_from_host(Host), Headers = erlcloud_aws:sign_v4_headers(Cfg, H1, ReqBody, Region, Service) ++ H2, case erlcloud_aws:aws_request_form_raw(post, Scheme, Host, Port, "/", ReqBody, Headers, Cfg) of {ok, Body} -> case proplists:get_value(out, Opts, json) of raw -> {ok, Body}; _ -> case Body of <<>> -> {ok, <<>>}; _ -> {ok, jsx:decode(Body, [{return_maps, false}])} end end; {error, {http_error, _Code, _StatusLine, ErrBody}} -> {error, {aws_error, jsx:decode(ErrBody, [{return_maps, false}])}}; {error, {socket_error, Reason}} -> {error, {socket_error, Reason}} end.