.rebar
logs
_build
+*.coverdata
The format is based on [Keep a Changelog](http://keepachangelog.com/)
and this project adheres to [Semantic Versioning](http://semver.org/).
+## [4.0.6] - Sep 11 2017
+* Get meck working for better unit tests
+* More unit tests
+* Removal of some "integration tests" that are now covered by unit tests
+* Small cleanups.
+
+## [4.0.5] - Missing =(
+
## [4.0.4] - Aug 29 2017
* Cleanup cdap undeploy code and fix misleading log statements
* Attempt to add ONAP build artifacts
-FROM erlang:19.2
+FROM erlang:19.3
MAINTAINER tommy at research dot eh tee tee dot com
ENV TERM=xterm
For now, this is an open question. It can always be done on the CDAP management interface by an operations team in the meantime.
# On Version Bumping (Development)
-Currently the CDAP Broker Version is in four places in this repo..
+Currently the CDAP Broker Version is in three places in this repo..
1. rebar.config
2. src/cdapbroker.app.src
-3. get_version.sh (could be made smarter, parse rebar.config)
-4. swagger spec
-If you make a developmemt change, please bump in all places until this is resolved..
+3. swagger spec
+If you make a developmemt change, please bump in all places
+++ /dev/null
-#!/bin/sh
-#todo... build this so that it parses automatically
-echo "4.0.5
;;
test)
echo "==> test phase script"
- /usr/bin/rebar3 eunit
+ /usr/bin/rebar3 eunit --cover;
+ /usr/bin/rebar3 cover;
;;
package)
echo "==> package phase script"
{relx, [
{release,
- {cdapbroker,"4.0.5"},
+ {cdapbroker,"4.0.6"},
[cdapbroker]
},
%{extend_start_script,true},
%generate RFC compliant UUIDs
{uuid, ".*", {git, "https://github.com/avtobiff/erlang-uuid.git", {branch, "master"}}},
%generate ISO8601 timestamps
- {iso8601, {git, "https://github.com/erlsci/iso8601.git", {tag, "1.2.3"}}}
+ {iso8601, {git, "https://github.com/erlsci/iso8601.git", {tag, "1.2.3"}}},
+ %mocking lib
+ {meck, {git, "https://github.com/eproxus/meck", {tag, "0.8.6"}}}
]}.
%%% Require OTP 19.2 at a bare minimum
{application, cdapbroker,
[{description, "Interface between Consul and CDAP in DCAE"},
- {vsn, "4.0.5"},
+ {vsn, "4.0.6"},
{registered, []},
{mod, { cdapbroker_app, []}},
{applications,
% Licensed under the Apache License, Version 2.0 (the "License");
% you may not use this file except in compliance with the License.
% You may obtain a copy of the License at
-%
+%
% http://www.apache.org/licenses/LICENSE-2.0
-%
+%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS,
% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-module(httpabs).
-export([get/2,
post/4, %I miss python's default arguments..
- post/5,
+ post/5,
put/4,
delete/2
]).
-define(SC(L), util:concat(L)).
%NOTE
-%Consider the Erlang statement:
+%Consider the Erlang statement:
%
%{ok, {{"HTTP/1.1",ReturnCode, State}, Head, Body}} = httpc:get(URL).
-%CDAP returns error messages in the “Body” field above.
+%CDAP returns error messages in the “Body” field above.
%
%However, Consul:
%1) Always (in all HTTP failures I’ve tested) returns Body == “500\n”
%%%
-spec parse_response({error|ok, any()}, string()) -> httpstat().
parse_response({Status, Response}, URL) ->
- case Status of
- error ->
+ case Status of
+ error ->
lager:error("httpc error: cannot hit: ~s", [URL]),
case Response of
no_scheme -> {400, io_lib:format("ERROR: The following URL is malformed: ~s", [URL])};
{bad_body, _} -> {400, "ERROR: The request Body is malformed"};
{bad_body_generator,_} -> {400, "ERROR: The request Body is malformed"};
_ ->
- lager:error(io_lib:format("Unexpected ERROR hitting ~s", [URL])),
+ lager:error(io_lib:format("Unexpected ERROR hitting ~s", [URL])),
{504, list_to_binary(io_lib:format("ERROR: The following URL is unreachable or the request was unable to be parsed due to an unknown error: ~s", [URL]))} %Are there other reasons other than bad body and unreachable that crash request? (Sneak peak: the answer is probably)
end;
- ok ->
+ ok ->
{{_, ReturnCode, State}, _Head, Body} = Response,
- case ReturnCode of
- 200 ->
+ case ReturnCode of
+ 200 ->
{ReturnCode, Body};
_ ->
lager:error("Error While hitting ~s, Non-200 status code returned. HTTP Code ~p, State ~s, ResponseBody ~s:", [URL, ReturnCode, State, Body]),
sanitize(URL) ->
%allow URL to look like "www.foo.com" or <<"www.foo.com">>, trim it
- case is_binary(URL) of
+ case is_binary(URL) of
true -> string:strip(binary_to_list(URL));
false -> string:strip(URL)
end.
%anywhere you see any() is essentially lazy typing.. fix these someday when time is abundant
-spec post(string(), string()|binary(), string(), any()) -> httpstat().
-post(XER, URL, ContentType, Body) ->
+post(XER, URL, ContentType, Body) ->
%post that sends the XER, no headers signature
Headers = [{"x-ecomp-requestid", XER}],
U = sanitize(URL),
parse_response(httpc:request(post, {U, [{"x-ecomp-requestid", XER} | Headers], ContentType, Body}, [],[]), U).
-spec get(string(), string()|binary()) -> httpstat().
-get(XER, URL) ->
+get(XER, URL) ->
%http get that always sends the XER.. even if the server doesn't want it; maybe this will blow up on me one day.
U = sanitize(URL),
Headers = [{"x-ecomp-requestid", XER}],
% Licensed under the Apache License, Version 2.0 (the "License");
% you may not use this file except in compliance with the License.
% You may obtain a copy of the License at
-%
+%
% http://www.apache.org/licenses/LICENSE-2.0
-%
+%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS,
% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-module(httpabs_tests).
-include_lib("eunit/include/eunit.hrl").
-import(httpabs, [
- sanitize/1
+ sanitize/1,
+ parse_response/2
]
).
?assert(sanitize(" www.foo.com ") == "www.foo.com"),
?assert(sanitize(<<"www.foo.com">>) == "www.foo.com").
+parse_response_test() ->
+ NoURL = "THIS IS NOT EVEN A URL WHAT ARE YOU DOING TO ME",
+ ?assert(parse_response({error,no_scheme},"THIS IS NOT EVEN A URL WHAT ARE YOU DOING TO ME") == {400, io_lib:format("ERROR: The following URL is malformed: ~s", [NoURL])}),
+ ?assert(httpabs:put("testxer", NoURL, "application/json", jiffy:encode({[{<<"a">>, <<"b">>}]})) == {400, io_lib:format("ERROR: The following URL is malformed: ~s", [NoURL])}),
+
+ %test httpabs bad body (not encoded as JSON)
+ ReconfigMap = util:ejson_to_map({[{<<"foo">>, <<"bar">>}]}),
+ BadBody = {[{<<"config">>, ReconfigMap}]},
+ ?assert(httpabs:put("testxer", "www.foo.com", "application/json", BadBody) == {400,"ERROR: The request Body is malformed"}),
+ ?assert(parse_response({error,{bad_body_generator, BadBody}}, "www.foo.com") == {400,"ERROR: The request Body is malformed"}),
+ ?assert(parse_response({ok,{{"HTTP/1.1",200,"OK"},[{"cache-control","private, max-age=0"},{"date","Mon, 11 Sep 2017 15:05:11 GMT"},{"accept-ranges","none"},{"server","gws"},{"vary","Accept-Encoding"},{"content-length","46376"},{"content-type","text/html; charset=ISO-8859-1"},{"expires","-1"},{"p3p","..."},{"x-xss-protection","1; mode=block"},{"x-frame-options","SAMEORIGIN"},{"set-cookie","NID=111=nGQHl8ljJ3nXHmGmIZmGaTgoq3WAdbgWaxAUOQJm-0AaOkS64iiXtm-HojIFSpqowj7Nr-KpqS8o-oDOROq-4AaDs0J4M92V7yBOAJQYPkuK7wtVav0BhpOOYgCHysUN; expires=Tue, 13-Mar-2018 15:05:11 GMT; path=/; domain=.google.com; HttpOnly"},{"alt-svc","quic=\":443\"; ma=2592000; v=\"39,38,37,35\""}],"<!doctype html>..."}}, "https://google.com") == {200, "<!doctype html>..."}).
-export([get/3, put/3, delete/3, post/3]).
-export([cross_domains/3]).
+-export([appname_to_field_vals/2]).
+
%%for keeping state
%%The application record is defined in application.hrl
%%In Mnesia, the first element is the type of record and the second element is the key
end
end.
-%%%CALLBACKS %%%
-init(_Route, _Req, State) ->
- {ok, State}.
-terminate(_Reason, _Route, _Req, _State) ->
- ok.
-%%%FOR Cors support
-%%%Note! only matches on host. Does not handle ports. See: https://github.com/s1n4/leptus/issues/55
-cross_domains(_Route, _Req, State) ->
- {['_'], State}.
-
-%%%GET Methods
-get("/", Req, State) ->
- %The broker's "info" endpoint; returns some possibly useful information
- {Bts, XER} = init_api_call(Req),
- Apps = util:get_all_appnames_from_db(),
- {UT, _} = statistics(wall_clock),
- CDAPVer = cdap_interface:get_cdap_cluster_version(XER, ?CDAPURL),
- RB = {[
- {<<"cdap cluster version">>, CDAPVer},
- {<<"managed cdap url">>, ?CDAPURL},
- {<<"cdap GUI port">>, cdap_interface:get_cdap_gui_port_from_version(CDAPVer)},
- {<<"number of applications registered">>, length(Apps)},
- {<<"uptime (s)">>, UT/1000},
- {<<"broker API version">>, util:get_my_version()}
- ]},
- {RCode, RBody, RState} = {200, {json, RB}, State},
- ?AUDI(Req, Bts, XER, Rcode),
- {RCode, RBody, RState};
-get("/application", Req, State) ->
- %get a list of all registered apps
- {Bts, XER} = init_api_call(Req),
- {RCode, RBody, RState} = {200, {json, util:get_all_appnames_from_db()}, State},
- ?AUDI(Req, Bts, XER, Rcode),
- {RCode, RBody, RState};
-get("/application/:appname", Req, State) ->
- %get information about a registered application
- {Bts, XER} = init_api_call(Req),
- Appname = leptus_req:param(Req, appname),
- {RCode, RBody, RState} = appname_to_application_http(XER, Appname, State),
- ?AUDI(Req, Bts, XER, Rcode),
- {RCode, RBody, RState};
-get("/application/:appname/metrics", Req, State) ->
- %get metrics for a registered application
- {Bts, XER} = init_api_call(Req),
- Appname = leptus_req:param(Req, appname),
- {RCode, RBody, RState} = case appname_to_field_vals(Appname, [<<"apptype">>, <<"namespace">>]) of
- none -> {404, "", State};
- [<<"program-flowlet">>, Namespace] ->
- {ReturnCode, ReturnBody} = cdap_interface:get_app_metrics(XER, Appname, Namespace, ?CDAPURL), %warning, see note in README, this always reutrns 200
- {ReturnCode, {json, ReturnBody}, State};
- [<<"hydrator-pipeline">>, Namespace] ->
- lager:warning("WARNING, metrics not actually implemented yet for pipelines!!"),
- {ReturnCode, ReturnBody} = cdap_interface:get_pipeline_metrics(Appname, Namespace, ?CDAPURL),
- {ReturnCode, {json, ReturnBody}, State}
- end,
- ?AUDI(Req, Bts, XER, Rcode),
- {RCode, RBody, RState};
-get("/application/:appname/healthcheck", Req, State) ->
- %get healthcheck of an application
- {Bts, XER} = init_api_call(Req),
- Appname = leptus_req:param(Req, appname),
- lager:info(io_lib:format("Get Healthcheck recieved for ~s", [Appname])),
- {RCode, RBody, RState} = case appname_to_field_vals(Appname, [<<"apptype">>, <<"namespace">>]) of
- none -> {404, "", State};
- [<<"program-flowlet">>, Namespace] ->
- {cdap_interface:get_app_healthcheck(XER, Appname, Namespace, ?CDAPURL), "", State};
- [<<"hydrator-pipeline">>, Namespace] ->
- {cdap_interface:get_pipeline_healthcheck(XER, Appname, Namespace, ?CDAPURL, ?PipelineHealthLimit), "", State}
- end,
- ?AUDI(Req, Bts, XER, Rcode),
- {RCode, RBody, RState}.
-
-%%%DELETE Methods
-delete("/application/:appname", Req, State) ->
- %Uninstall and delete a CDAP app
- {Bts, XER} = init_api_call(Req),
- Appname = leptus_req:param(Req, appname),
- {RCode, RBody, RState} = delete_app_helper(Appname, State, XER, Req),
- ?AUDI(Req, Bts, XER, Rcode),
- {RCode, RBody, RState}.
-
-%%%PUT Methods
parse_put_body(B) ->
%parse the PUT body to application
try
maps:get(<<"program_id">>, P),
maps:get(<<"program_pref">>, P)}
end, maps:get(<<"program_preferences">>, Body)),
- {pf, <<"program-flowlet">>, {Namespace, Streamname, JarURL, ArtifactName, ArtifactVersion, AppConfig, AppPreferences, ParsedServices, Programs, ParsedProgramPreferences}};
+ {<<"program-flowlet">>, {Namespace, Streamname, JarURL, ArtifactName, ArtifactVersion, AppConfig, AppPreferences, ParsedServices, Programs, ParsedProgramPreferences}};
<<"hydrator-pipeline">> ->
PipelineConfigJsonURL = maps:get(<<"pipeline_config_json_url">>, Body),
} end, D);
false -> [] %normalize optional user input into []; just prevents user from having to explicitly pass in []
end,
- {hp, <<"hydrator-pipeline">>, {Namespace, Streamname, PipelineConfigJsonURL, ParsedDependencies}}
+ {<<"hydrator-pipeline">>, {Namespace, Streamname, PipelineConfigJsonURL, ParsedDependencies}}
end
end
catch _:_ -> invalid
catch _:_ -> invalid
end.
-handle_reconfigure_put(Req, State, XER, Appname, ReqBody, AppnameToNS) ->
+handle_reconfigure_put(Req, State, XER, Appname, ReqBody) ->
%handle the reconfiguration put. broker out from the http call, and takes the lookup func as an arg, to allow for better unit testing.
%this is still not a pure function due to the workflows call, still needs enhancement
- case AppnameToNS(Appname) of
+ case ?MODULE:appname_to_field_vals(Appname, [<<"namespace">>]) of
none -> {404, "Reconfigure recieved but the app is not registered", State};
- Namespace ->
+ [Namespace] ->
ParsedBody = parse_reconfiguration_put_body(ReqBody),
case ParsedBody of
invalid -> {400, "Invalid PUT Reconfigure Body", State};
end
end.
-put("/application/:appname", Req, State) ->
- %create a new registration; deploys and starts a cdap application
- {Bts, XER} = init_api_call(Req),
- Appname = leptus_req:param(Req, appname),
- {RCode, RBody, RState} = case appname_to_field_vals(Appname, [<<"appname">>]) of
- [Appname] ->
- {400, "Put recieved on /application/:appname but appname is already registered. Call /application/:appname/reconfigure if trying to reconfigure or delete first", State};
- none -> %no matches, create the resource, return the application record
+handle_put(Req, State, XER, Appname, ReqBody, RequestUrl) ->
+ %use of ?MODULE here is due to the meck limitation described here: https://github.com/eproxus/meck
+ case ?MODULE:appname_to_field_vals(Appname, [<<"appname">>]) == none of
+ false -> {400, "Put recieved on /application/:appname but appname is already registered. Call /application/:appname/reconfigure if trying to reconfigure or delete first", State};
+ true ->
%Initial put requires the put body parameters
- case parse_put_body(leptus_req:body_raw(Req)) of
- %could not parse the body
- invalid -> {400, "Invalid PUT Body or unparseable URL", State};
-
- %unsupported cdap application type
- unsupported -> {404, "Unsupported CDAP Application Type", State};
-
- {Type, AppType, Params} ->
+ ParsedBody = parse_put_body(ReqBody),
+ case ParsedBody of
+ invalid -> {400, "Invalid PUT Body or unparseable URL", State}; %could not parse the body
+ unsupported -> {400, "Unsupported CDAP Application Type", State}; %unsupported cdap application type
+ {AppType, Params} ->
%form shared info
- %hateaos cuz they aintaos
- {RequestUrl,_} = cowboy_req:url((leptus_req:get_req(Req))),
Metricsurl = <<RequestUrl/binary, <<"/metrics">>/binary>>,
Healthcheckurl = <<RequestUrl/binary, <<"/healthcheck">>/binary>>,
try
- case Type of
- hp ->
+ case AppType of
+ <<"hydrator-pipeline">> ->
{Namespace, Streamname, PipelineConfigJsonURL, ParsedDependencies} = Params,
ConnectionURL = cdap_interface:form_stream_url_from_streamname(?CDAPURL, Namespace, Streamname),
-
- %TODO: This!
- ServiceEndpoints = [], %unclear if this is possible with pipelines
+ ServiceEndpoints = [], %TODO: unclear if this is possible with pipelines
%write into mnesia, deploy
A = #application{appname = Appname, apptype = AppType, namespace = Namespace, healthcheckurl = Healthcheckurl, metricsurl = Metricsurl, url = RequestUrl, connectionurl = ConnectionURL, serviceendpoints = ServiceEndpoints, creationtime=erlang:system_time()},
ok = workflows:deploy_hydrator_pipeline(Req, XER, Appname, Namespace, ?CDAPURL, PipelineConfigJsonURL, ParsedDependencies, ?CONSURL, RequestUrl, Healthcheckurl, ?HCInterval, ?AutoDeregisterAfter),
metrics(info, Req, [{bts, iso()}, {xer, XER}, {mod, mod()}, {msg, io_lib:format("New Hydrator Application Created: ~p", [lager:pr(A, ?MODULE)])}]), %see Record Pretty Printing: https://github.com/basho/lager
ok;
- pf ->
+ <<"program-flowlet">> ->
{Namespace, Streamname, JarURL, ArtifactName, ArtifactVersion, AppConfig, AppPreferences, ParsedServices, Programs, ParsedProgramPreferences} = Params,
%Form URLs that are part of the record
%NOTE: These are both String concatenation functions and neither make an HTTP call so not catching normal {Code, Status} return here
{500, "Please report this error", State}
end
end
- end,
+ end.
+
+%%% HTTP API CALLBACKS %%%
+init(_Route, _Req, State) ->
+ {ok, State}.
+terminate(_Reason, _Route, _Req, _State) ->
+ ok.
+%%%FOR Cors support
+%%%Note! only matches on host. Does not handle ports. See: https://github.com/s1n4/leptus/issues/55
+cross_domains(_Route, _Req, State) ->
+ {['_'], State}.
+
+%%%GET Methods
+get("/", Req, State) ->
+ %The broker's "info" endpoint; returns some possibly useful information
+ {Bts, XER} = init_api_call(Req),
+ Apps = util:get_all_appnames_from_db(),
+ {UT, _} = statistics(wall_clock),
+ CDAPVer = cdap_interface:get_cdap_cluster_version(XER, ?CDAPURL),
+ RB = {[
+ {<<"cdap cluster version">>, CDAPVer},
+ {<<"managed cdap url">>, ?CDAPURL},
+ {<<"cdap GUI port">>, cdap_interface:get_cdap_gui_port_from_version(CDAPVer)},
+ {<<"number of applications registered">>, length(Apps)},
+ {<<"uptime (s)">>, UT/1000},
+ {<<"broker API version">>, util:get_my_version()}
+ ]},
+ {RCode, RBody, RState} = {200, {json, RB}, State},
+ ?AUDI(Req, Bts, XER, Rcode),
+ {RCode, RBody, RState};
+get("/application", Req, State) ->
+ %get a list of all registered apps
+ {Bts, XER} = init_api_call(Req),
+ {RCode, RBody, RState} = {200, {json, util:get_all_appnames_from_db()}, State},
+ ?AUDI(Req, Bts, XER, Rcode),
+ {RCode, RBody, RState};
+get("/application/:appname", Req, State) ->
+ %get information about a registered application
+ {Bts, XER} = init_api_call(Req),
+ Appname = leptus_req:param(Req, appname),
+ {RCode, RBody, RState} = appname_to_application_http(XER, Appname, State),
+ ?AUDI(Req, Bts, XER, Rcode),
+ {RCode, RBody, RState};
+get("/application/:appname/metrics", Req, State) ->
+ %get metrics for a registered application
+ {Bts, XER} = init_api_call(Req),
+ Appname = leptus_req:param(Req, appname),
+ {RCode, RBody, RState} = case appname_to_field_vals(Appname, [<<"apptype">>, <<"namespace">>]) of
+ none -> {404, "", State};
+ [<<"program-flowlet">>, Namespace] ->
+ {ReturnCode, ReturnBody} = cdap_interface:get_app_metrics(XER, Appname, Namespace, ?CDAPURL), %warning, see note in README, this always reutrns 200
+ {ReturnCode, {json, ReturnBody}, State};
+ [<<"hydrator-pipeline">>, Namespace] ->
+ lager:warning("WARNING, metrics not actually implemented yet for pipelines!!"),
+ {ReturnCode, ReturnBody} = cdap_interface:get_pipeline_metrics(Appname, Namespace, ?CDAPURL),
+ {ReturnCode, {json, ReturnBody}, State}
+ end,
+ ?AUDI(Req, Bts, XER, Rcode),
+ {RCode, RBody, RState};
+get("/application/:appname/healthcheck", Req, State) ->
+ %get healthcheck of an application
+ {Bts, XER} = init_api_call(Req),
+ Appname = leptus_req:param(Req, appname),
+ lager:info(io_lib:format("Get Healthcheck recieved for ~s", [Appname])),
+ {RCode, RBody, RState} = case appname_to_field_vals(Appname, [<<"apptype">>, <<"namespace">>]) of
+ none -> {404, "", State};
+ [<<"program-flowlet">>, Namespace] ->
+ {cdap_interface:get_app_healthcheck(XER, Appname, Namespace, ?CDAPURL), "", State};
+ [<<"hydrator-pipeline">>, Namespace] ->
+ {cdap_interface:get_pipeline_healthcheck(XER, Appname, Namespace, ?CDAPURL, ?PipelineHealthLimit), "", State}
+ end,
+ ?AUDI(Req, Bts, XER, Rcode),
+ {RCode, RBody, RState}.
+%%%DELETE Methods
+delete("/application/:appname", Req, State) ->
+ %Uninstall and delete a CDAP app
+ {Bts, XER} = init_api_call(Req),
+ Appname = leptus_req:param(Req, appname),
+ {RCode, RBody, RState} = delete_app_helper(Appname, State, XER, Req),
+ ?AUDI(Req, Bts, XER, Rcode),
+ {RCode, RBody, RState}.
+%%%PUT Methods
+put("/application/:appname", Req, State) ->
+ %create a new registration; deploys and starts a cdap application
+ {Bts, XER} = init_api_call(Req),
+ Appname = leptus_req:param(Req, appname),
+ ReqBody = leptus_req:body_raw(Req),
+ {RequestUrl,_} = cowboy_req:url((leptus_req:get_req(Req))),
+ {RCode, RBody, RState} = handle_put(Req, State, XER, Appname, ReqBody, RequestUrl),
?AUDI(Req, Bts, XER, Rcode),
{RCode, RBody, RState};
put("/application/:appname/reconfigure", Req, State) ->
{Bts, XER} = init_api_call(Req),
Appname = leptus_req:param(Req, appname),
ReqBody = leptus_req:body_raw(Req),
- AppnameToNS = fun(App) ->
- X = appname_to_field_vals(App, [<<"namespace">>]),
- case X of
- none -> X;
- [Namespace] -> Namespace
- end
- end,
- {RCode, RBody, RState} = handle_reconfigure_put(Req, State, XER, Appname, ReqBody, AppnameToNS),
+ {RCode, RBody, RState} = handle_reconfigure_put(Req, State, XER, Appname, ReqBody),
?AUDI(Req, Bts, XER, Rcode),
{RCode, RBody, RState}.
-import(resource_handler, [
parse_put_body/1,
parse_reconfiguration_put_body/1,
- handle_reconfigure_put/6
+ handle_reconfigure_put/5,
+ handle_put/6
]).
parse_put_body_test() ->
[{<<"Greeting">>,<<"greet">>,<<"GET">>}],
[#program{type = <<"flows">>, id = <<"WhoFlow">>}, #program{type = <<"services">>, id = <<"Greeting">>}],
[{<<"flows">>,<<"WhoFlow">>,#{<<"foopprog">>=><<"barpprog">>}}]},
- Expected = {pf, <<"program-flowlet">>, ExpectedL},
+ Expected = {<<"program-flowlet">>, ExpectedL},
?assert(parse_put_body(jiffy:encode(Valid)) == Expected),
ValidHydrator1 =
{<<"streamname">>, <<"sn">>},
{<<"pipeline_config_json_url">>, "www.foo.com"}
]},
- ExpectedHy1 = {hp,<<"hydrator-pipeline">>,{<<"ns">>,<<"sn">>,"www.foo.com",[]}},
+ ExpectedHy1 = {<<"hydrator-pipeline">>, {<<"ns">>,<<"sn">>,"www.foo.com",[]}},
?assert(parse_put_body(jiffy:encode(ValidHydrator1)) == ExpectedHy1),
ValidHydrator2 =
]}
]},
%{hp, <<"hydrator-pipeline">>, {Namespace, Streamname, PipelineConfigJsonURL, ParsedDependencies}}
- ExpectedHy2 = {hp,<<"hydrator-pipeline">>,{<<"ns">>,<<"sn">>,"www.foo.com",[{"system:cdap-data-pipeline[4.1.0,5.0.0)",<<"art carney">>,"1.0.0-SNAPSHOT",<<"www.foo.com/sup/baphomet.jar">>,<<"www.foo2.com/sup/baphomet.jar">>}]}},
+ ExpectedHy2 = {<<"hydrator-pipeline">>, {<<"ns">>,<<"sn">>,"www.foo.com",[{"system:cdap-data-pipeline[4.1.0,5.0.0)",<<"art carney">>,"1.0.0-SNAPSHOT",<<"www.foo.com/sup/baphomet.jar">>,<<"www.foo2.com/sup/baphomet.jar">>}]}},
?assert(parse_put_body(jiffy:encode(ValidHydrator2)) == ExpectedHy2),
- InvalidType = {[{<<"cdap_application_type">>, <<"NOT TODAY">>}]},
- erlang:display(parse_put_body(jiffy:encode(InvalidType))),
- ?assert(parse_put_body(jiffy:encode(InvalidType)) == unsupported),
+ %Test the unexpected cases
+ EmptyD = dict:new(),
+ try meck:new(resource_handler, [passthrough]) catch _:_ -> ok end,
+ meck:expect(resource_handler, appname_to_field_vals, fun(X, [<<"appname">>]) ->
+ case X of
+ <<"notexist">> -> none;
+ <<"exist">> -> [<<"exist">>]
+ end
+ end),
+
+
+ %check already exists
+ ?assert(handle_put("", EmptyD, "textxer", <<"exist">>, Valid, "www.validurl.com") == {400, "Put recieved on /application/:appname but appname is already registered. Call /application/:appname/reconfigure if trying to reconfigure or delete first", EmptyD}),
+
+ InvalidType = jiffy:encode({[{<<"cdap_application_type">>, <<"NOT TODAY">>}]}),
+ ?assert(parse_put_body(InvalidType) == unsupported),
+ ?assert(handle_put("", EmptyD, "textxer", <<"notexist">>, InvalidType, "www.validurl.com") == {400,"Unsupported CDAP Application Type", EmptyD}),
InvalidMissing = {[
{<<"cdap_application_type">>, <<"program-flowlet">>},
{<<"namespace">>, <<"ns">>}
]},
- ?assert(parse_put_body(jiffy:encode(InvalidMissing)) == invalid).
+ ?assert(parse_put_body(jiffy:encode(InvalidMissing)) == invalid),
+ ?assert(handle_put("", EmptyD, "textxer", <<"notexist">>, InvalidMissing, "www.validurl.com") == {400, "Invalid PUT Body or unparseable URL", EmptyD}),
+
+ InvalidMissing2 = {[{<<"malformed">>, <<"i am">>}]},
+ ?assert(parse_put_body(jiffy:encode(InvalidMissing2)) == invalid),
+ ?assert(handle_put("", EmptyD, "textxer", <<"notexist">>, InvalidMissing2, "www.validurl.com") == {400, "Invalid PUT Body or unparseable URL", EmptyD}),
+
+ meck:unload(resource_handler).
+
reconfiguration_put_test() ->
%test reconfiguring with an invalid PUT body (missing "reconfiguration_type")
- AppnameToNS = fun(X) ->
- case X of
- <<"notexist">> -> none;
- <<"exist">> -> <<"ns">>
- end
- end,
EmptyD = dict:new(),
+ try meck:new(resource_handler, [passthrough]) catch _:_ -> ok end,
+ meck:expect(resource_handler, appname_to_field_vals, fun(X, _) ->
+ case X of
+ <<"notexist">> -> none;
+ <<"exist">> -> [<<"ns">>]
+ end
+ end),
I1 = jiffy:encode({[{<<"config">>, <<"bar">>}]}),
?assert(parse_reconfiguration_put_body(I1) == invalid),
- ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"exist">>, I1, AppnameToNS) == {400,"Invalid PUT Reconfigure Body",EmptyD}),
+ ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"exist">>, I1) == {400,"Invalid PUT Reconfigure Body",EmptyD}),
%test reconfiguring with an invalid PUT body (missing app_config)
I2 = jiffy:encode({[{<<"reconfiguration_type">>, <<"program-flowlet-app-config">>}, {<<"foo">>, <<"bar">>}]}),
?assert(parse_reconfiguration_put_body(I2) == invalid),
- ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"exist">>, I2, AppnameToNS) == {400,"Invalid PUT Reconfigure Body",EmptyD}),
+ ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"exist">>, I2) == {400,"Invalid PUT Reconfigure Body",EmptyD}),
%test reconfiguring an invalid (unimplemented) type
I3 = jiffy:encode({[{<<"config">>, <<"bar">>}, {<<"reconfiguration_type">>, <<"EMPTINESS">>}]}),
?assert(parse_reconfiguration_put_body(I3) == notimplemented),
- ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"exist">>, I3, AppnameToNS) == {501,"This type of reconfiguration is not implemented",EmptyD}),
+ ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"exist">>, I3) == {501,"This type of reconfiguration is not implemented",EmptyD}),
Valid = jiffy:encode({[{<<"config">>, {[{<<"foo">>, <<"bar">>}]}}, {<<"reconfiguration_type">>,<<"program-flowlet-app-config">>}]}),
?assert(parse_reconfiguration_put_body(Valid) == {<<"program-flowlet-app-config">>,#{<<"foo">>=><<"bar">>}}),
%test for valid but missing
%?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"exist">>, I3, AppnameToNS) == {501,"This type of reconfiguration is not implemented",EmptyD}),
- ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"notexist">>, Valid, AppnameToNS) == {404,"Reconfigure recieved but the app is not registered", EmptyD}).
+ ?assert(handle_reconfigure_put("", EmptyD, "testXER", <<"notexist">>, Valid) == {404,"Reconfigure recieved but the app is not registered", EmptyD}),
+
+ meck:unload(resource_handler).
+
% Licensed under the Apache License, Version 2.0 (the "License");
% you may not use this file except in compliance with the License.
% You may obtain a copy of the License at
-%
+%
% http://www.apache.org/licenses/LICENSE-2.0
-%
+%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS,
% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-module(util).
-include("application.hrl").
--export([concat/1,
- get_platform_envs_and_config/0,
+-export([concat/1,
+ get_platform_envs_and_config/0,
resolve_cbs/2,
initialize_database/0,
get_all_appnames_from_db/0,
%http://stackoverflow.com/questions/39757020/erlang-drying-up-stringbinary-concatenation
%NOTE! Does not work or bomb when an element in the list is an atom. Must be a string or binary. Maybe add a check for this
-to_string(Value) when is_binary(Value) -> binary_to_list(Value);
-to_string(Value) -> Value.
-concat(List) ->
- lists:flatten(lists:map(fun to_string/1, List)).
+concat(List) -> lists:flatten(lists:map(fun to_str/1, List)).
resolve_cbs(XER, ConsulURL) ->
%Ideally this function would dissapear if we get real DNS. This essentially is doing an SRV record lookup every time someone needs the bindng URL
concat(["http://", IP, ":", integer_to_binary(Port)]).
get_platform_envs_and_config() ->
- %Get platform envs needed for broker operation, then fetch my config.
+ %Get platform envs needed for broker operation, then fetch my config.
%If something critical fails, returns [], else [ConsulURL, CDAPUrl, BoundConfigMap]
MyName = os:getenv("HOSTNAME"),
ConsulHost = os:getenv("CONSUL_HOST"),
- case MyName == false orelse ConsulHost == false of
+ case MyName == false orelse ConsulHost == false of
true -> [];
- false ->
+ false ->
%build Consul URL
ConsulURL = concat(["http://", ConsulHost, ":8500"]),
-
+
%Bind my own config map
%generate my own XER here
XER = gen_uuid(),
{200, BoundConfig} = consul_interface:consul_bind_config(XER, MyName, ConsulURL),
BoundConfigMap = jiffy:decode(jiffy:encode(BoundConfig), [return_maps]), %kind of an interesting way to turn an erlang proplist into a map
-
+
%Here, we waterfall looking for "CDAP_CLUSTER_TO_MANAGE".
%First, we will check for environmnental variables for a cluster *NAME*
%If that is not found, then we will check out bound config for a fully bound URL
%If that is also not found, let it crash baby.
- CDAPURL = case os:getenv("CDAP_CLUSTER_TO_MANAGE") of
- false ->
+ CDAPURL = case os:getenv("CDAP_CLUSTER_TO_MANAGE") of
+ false ->
list_to_binary(concat(["http://", lists:nth(1, maps:get(<<"cdap_cluster_to_manage">>, BoundConfigMap))])); %cbs returns ip:port. need http:// or will get "no adaptors found" error
- CDAPName ->
+ CDAPName ->
{IP, Port} = consul_interface:consul_get_service_ip_port(XER, CDAPName, ConsulURL),
list_to_binary(concat(["http://", IP, ":", integer_to_binary(Port)]))
end,
initialize_database() ->
%Create the database (currently MNesia) if it does not exist, and the application table.
- %Or, do nothing.
+ %Or, do nothing.
N = node(),
lager:info(io_lib:format("Initializing database. My node name is ~s", [N])),
%set MNesia dir
application:set_env(mnesia, dir, "/var/mnesia/"),
-
+
%stop if running, can't create schema if it is. Dont check status, OK if stopped
mnesia:stop(),
%erlang:display(mnesia:delete_schema([N])),
mnesia:create_schema([N]),
%start MNesia, assert it works
-
+
ok = mnesia:start(), %start MNesia, bomb if alreay started, should not happen
lager:info("Mnesia started"),
-
+
%try to create the table, or if it exists, do nothing
%erlang:display(mnesia:delete_table(application)),
case mnesia:create_table(application, [{attributes, record_info(fields, application)}, {disc_copies, [N]}]) of
{aborted,{already_exists,application}} ->
lager:info("Application table already exists");
- {atomic,ok} ->
+ {atomic,ok} ->
lager:info(io_lib:format("Created application table on ~s", [N]))
end,
case mnesia:create_table(prog_flow_supp, [{attributes, record_info(fields, prog_flow_supp)}, {disc_copies, [N]}]) of
{aborted,{already_exists, prog_flow_supp}} ->
lager:info("prog_flow_supp table already exists");
- {atomic,ok} ->
+ {atomic,ok} ->
lager:info(io_lib:format("Created prog_flow_supp table on ~s", [N]))
end,
%wait up to 30s for the table to come up. Usually instantaneous. If it takes more crash abd burn
ok = mnesia:wait_for_tables([application, prog_flow_supp], 30000),
ok.
-
+
get_all_appnames_from_db() ->
{atomic, Apps} = mnesia:transaction(fun() -> mnesia:match_object(application, #application{_ = '_'}, read) end),
lists:map(fun(X) -> {application, Appname,_,_,_,_,_,_,_,_} = X,
Egs - Sgs.
to_str("") -> "";
-to_str(Term) -> lists:flatten(io_lib:format("~p", [Term])).
+to_str(Term) when is_list(Term) -> Term;
+to_str(Term) when is_binary(Term) -> binary_to_list(Term);
+to_str(Term) ->
+ lists:flatten(io_lib:format("~p", [Term])).
-spec ip_to_str({inet:ip_address(), inet:port_number()}) -> binary().
%nasty.. I miss pythons x <= Foo <= Y syntax.. or something mathematical like Foo in [X..Y].. erlang not good 4 math
%This is very similar to the maps:merge/2 builtin but that will inject keys of newconfig that were not in oldconfig. We need a "RIGHT JOIN"
NCKeys = maps:keys(NewConfig),
ConfigOverlaps = [X || X <- NCKeys, maps:is_key(X, OldConfig)],
- case ConfigOverlaps of
+ case ConfigOverlaps of
[] -> nooverlap;
- _ ->
+ _ ->
%we have an entry that should be in app config
%build a new map with just the keys to update
Pred = fun(X,_) -> lists:member(X, ConfigOverlaps) end,
end.
ejson_to_map(E) ->
- %takes the jiffy "ejson: format of {[{<<"foo">>, <<"bar">>}, {<<"foo2">>, <<"bar2">>}]} and turns it into a map,
+ %takes the jiffy "ejson: format of {[{<<"foo">>, <<"bar">>}, {<<"foo2">>, <<"bar2">>}]} and turns it into a map,
%usefu because ejsons do not appear to be order-independent-comparable, but maps are (e.g., two maps are equal if all their k+v are equal but agnostic to order)
jiffy:decode(jiffy:encode(E), [return_maps]).
% Licensed under the Apache License, Version 2.0 (the "License");
% you may not use this file except in compliance with the License.
% You may obtain a copy of the License at
-%
+%
% http://www.apache.org/licenses/LICENSE-2.0
-%
+%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS,
% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-module(util_tests).
-include_lib("eunit/include/eunit.hrl").
-import(util, [
- iso_elapsed/2,
+ iso_elapsed/2,
ip_to_str/1,
update_with_new_config_map/2,
- ejson_to_map/1]).
+ ejson_to_map/1,
+ to_str/1
+ ]).
+
+to_str_test() ->
+ ?assert(to_str("") == ""),
+ ?assert(to_str(<<"asdf">>) == "asdf").
iso_elapsed_test() ->
?assert(iso_elapsed(<<"2017-04-27T18:38:10Z">>, <<"2017-04-27T18:38:08Z">>) == 2),
<body>
<div class="container">
<h1>CDAP Broker API</h1>
- <p class="sw-info">Version: <span class="sw-info-version">4.0.5</span></p>
+ <p class="sw-info">Version: <span class="sw-info-version">4.0.6</span></p>
<p></p>
{
"swagger": "2.0",
"info": {
- "version": "4.0.5",
+ "version": "4.0.6",
"title": "CDAP Broker API"
},
"paths": {
# This is your document metadata
info:
- version: "4.0.5"
+ version: "4.0.6"
title: CDAP Broker API
paths:
hydrator_test(C) ->
%test te app by injecting some data into the stream and getting it out
%Sleeping since HTTP services may still be booting up: see https://issues.cask.co/browse/CDAP-812
- ok = timer:sleep(30000), %30s
+ ok = timer:sleep(60000), %60s
%curl into stream
{200, _} = httpabs:post(?XER, ?PLG(hydrator_stream_url, C), "text/plain", "beer, vodka, gin"),
%query data out
app_test(C) ->
%Sleeping since HTTP services may still be booting up: see https://issues.cask.co/browse/CDAP-812
- ok = timer:sleep(30000), %30s
+ ok = timer:sleep(60000), %60s
{200, _} = httpabs:post(?XER, ?PLG(stream_url, C), "text/plain", "'Prince of Darkness'"),
{200, "Hello 'Prince of Darkness'!"} = httpabs:get(?XER,?SC([?PLG(app_url, C), "/services/Greeting/methods/greet"])).
%do the reconfig
ReconfigMap = util:ejson_to_map({[{<<"foo REDUX EDITION">>, <<"bar">>}, {<<"LEAVE ME ALONE">>, <<"CONFIG EDITION">>}]}),
- %test httpabs bad body (not encoded as JSON)
- {400,"ERROR: The request Body is malformed"} = httpabs:put(?XER, ?SC([?PLG(broker_app_url, C), "/reconfigure"]), "application/json", {[{<<"reconfiguration_type">>, <<"program-flowlet-app-config">>}, {<<"config">>, ReconfigMap}]}),
%do it properly
{200, _} = httpabs:put(?XER, ?SC([?PLG(broker_app_url, C), "/reconfigure"]), "application/json", jiffy:encode({[{<<"reconfiguration_type">>, <<"program-flowlet-app-config">>},{<<"config">>, ReconfigMap}]})),
%test new config right in consul
%malformed Broker put
URL = ?SC([?PLG(broker_app_url, C), "FAILURETEST"]),
- Body = {[
- {<<"malformed">>, <<"i am">>}
- ]},
- {400, "State: Bad Request. Return Body: Invalid PUT Body or unparseable URL"} = httpabs:put(?XER, URL, "application/json", jiffy:encode(Body)),
%deploy a bad CDAP app with a bad program_id
Body2 = {[
{<<"programs">>, [{[{<<"program_type">>, <<"flows">>},{<<"program_id">>, <<"WhoFlow">>}]},{[{<<"program_type">>, <<"services">>},{<<"program_id">>, <<"Greeting">>}]}]},
{<<"program_preferences">>, [{[{<<"program_type">>,<<"flows">>}, {<<"program_id">>, <<"WhoFlow">>}, {<<"program_pref">>, ?PLG(whoflowpref, C)}]}]}
]},
- {404, _} = httpabs:put(?XER, URL, "application/json", jiffy:encode(Body3)),
-
- %try to deploy with a bad URL where bad means malformed
- Body4 = {[
- {<<"cdap_application_type">>, <<"program-flowlet">>},
- {<<"namespace">>, ?PLG(namespace, C)},
- {<<"streamname">>, ?PLG(streamname, C)},
- {<<"jar_url">>, <<"THIS IS NOT EVEN A URL WHAT ARE YOU DOING TO ME">>},
- {<<"artifact_name">>, ?PLG(art_name, C)},
- {<<"artifact_version">>, ?PLG(art_ver, C)},
- {<<"app_config">>, ?PLG(init_config, C)},
- {<<"app_preferences">>, ?PLG(init_preferences, C)},
- {<<"services">>, [{[{<<"service_name">>, <<"Greeting">>}, {<<"service_endpoint">>, <<"greet">>}, {<<"endpoint_method">>, <<"GET">>}]}]},
- {<<"programs">>, [{[{<<"program_type">>, <<"flows">>},{<<"program_id">>, <<"WhoFlow">>}]},{[{<<"program_type">>, <<"services">>},{<<"program_id">>, <<"Greeting">>}]}]},
- {<<"program_preferences">>, [{[{<<"program_type">>,<<"flows">>}, {<<"program_id">>, <<"WhoFlow">>}, {<<"program_pref">>, ?PLG(whoflowpref, C)}]}]}
- ]},
- {400,"State: Bad Request. Return Body: ERROR: The following URL is malformed: THIS IS NOT EVEN A URL WHAT ARE YOU DOING TO ME"} = httpabs:put(?XER, URL, "application/json", jiffy:encode(Body4))
- .
+ {404, _} = httpabs:put(?XER, URL, "application/json", jiffy:encode(Body3)).
delete_all(C) ->
%test invalid key
Code Review / dcaegen2 / platform / cdapbroker.git / commitdiff