diff --git a/apps/iot/src/influxdb/influx_client.erl b/apps/iot/src/influxdb/influx_client.erl index 9739193..eae88d7 100644 --- a/apps/iot/src/influxdb/influx_client.erl +++ b/apps/iot/src/influxdb/influx_client.erl @@ -12,7 +12,7 @@ -behaviour(gen_server). %% API --export([start_link/1, write/4, write/5, write_data/4, create_bucket/3]). +-export([start_link/1, write/4, write/5, write_data/4, create_bucket/3, get_orgs/1]). -export([get_precision/1]). -export([get_bucket/1]). -export([create_all_buckets/1]). @@ -97,6 +97,8 @@ write(Pid, Bucket, Org, Points) when is_pid(Pid), is_binary(Bucket), is_binary(O -spec create_all_buckets(SuperDeviceUUIDs :: list()) -> no_return(). create_all_buckets(SuperDeviceUUIDs) when is_list(SuperDeviceUUIDs) -> poolboy:transaction(influx_pool, fun(Pid) -> + {ok, Orgs} = get_orgs(Pid), + lager:debug("[influx_client] orgs is: ~p", [Orgs]), lists:foreach(fun(DeviceUUID) -> Bucket = <<"metric_", DeviceUUID/binary>>, influx_client:create_bucket(Pid, ?DEFAULT_ORG, Bucket) @@ -112,6 +114,10 @@ write(Pid, Bucket, Org, Precision, Points) when is_pid(Pid), is_binary(Bucket), create_bucket(Pid, Org, Bucket) when is_pid(Pid), is_binary(Org), is_binary(Bucket) -> gen_server:call(Pid, {create_bucket, Org, Bucket}). +-spec get_orgs(Pid :: pid()) -> {ok, Orgs :: list()} | {error, Reason :: any()}. +get_orgs(Pid) when is_pid(Pid) -> + gen_server:call(Pid, get_orgs). + %% @doc Spawns the server and registers the local name (unique) -spec(start_link(Opts :: list()) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}). @@ -144,6 +150,49 @@ init([InfluxProps]) -> {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | {stop, Reason :: term(), NewState :: #state{}}). +handle_call(get_orgs, _From, State = #state{token = Token, host = Host, port = Port}) -> + %% 处理headers + Headers = [ + {<<"Accept">>, <<"application/json">>}, + {<<"Authorization">>, <<"Token ", Token/binary>>} + ], + + Url = uri_string:normalize(#{ + scheme => "http", + host => Host, + port => Port, + path => "/api/v2/orgs" + }), + + lager:debug("[influx_client] url is: ~p, headers: ~p", [Url, Headers]), + case hackney:request(get, Url, Headers, [], [{pool, false}]) of + {ok, 200, _RespHeaders, ClientRef} -> + case hackney:body(ClientRef) of + {ok, RespBody} -> + hackney:close(ClientRef), + case catch jiffy:decode(RespBody, [return_maps]) of + #{<<"orgs">> := Orgs} -> + {reply, {ok, Orgs}, State}; + Error -> + {reply, Error, State} + end; + {error, Error} -> + hackney:close(ClientRef), + {reply, {error, Error}, State} + end; + {ok, StatusCode, _, ClientRef} -> + case hackney:body(ClientRef) of + {ok, RespBody} -> + hackney:close(ClientRef), + {reply, {error, {StatusCode, RespBody}}, State}; + {error, Error} -> + hackney:close(ClientRef), + {reply, {error, Error}, State} + end; + {error, Reason} -> + {reply, {error, Reason}, State} + end; + handle_call({create_bucket, Org, Bucket}, _From, State = #state{host = Host, port = Port, token = Token}) -> %% 处理headers Headers = [