%%%------------------------------------------------------------------- %%% @author Heinz Nikolaus Gies %%% @copyright (C) 2014, Heinz Nikolaus Gies %%% @doc This module provides a wrapper around DalmatinerDB's TCP %%% protocol. It can be used for both sending and requesting data and %%% does support the streaming send mode for TCP. %%% %%% The {@link connect/2} call will return a new connection with the %%% required information stored. It also contens a gen_tco connection %%% so the same limits of ownership apply here. %%% %%% The connection is not guarantted to be alive at all the time and %%% in the case of a failure will be tried to be reestablished before %%% forwarding the error to the caller. %%% %%% Once entering the stream mode by calling {@link stream/2} only %%% the send command is supported, other commands will cause an error, %%% however not disconect the system. %%% %%% @end %%% Created : 15 Dec 2014 by Heinz Nikolaus Gies %%%------------------------------------------------------------------- -module(ddb_tcp). -include_lib("dproto/include/dproto.hrl"). -define(TIMEOUT, 30000). -export([ connect/2, connect/1, mode/1, connected/1, close/1, stream_mode/3, list/1, list/2, list/3, get/5, send/4, batch_start/2, batch/2, batch/3, batch_end/1 ]). -export_type([connection/0]). -type socket() :: port(). -record(ddb_connection, {socket :: socket() | undefined, host, port, mode = normal, bucket, error = none, delay = 1, batch = false}). %%-------------------------------------------------------------------- %% @type connection(). %% %% A connection to the DalmatinerDB backend server. %% @end %%-------------------------------------------------------------------- -opaque connection() :: #ddb_connection{}. -define(OPTS, [binary, {packet, 4}, {active, false}]). %%-------------------------------------------------------------------- %% @doc Connects to a DalmatinerDB instance. It will try to create a %% gen_tcp connection however it will return successfully even if that %% could not be established! %% %% To test for connection use {@link connected/1}. %% %% @end %%-------------------------------------------------------------------- -spec connect(Host :: inet:ip_address() | inet:hostname(), Port :: inet:port_number()) -> {ok, connection()}. connect(Host, Port) -> case gen_tcp:connect(Host, Port, ?OPTS) of {ok, Socket} -> {ok, #ddb_connection{ socket = Socket, host = Host, port = Port }}; {error, E} -> {ok, #ddb_connection{ socket = undefined, host = Host, port = Port, error = E }} end. %%-------------------------------------------------------------------- %% @doc Forces a reconnect on a disconnected connection, has no effect %% on connections that are still connected. %% %% @end %%-------------------------------------------------------------------- -spec connect(Connection :: connection()) -> {ok, connection()}. connect(Con) -> {ok, reconnect(Con)}. %%-------------------------------------------------------------------- %% @doc Shows what mode the connection is in and if in stream mode %% which bucket the data is streamed to. %% %% @end %%-------------------------------------------------------------------- -spec mode(Connection :: connection()) -> {ok, normal} | {ok, {stream, Bucket :: binary()}} | {ok, {batch, Bucket :: binary()}}. mode(#ddb_connection{mode = stream, bucket=Bucket, batch = true}) -> {ok, {batch, Bucket}}; mode(#ddb_connection{mode = stream, bucket=Bucket}) -> {ok, {stream, Bucket}}; mode(#ddb_connection{mode = normal}) -> {ok, normal}. %%-------------------------------------------------------------------- %% @doc Shows weather a connection is currently connected to the kback %% backend or awaiting reconncet. %% %% @end %%-------------------------------------------------------------------- -spec connected(Connection :: connection()) -> boolean(). connected(#ddb_connection{socket = undefined}) -> false; connected(_) -> true. %%-------------------------------------------------------------------- %% @doc Puts a connection into stream mode, if the connection was in %% stream mode before an error is returned unless the requested stream %% parameters are equal to the current ones. %% %% @end %%-------------------------------------------------------------------- -spec stream_mode(Bucket :: binary(), Delay :: pos_integer(), Connection :: connection()) -> {ok, Connection :: connection()} | {error, Error :: inet:posix(), Connection :: connection()} | {error, {stream, OldBucket :: binary(), OldDelay :: pos_integer()}, Connection :: connection()}. stream_mode(Bucket, Delay, Con = #ddb_connection{mode = stream, bucket = Bucket, delay = Delay}) -> {ok, Con}; stream_mode(_Bucket, _Delay, Con = #ddb_connection{mode = stream, bucket = OldBucket, delay = OldDelay}) -> {error, {stream, OldBucket, OldDelay}, Con}; stream_mode(Bucket, Delay, Con) -> Bin = dproto_tcp:encode({stream, Bucket, Delay}), Con1 = Con#ddb_connection{mode = stream, bucket = Bucket, delay = Delay}, case send_bin(Bin, Con1) of {ok, Con2} -> {ok, reset_state(Con2)}; E -> E end. %%-------------------------------------------------------------------- %% @doc Starts a batch transfair for a given timeslot, once started %% Additional metrics with the same time can be send via the %% {@link batch/2} and {@link batch/3} functions. %% %% @end %%-------------------------------------------------------------------- -spec batch_start(Time :: non_neg_integer(), Connection :: connection()) -> {ok, Connection :: connection()} | {error, {batch, Time :: non_neg_integer()}, Connection :: connection()} | {error, Error :: inet:posix(), Connection :: connection()} | {error, {bad_mode, normal}, Connection :: connection()}. batch_start(_Time, Con = #ddb_connection{batch = Time}) when is_integer(Time) -> {error, {batch, Time}, Con}; batch_start(_Time, Con = #ddb_connection{mode = normal}) -> {error, {bad_mode, normal}, Con}; batch_start(Time, Con) when is_integer(Time), Time >= 0 -> Con1 = Con#ddb_connection{batch = Time}, Bin = dproto_tcp:encode({batch, Time}), case send_bin(Bin, Con1) of {ok, Con2} -> {ok, Con2}; E -> E end. %%-------------------------------------------------------------------- %% @doc Sends a batch of multiple values with a single tcp call. %% @end %%-------------------------------------------------------------------- -spec batch([{Metric :: binary() | [binary()], Point :: integer() | binary()}], Connection :: connection()) -> {ok, Connection :: connection()} | {error, Error :: inet:posix(), Connection :: connection()} | {error, no_batch, Connection :: connection()}. batch(MPs, Con = #ddb_connection{batch = _Time}) when is_integer(_Time), is_list(MPs) -> Bin = << <<(to_batch(Metric, Point))/binary>> || {Metric, Point} <- MPs >>, case send_bin(Bin, Con) of {ok, Con1} -> {ok, Con1}; E -> E end; batch(_MPs, Con) -> {error, no_batch, Con}. %%-------------------------------------------------------------------- %% @doc Sends a single metric value pair for a batch %% @end %%-------------------------------------------------------------------- -spec batch(Metric :: binary() | [binary()], Point :: integer() | binary(), Connection :: connection()) -> {ok, Connection :: connection()} | {error, Error :: inet:posix(), Connection :: connection()} | {error, no_batch, Connection :: connection()}. batch(Metric, Point, Con) when is_integer(Point) -> batch(Metric, mmath_bin:from_list([Point]), Con); batch([_M | _] = Metric, Point, Con) when is_binary(_M) -> batch(dproto:metric_from_list(Metric), Point, Con); batch(Metric, Point, Con = #ddb_connection{batch = _Time}) when is_binary(Metric), is_binary(Point), is_integer(_Time) -> Bin = dproto_tcp:encode({batch, Metric, Point}), case send_bin(Bin, Con) of {ok, Con1} -> {ok, Con1}; E -> E end; batch(_Metric, _Point, Con) -> {error, no_batch, Con}. %%-------------------------------------------------------------------- %% @doc Finalizes the batch trainsfair %% @end %%-------------------------------------------------------------------- -spec batch_end(Connection :: connection()) -> {error, Error :: inet:posix(), Connection :: connection()} | {ok, Connection :: connection()}. batch_end(Con = #ddb_connection{batch = _Time}) when is_integer(_Time) -> Con1 = Con#ddb_connection{batch = false}, Bin = dproto_tcp:encode(batch_end), case send_bin(Bin, Con1) of {ok, Con2} -> {ok, Con2}; E -> E end; batch_end(Con) -> Con1 = Con#ddb_connection{batch = false}, {ok, Con1}. %%-------------------------------------------------------------------- %% @doc Retrives a list fo all buckets on the srever. Returns an error %% when in stream mode. %% %% @end %%-------------------------------------------------------------------- -spec list(Connection :: connection()) -> {ok, [Bucket :: binary()], Connection :: connection()} | {error, stream, Connection :: connection()}. list(Con = #ddb_connection{mode = normal}) -> do_list(send_bin(dproto_tcp:encode(buckets), Con)); list(Con) -> {error, stream, Con}. %%-------------------------------------------------------------------- %% @doc Retrives a list fo all metrics in a bucket. Returns an error %% when in stream mode. %% %% @end %%-------------------------------------------------------------------- -spec list(Bucket :: binary(), Connection :: connection()) -> {ok, [Metric :: binary()], Connection :: connection()} | {error, stream, Connection :: connection()}. list(Bucket, Con = #ddb_connection{mode = normal}) -> do_list(send_bin(dproto_tcp:encode({list, Bucket}), Con)); list(_Bucket, Con) -> {error, stream, Con}. %%-------------------------------------------------------------------- %% @doc Retrives a list fo all metrics with a given prefix. Returns an %% error when in stream mode. %% %% @end %%-------------------------------------------------------------------- -spec list(Bucket :: binary(), Prefix :: binary(),Connection :: connection()) -> {ok, [Metric :: binary()], Connection :: connection()} | {error, stream, Connection :: connection()}. list(Bucket, Prefix, Con = #ddb_connection{mode = normal}) -> do_list(send_bin(dproto_tcp:encode({list, Bucket, Prefix}), Con)); list(_Bucket, _Prefix, Con) -> {error, stream, Con}. %%-------------------------------------------------------------------- %% @doc Retrives a range of data from a metric or an error when in %% stream mode. %% %% @end %%-------------------------------------------------------------------- -spec get(Bucket :: binary(), Metric :: binary(), Time :: pos_integer(), Count :: pos_integer(), Connection :: connection()) -> {ok, {Resolution :: pos_integer(), Data :: binary()}, Connection :: connection()} | {error, Error :: inet:posix(), Connection :: connection()} | {error, stream, Connection :: connection()}. get(Bucket, Metric, Time, Count, Con = #ddb_connection{mode = normal}) -> case send_bin(dproto_tcp:encode({get, Bucket, Metric, Time, Count}), Con) of {ok, Con1 = #ddb_connection{socket = Socket}} -> case gen_tcp:recv(Socket, 0, ?TIMEOUT) of {ok, <>} -> {ok, {Resolution, D}, Con1}; {error, E} -> {error, E, close(Con1)} end; E -> E end; get(_, _, _, _, Con) -> {error, stream, Con}. %%-------------------------------------------------------------------- %% @doc Sends data to the server on streaming mode. Returns an error %% when in stream mode. %% %% @end %%-------------------------------------------------------------------- -spec send(Metric :: binary() | [binary()], Time :: pos_integer(), Points :: [integer()] | binary(), Connection :: connection()) -> {ok, Connection :: connection()} | {error, Error :: inet:posix(), Connection :: connection()} | {error, no_stream, Connection :: connection()}. send([_M | _] = Metric, Time, Points, Con = #ddb_connection{mode = stream}) when is_binary(_M) -> send(dproto:metric_from_list(Metric), Time, Points, Con); send(_, _, _, Con =#ddb_connection{batch = Time}) when is_integer(Time) -> {error, {batch, Time}, Con}; send(Metric, Time, Points, Con = #ddb_connection{mode = stream}) -> send_bin(dproto_tcp:encode({stream, Metric, Time, Points}), Con); send(_, _, _, Con) -> {error, no_stream, Con}. %%-------------------------------------------------------------------- %% @doc Forces to close a conneciton. %% %% @end %%-------------------------------------------------------------------- -spec close(Connection :: connection()) -> Connection :: connection(). close(Con = #ddb_connection{socket = undefined}) -> Con; close(Con = #ddb_connection{socket = Sock}) -> gen_tcp:close(Sock), Con#ddb_connection{socket = undefined}. %%%=================================================================== %%% Internal functions %%%=================================================================== send_bin(Bin, Con = #ddb_connection{socket = undefined}) -> send1(Bin, reconnect(Con)); send_bin(Bin, Con = #ddb_connection{socket = Sock}) -> case gen_tcp:send(Sock, Bin) of {error, _E} -> send1(Bin, reconnect(close(Con))); _ -> {ok, Con} end. send1(_Bin, Con = #ddb_connection{socket = undefined, error = E}) -> {error, E, Con}; send1(Bin, Con = #ddb_connection{socket = Sock}) -> case gen_tcp:send(Sock, Bin) of {error, E} -> {error, E, close(Con)}; _ -> {ok, Con} end. -spec reconnect(connection()) -> connection(). reconnect(Con = #ddb_connection{socket = undefined, host = Host, port = Port}) -> case gen_tcp:connect(Host, Port, ?OPTS) of {ok, Socket} -> reset_stream(Con#ddb_connection{socket = Socket, error = none}); {error, E} -> Con#ddb_connection{error = E} end; reconnect(Con) -> Con. reset_stream(Con = #ddb_connection{socket = _S, mode = stream, bucket = Bucket, delay = Delay}) when _S /= undefined -> Bin = dproto_tcp:encode({stream, Bucket, Delay}), case send_bin(Bin, Con) of {ok, Con1} -> reset_batch(reset_state(Con1)); E -> E end; reset_stream(Con) -> reset_state(Con). reset_batch(Con = #ddb_connection{batch = Time}) when is_integer(Time) -> Bin = dproto_tcp:encode({batch, Time}), case send_bin(Bin, Con) of {ok, Con1} -> Con1; E -> E end; reset_batch(Con) -> Con. reset_state(Con = #ddb_connection{socket = Socket, mode = stream}) when Socket /= undefined -> inet:setopts(Socket, [{packet, 0}]), Con; reset_state(Con) -> Con. decode_metrics(<<>>, Acc) -> Acc; decode_metrics(<>, Acc) -> decode_metrics(R, [M | Acc]). do_list({ok, Con1 = #ddb_connection{socket = S}}) -> case gen_tcp:recv(S, 0, ?TIMEOUT) of {ok, <>} -> {ok, decode_metrics(Reply, []), Con1}; {error, E} -> {error, E, Con1} end; do_list(Error) -> Error. to_batch(Metric, Point) when is_integer(Point) -> to_batch(Metric, mmath_bin:from_list([Point])); to_batch([_M | _] = Metric, Point) when is_binary(_M) -> to_batch(dproto:metric_from_list(Metric), Point); to_batch(Metric, Point) when is_binary(Metric), is_binary(Point) -> dproto_tcp:encode({batch, Metric, Point}).