anlicheng/ekfa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix data

Browse Source
This commit is contained in:
anlicheng 2025-05-06 22:54:30 +08:00
parent 3797fda241
commit cceec7c11c
9 changed files with 115 additions and 269 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -3,6 +3,8 @@ efka
An OTP application
1. 先解决数据的上行问题
Build
-----

5
apps/efka/include/message_pb.hrl
View File

@ -81,10 +81,7 @@
-define('DATA_PB_H', true).
-record(data,
{service_id = <<>> :: unicode:chardata() | undefined, % = 1, optional
device_uuid = <<>> :: unicode:chardata() | undefined, % = 2, optional
timestamp = 0 :: integer() | undefined, % = 3, optional, 32 bits
tags = [] :: [{unicode:chardata(), unicode:chardata()}] | undefined, % = 4
fields = <<>> :: unicode:chardata() | undefined % = 5, optional
metric = <<>> :: unicode:chardata() | undefined % = 2, optional
}).
-endif.

9
apps/efka/src/efka_client.erl → apps/efka/src/client/efka_client.erl
View File

@ -195,8 +195,11 @@ handle_call({request_param, ReceiverPid}, _From, State = #state{socket = Socket,
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_cast({send_metric_data, Fields, Tags}, State = #state{socket = Socket}) ->
Body = jiffy:encode(#{<<"fields">> => Fields, <<"tags">> => Tags}, [force_utf8]),
handle_cast({send_metric_data, Measurement, Tags, Fields}, State = #state{socket = Socket}) ->
%% 基于Line Protocol实现数据的传输
Point = efka_point:new(Measurement, Tags, Fields, efka_util:timestamp()),
Body = efka_point:normalized(Point),
Packet = <<0:32, ?PACKET_METRIC_DATA, Body/binary>>,
ok = gen_tcp:send(Socket, Packet),
@ -326,7 +329,7 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
%% 采用32位编码
-spec next_packet_id(PacketId :: integer()) -> NextPacketId :: integer().
next_packet_id(PacketId) when PacketId >= 4294967295 ->
0;
1;
next_packet_id(PacketId) ->
PacketId + 1.

49
apps/efka/src/client/efka_point.erl Normal file
View File

@ -0,0 +1,49 @@
%%%-------------------------------------------------------------------
%%% @author aresei
%%% @copyright (C) 2023, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 30. 5月 2023 11:28
%%%-------------------------------------------------------------------
-module(efka_point).
-author("aresei").
-record(point, {
measurement,
tags = [],
fields = [],
time = 0 :: integer()
}).
%% API
-export([new/4, normalized/1]).
-spec new(Measurement :: binary(), Tags :: map(), Fields :: map(), Timestamp :: integer()) -> #point{}.
new(Measurement, Tags, Fields, Timestamp) when is_binary(Measurement), is_map(Tags), is_map(Fields), is_integer(Timestamp) ->
#point{measurement = Measurement, tags = Tags, fields = Fields, time = Timestamp}.
-spec normalized(Point :: #point{}) -> binary().
normalized(#point{measurement = Measurement, tags = Tags, fields = Fields, time = Time}) ->
NTags = lists:map(fun({N, V}) -> <<N/binary, $=, V/binary>> end, maps:to_list(Tags)),
NFields = lists:map(fun({K, V}) -> <<K/binary, $=, (field_val(V))/binary>> end, maps:to_list(Fields)),
TagItems = lists:join(<<",">>, [Measurement | NTags]),
FieldItems = lists:join(<<",">>, NFields),
erlang:iolist_to_binary([TagItems, <<" ">>, FieldItems, <<" ">>, integer_to_binary(Time)]).
field_val(V) when is_integer(V) ->
integer_to_binary(V);
field_val(V) when is_float(V) ->
%% 默认按照浮点数表示
S = float_to_list(V, [{decimals, 6}, compact]),
list_to_binary(S);
field_val(V) when is_binary(V) ->
<<$", V/binary, $">>;
field_val(true) ->
<<"true">>;
field_val(false) ->
<<"false">>;
field_val(_Other) ->
erlang:error(unsupported_type).

13
apps/efka/src/efka_agent.erl
View File

@ -16,7 +16,7 @@
%% API
-export([start_link/0]).
-export([metric_data/5, event/3, ai_event/3, ping/13]).
-export([metric_data/2, event/3, ai_event/3, ping/13]).
-export([feedback_phase/4, feedback_phase/5]).
%% gen_server callbacks
@ -43,8 +43,8 @@
%%%===================================================================
%% 发送数据
metric_data(ServiceId, DeviceUUID, Timestamp, Tags, Fields) when is_binary(ServiceId), is_binary(DeviceUUID), is_integer(Timestamp), is_map(Tags), is_binary(Fields) ->
gen_server:cast(?SERVER, {metric_data, ServiceId, DeviceUUID, Timestamp, Tags, Fields}).
metric_data(ServiceId, LineProtocolData) when is_binary(ServiceId), is_binary(LineProtocolData) ->
gen_server:cast(?SERVER, {metric_data, ServiceId, LineProtocolData}).
ping(AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces) ->
gen_server:cast(?SERVER, {ping, AdCode, BootTime, Province, City, EfkaVersion, KernelArch, Ips, CpuCore, CpuLoad, CpuTemperature, Disk, Memory, Interfaces}).
@ -104,13 +104,10 @@ handle_call(_Request, _From, State = #state{}) ->
{stop, Reason :: term(), NewState :: #state{}}).
%% 发送数据
handle_cast({metric_data, ServiceId, DeviceUUID, Timestamp, Tags, Fields}, State = #state{status = Status, transport_pid = TransportPid}) ->
handle_cast({metric_data, ServiceId, LineProtocolData}, State = #state{status = Status, transport_pid = TransportPid}) ->
Packet = message_pb:encode_msg(#data{
service_id = ServiceId,
device_uuid = DeviceUUID,
timestamp = Timestamp,
tags = Tags,
fields = Fields
metric = LineProtocolData
}),
case Status =:= ?STATE_ACTIVATED of

10
apps/efka/src/efka_micro_service.erl
View File

@ -21,7 +21,7 @@
-export([start_link/2]).
-export([get_name/1, get_pid/1, start_service/1, stop_service/1, attach_channel/2]).
-export([push_params/2, push_metrics/2, request_params/1, request_metrics/1]).
-export([metric_data/4, send_event/3, send_ai_event/3]).
-export([metric_data/2, send_event/3, send_ai_event/3]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@ -77,8 +77,8 @@ request_params(Pid) when is_pid(Pid) ->
request_metrics(Pid) when is_pid(Pid) ->
gen_server:call(Pid, request_metrics).
metric_data(Pid, DeviceUUID, Tags, Fields) ->
gen_server:cast(Pid, {DeviceUUID, Tags, Fields}).
metric_data(Pid, Data) when is_pid(Pid), is_binary(Data) ->
gen_server:cast(Pid, {metric_data, Data}).
send_event(Pid, EventType, Params) ->
gen_server:cast(Pid, {send_event, EventType, Params}).
@ -211,8 +211,8 @@ handle_call(_Request, _From, State = #state{}) ->
{noreply, NewState :: #state{}} |
{noreply, NewState :: #state{}, timeout() | hibernate} |
{stop, Reason :: term(), NewState :: #state{}}).
handle_cast({metric_data, DeviceUUID, Tags, Fields}, State = #state{service = #micro_service{service_id = ServiceId}}) ->
efka_agent:metric_data(ServiceId, DeviceUUID, efka_util:timestamp(), Tags, Fields),
handle_cast({metric_data, LineProtocolData}, State = #state{service = #micro_service{service_id = ServiceId}}) ->
efka_agent:metric_data(ServiceId, LineProtocolData),
{noreply, State};
handle_cast({send_event, EventType, Params}, State = #state{service = #micro_service{service_id = ServiceId}}) ->

7
apps/efka/src/efka_tcp_channel.erl
View File

@ -159,12 +159,7 @@ handle_info({tcp, Socket, <<PacketId:32, ?PACKET_REQUEST_METRIC:8>>}, State = #s
%% 数据项
handle_info({tcp, Socket, <<0:32, ?PACKET_METRIC_DATA:8, Data/binary>>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
case catch jiffy:decode(Data, [return_maps]) of
#{<<"fields">> := Fields, <<"tag">> := Tags} ->
efka_micro_service:metric_data(ServicePid, Fields, Tags);
_ ->
ok
end,
efka_micro_service:metric_data(ServicePid, Data),
{noreply, State};
%% 远程日志

282
apps/efka/src/proto/message_pb.erl
View File

@ -88,7 +88,6 @@
-type '$msg'() :: auth_request() | auth_reply() | activate_push() | deploy() | efka_response() | topic_message() | service_params() | service_metrics() | data() | ping() | service_inform() | feedback_phase() | event() | ai_event().
-export_type(['$msg_name'/0, '$msg'/0]).
-record('map<string,string>',{key, value}).
-if(?OTP_RELEASE >= 24).
-dialyzer({no_underspecs, encode_msg/1}).
-endif.
@ -378,7 +377,7 @@ encode_msg_service_metrics(#service_metrics{service_id = F1, metrics = F2}, Bin,
encode_msg_data(Msg, TrUserData) -> encode_msg_data(Msg, <<>>, TrUserData).
encode_msg_data(#data{service_id = F1, device_uuid = F2, timestamp = F3, tags = F4, fields = F5}, Bin, TrUserData) ->
encode_msg_data(#data{service_id = F1, metric = F2}, Bin, TrUserData) ->
B1 = if F1 == undefined -> Bin;
true ->
begin
@ -389,38 +388,13 @@ encode_msg_data(#data{service_id = F1, device_uuid = F2, timestamp = F3, tags =
end
end
end,
B2 = if F2 == undefined -> B1;
true ->
begin
TrF2 = id(F2, TrUserData),
case is_empty_string(TrF2) of
true -> B1;
false -> e_type_string(TrF2, <<B1/binary, 18>>, TrUserData)
end
end
end,
B3 = if F3 == undefined -> B2;
true ->
begin
TrF3 = id(F3, TrUserData),
if TrF3 =:= 0 -> B2;
true -> e_type_int32(TrF3, <<B2/binary, 24>>, TrUserData)
end
end
end,
B4 = begin
TrF4 = id(F4, TrUserData),
if TrF4 == [] -> B3;
true -> e_field_data_tags(TrF4, B3, TrUserData)
end
end,
if F5 == undefined -> B4;
if F2 == undefined -> B1;
true ->
begin
TrF5 = id(F5, TrUserData),
case is_empty_string(TrF5) of
true -> B4;
false -> e_type_string(TrF5, <<B4/binary, 42>>, TrUserData)
TrF2 = id(F2, TrUserData),
case is_empty_string(TrF2) of
true -> B1;
false -> e_type_string(TrF2, <<B1/binary, 18>>, TrUserData)
end
end
end.
@ -708,17 +682,6 @@ encode_msg_ai_event(#ai_event{service_id = F1, event_type = F2, params = F3}, Bi
end
end.
e_mfield_data_tags(Msg, Bin, TrUserData) ->
SubBin = 'encode_msg_map<string,string>'(Msg, <<>>, TrUserData),
Bin2 = e_varint(byte_size(SubBin), Bin),
<<Bin2/binary, SubBin/binary>>.
e_field_data_tags([Elem | Rest], Bin, TrUserData) ->
Bin2 = <<Bin/binary, 34>>,
Bin3 = e_mfield_data_tags('tr_encode_data.tags[x]'(Elem, TrUserData), Bin2, TrUserData),
e_field_data_tags(Rest, Bin3, TrUserData);
e_field_data_tags([], Bin, _TrUserData) -> Bin.
e_field_ping_ips([Elem | Rest], Bin, TrUserData) ->
Bin2 = <<Bin/binary, 58>>,
Bin3 = e_type_string(id(Elem, TrUserData), Bin2, TrUserData),
@ -749,10 +712,6 @@ e_pfield_ping_memory([Value | Rest], Bin, TrUserData) ->
e_pfield_ping_memory(Rest, Bin2, TrUserData);
e_pfield_ping_memory([], Bin, _TrUserData) -> Bin.
'encode_msg_map<string,string>'(#'map<string,string>'{key = F1, value = F2}, Bin, TrUserData) ->
B1 = begin TrF1 = id(F1, TrUserData), e_type_string(TrF1, <<Bin/binary, 10>>, TrUserData) end,
begin TrF2 = id(F2, TrUserData), e_type_string(TrF2, <<B1/binary, 18>>, TrUserData) end.
-compile({nowarn_unused_function,e_type_sint/3}).
e_type_sint(Value, Bin, _TrUserData) when Value >= 0 -> e_varint(Value * 2, Bin);
e_type_sint(Value, Bin, _TrUserData) -> e_varint(Value * -2 - 1, Bin).
@ -1351,77 +1310,56 @@ skip_32_service_metrics(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData
skip_64_service_metrics(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
decode_msg_data(Bin, TrUserData) -> dfp_read_field_def_data(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), id(0, TrUserData), 'tr_decode_init_default_data.tags'([], TrUserData), id(<<>>, TrUserData), TrUserData).
decode_msg_data(Bin, TrUserData) -> dfp_read_field_def_data(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData).
dfp_read_field_def_data(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> d_field_data_service_id(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
dfp_read_field_def_data(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> d_field_data_device_uuid(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
dfp_read_field_def_data(<<24, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> d_field_data_timestamp(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
dfp_read_field_def_data(<<34, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> d_field_data_tags(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
dfp_read_field_def_data(<<42, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> d_field_data_fields(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
dfp_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, F@_3, R1, F@_5, TrUserData) -> #data{service_id = F@_1, device_uuid = F@_2, timestamp = F@_3, tags = 'tr_decode_repeated_finalize_data.tags'(R1, TrUserData), fields = F@_5};
dfp_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> dg_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData).
dfp_read_field_def_data(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_data_service_id(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
dfp_read_field_def_data(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_data_metric(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
dfp_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, _) -> #data{service_id = F@_1, metric = F@_2};
dfp_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dg_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, TrUserData).
dg_read_field_def_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) when N < 32 - 7 -> dg_read_field_def_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
dg_read_field_def_data(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) ->
dg_read_field_def_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 32 - 7 -> dg_read_field_def_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
dg_read_field_def_data(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, TrUserData) ->
Key = X bsl N + Acc,
case Key of
10 -> d_field_data_service_id(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
18 -> d_field_data_device_uuid(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
24 -> d_field_data_timestamp(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
34 -> d_field_data_tags(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
42 -> d_field_data_fields(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
10 -> d_field_data_service_id(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
18 -> d_field_data_metric(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
_ ->
case Key band 7 of
0 -> skip_varint_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
1 -> skip_64_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
2 -> skip_length_delimited_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
3 -> skip_group_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
5 -> skip_32_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData)
0 -> skip_varint_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
1 -> skip_64_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
2 -> skip_length_delimited_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
3 -> skip_group_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
5 -> skip_32_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData)
end
end;
dg_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, F@_3, R1, F@_5, TrUserData) -> #data{service_id = F@_1, device_uuid = F@_2, timestamp = F@_3, tags = 'tr_decode_repeated_finalize_data.tags'(R1, TrUserData), fields = F@_5}.
dg_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, _) -> #data{service_id = F@_1, metric = F@_2}.
d_field_data_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) when N < 57 -> d_field_data_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
d_field_data_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, F@_3, F@_4, F@_5, TrUserData) ->
d_field_data_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_data_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
d_field_data_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
{NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
dfp_read_field_def_data(RestF, 0, 0, F, NewFValue, F@_2, F@_3, F@_4, F@_5, TrUserData).
dfp_read_field_def_data(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
d_field_data_device_uuid(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) when N < 57 -> d_field_data_device_uuid(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
d_field_data_device_uuid(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, F@_3, F@_4, F@_5, TrUserData) ->
d_field_data_metric(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_data_metric(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
d_field_data_metric(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, TrUserData) ->
{NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
dfp_read_field_def_data(RestF, 0, 0, F, F@_1, NewFValue, F@_3, F@_4, F@_5, TrUserData).
dfp_read_field_def_data(RestF, 0, 0, F, F@_1, NewFValue, TrUserData).
d_field_data_timestamp(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) when N < 57 -> d_field_data_timestamp(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
d_field_data_timestamp(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, _, F@_4, F@_5, TrUserData) ->
{NewFValue, RestF} = {begin <<Res:32/signed-native>> = <<(X bsl N + Acc):32/unsigned-native>>, id(Res, TrUserData) end, Rest},
dfp_read_field_def_data(RestF, 0, 0, F, F@_1, F@_2, NewFValue, F@_4, F@_5, TrUserData).
skip_varint_data(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> skip_varint_data(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
skip_varint_data(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
d_field_data_tags(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) when N < 57 -> d_field_data_tags(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
d_field_data_tags(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, Prev, F@_5, TrUserData) ->
{NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bs:Len/binary, Rest2/binary>> = Rest, {id('decode_msg_map<string,string>'(Bs, TrUserData), TrUserData), Rest2} end,
dfp_read_field_def_data(RestF, 0, 0, F, F@_1, F@_2, F@_3, 'tr_decode_repeated_add_elem_data.tags'(NewFValue, Prev, TrUserData), F@_5, TrUserData).
d_field_data_fields(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) when N < 57 -> d_field_data_fields(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
d_field_data_fields(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, _, TrUserData) ->
{NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
dfp_read_field_def_data(RestF, 0, 0, F, F@_1, F@_2, F@_3, F@_4, NewFValue, TrUserData).
skip_varint_data(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> skip_varint_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
skip_varint_data(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData).
skip_length_delimited_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) when N < 57 -> skip_length_delimited_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData);
skip_length_delimited_data(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) ->
skip_length_delimited_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> skip_length_delimited_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
skip_length_delimited_data(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) ->
Length = X bsl N + Acc,
<<_:Length/binary, Rest2/binary>> = Rest,
dfp_read_field_def_data(Rest2, 0, 0, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData).
dfp_read_field_def_data(Rest2, 0, 0, F, F@_1, F@_2, TrUserData).
skip_group_data(Bin, _, Z2, FNum, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) ->
skip_group_data(Bin, _, Z2, FNum, F@_1, F@_2, TrUserData) ->
{_, Rest} = read_group(Bin, FNum),
dfp_read_field_def_data(Rest, 0, Z2, FNum, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData).
dfp_read_field_def_data(Rest, 0, Z2, FNum, F@_1, F@_2, TrUserData).
skip_32_data(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData).
skip_32_data(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
skip_64_data(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, F@_5, TrUserData).
skip_64_data(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
decode_msg_ping(Bin, TrUserData) ->
dfp_read_field_def_ping(Bin,
@ -1896,57 +1834,6 @@ skip_32_ai_event(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData)
skip_64_ai_event(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_ai_event(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData).
'decode_msg_map<string,string>'(Bin, TrUserData) -> 'dfp_read_field_def_map<string,string>'(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData).
'dfp_read_field_def_map<string,string>'(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> 'd_field_map<string,string>_key'(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
'dfp_read_field_def_map<string,string>'(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> 'd_field_map<string,string>_value'(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
'dfp_read_field_def_map<string,string>'(<<>>, 0, 0, _, F@_1, F@_2, _) -> #'map<string,string>'{key = F@_1, value = F@_2};
'dfp_read_field_def_map<string,string>'(Other, Z1, Z2, F, F@_1, F@_2, TrUserData) -> 'dg_read_field_def_map<string,string>'(Other, Z1, Z2, F, F@_1, F@_2, TrUserData).
'dg_read_field_def_map<string,string>'(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 32 - 7 -> 'dg_read_field_def_map<string,string>'(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
'dg_read_field_def_map<string,string>'(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, TrUserData) ->
Key = X bsl N + Acc,
case Key of
10 -> 'd_field_map<string,string>_key'(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
18 -> 'd_field_map<string,string>_value'(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
_ ->
case Key band 7 of
0 -> 'skip_varint_map<string,string>'(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
1 -> 'skip_64_map<string,string>'(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
2 -> 'skip_length_delimited_map<string,string>'(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
3 -> 'skip_group_map<string,string>'(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
5 -> 'skip_32_map<string,string>'(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData)
end
end;
'dg_read_field_def_map<string,string>'(<<>>, 0, 0, _, F@_1, F@_2, _) -> #'map<string,string>'{key = F@_1, value = F@_2}.
'd_field_map<string,string>_key'(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> 'd_field_map<string,string>_key'(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
'd_field_map<string,string>_key'(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
{NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
'dfp_read_field_def_map<string,string>'(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
'd_field_map<string,string>_value'(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> 'd_field_map<string,string>_value'(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
'd_field_map<string,string>_value'(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, TrUserData) ->
{NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
'dfp_read_field_def_map<string,string>'(RestF, 0, 0, F, F@_1, NewFValue, TrUserData).
'skip_varint_map<string,string>'(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> 'skip_varint_map<string,string>'(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
'skip_varint_map<string,string>'(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> 'dfp_read_field_def_map<string,string>'(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
'skip_length_delimited_map<string,string>'(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> 'skip_length_delimited_map<string,string>'(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
'skip_length_delimited_map<string,string>'(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) ->
Length = X bsl N + Acc,
<<_:Length/binary, Rest2/binary>> = Rest,
'dfp_read_field_def_map<string,string>'(Rest2, 0, 0, F, F@_1, F@_2, TrUserData).
'skip_group_map<string,string>'(Bin, _, Z2, FNum, F@_1, F@_2, TrUserData) ->
{_, Rest} = read_group(Bin, FNum),
'dfp_read_field_def_map<string,string>'(Rest, 0, Z2, FNum, F@_1, F@_2, TrUserData).
'skip_32_map<string,string>'(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> 'dfp_read_field_def_map<string,string>'(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
'skip_64_map<string,string>'(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> 'dfp_read_field_def_map<string,string>'(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
read_group(Bin, FieldNum) ->
{NumBytes, EndTagLen} = read_gr_b(Bin, 0, 0, 0, 0, FieldNum),
<<Group:NumBytes/binary, _:EndTagLen/binary, Rest/binary>> = Bin,
@ -2138,28 +2025,14 @@ merge_msg_service_metrics(#service_metrics{service_id = PFservice_id, metrics =
end}.
-compile({nowarn_unused_function,merge_msg_data/3}).
merge_msg_data(#data{service_id = PFservice_id, device_uuid = PFdevice_uuid, timestamp = PFtimestamp, tags = PFtags, fields = PFfields},
#data{service_id = NFservice_id, device_uuid = NFdevice_uuid, timestamp = NFtimestamp, tags = NFtags, fields = NFfields}, TrUserData) ->
merge_msg_data(#data{service_id = PFservice_id, metric = PFmetric}, #data{service_id = NFservice_id, metric = NFmetric}, _) ->
#data{service_id =
if NFservice_id =:= undefined -> PFservice_id;
true -> NFservice_id
end,
device_uuid =
if NFdevice_uuid =:= undefined -> PFdevice_uuid;
true -> NFdevice_uuid
end,
timestamp =
if NFtimestamp =:= undefined -> PFtimestamp;
true -> NFtimestamp
end,
tags =
if PFtags /= undefined, NFtags /= undefined -> 'tr_merge_data.tags'(PFtags, NFtags, TrUserData);
PFtags == undefined -> NFtags;
NFtags == undefined -> PFtags
end,
fields =
if NFfields =:= undefined -> PFfields;
true -> NFfields
metric =
if NFmetric =:= undefined -> PFmetric;
true -> NFmetric
end}.
-compile({nowarn_unused_function,merge_msg_ping/3}).
@ -2439,19 +2312,12 @@ v_msg_service_metrics(X, Path, _TrUserData) -> mk_type_error({expected_msg, serv
-compile({nowarn_unused_function,v_msg_data/3}).
-dialyzer({nowarn_function,v_msg_data/3}).
v_msg_data(#data{service_id = F1, device_uuid = F2, timestamp = F3, tags = F4, fields = F5}, Path, TrUserData) ->
v_msg_data(#data{service_id = F1, metric = F2}, Path, TrUserData) ->
if F1 == undefined -> ok;
true -> v_type_string(F1, [service_id | Path], TrUserData)
end,
if F2 == undefined -> ok;
true -> v_type_string(F2, [device_uuid | Path], TrUserData)
end,
if F3 == undefined -> ok;
true -> v_type_int32(F3, [timestamp | Path], TrUserData)
end,
'v_map<string,string>'(F4, [tags | Path], TrUserData),
if F5 == undefined -> ok;
true -> v_type_string(F5, [fields | Path], TrUserData)
true -> v_type_string(F2, [metric | Path], TrUserData)
end,
ok;
v_msg_data(X, Path, _TrUserData) -> mk_type_error({expected_msg, data}, X, Path).
@ -2616,19 +2482,6 @@ v_type_string(S, Path, _TrUserData) when is_list(S); is_binary(S) ->
end;
v_type_string(X, Path, _TrUserData) -> mk_type_error(bad_unicode_string, X, Path).
-compile({nowarn_unused_function,'v_map<string,string>'/3}).
-dialyzer({nowarn_function,'v_map<string,string>'/3}).
'v_map<string,string>'(KVs, Path, TrUserData) when is_list(KVs) ->
[case X of
{Key, Value} ->
v_type_string(Key, [key | Path], TrUserData),
v_type_string(Value, [value | Path], TrUserData);
_ -> mk_type_error(invalid_key_value_tuple, X, Path)
end
|| X <- KVs],
ok;
'v_map<string,string>'(X, Path, _TrUserData) -> mk_type_error(invalid_list_of_key_value_tuples, X, Path).
-compile({nowarn_unused_function,mk_type_error/3}).
-spec mk_type_error(_, _, list()) -> no_return().
mk_type_error(Error, ValueSeen, Path) ->
@ -2664,43 +2517,6 @@ cons(Elem, Acc, _TrUserData) -> [Elem | Acc].
-compile({nowarn_unused_function,'erlang_++'/3}).
-compile({inline,'erlang_++'/3}).
'erlang_++'(A, B, _TrUserData) -> A ++ B.
-compile({inline,'tr_decode_init_default_data.tags'/2}).
'tr_decode_init_default_data.tags'(_, _) -> mt_empty_map_r().
-compile({inline,'tr_merge_data.tags'/3}).
'tr_merge_data.tags'(X1, X2, _) -> mt_merge_maptuples_r(X1, X2).
-compile({inline,'tr_decode_repeated_finalize_data.tags'/2}).
'tr_decode_repeated_finalize_data.tags'(L, _) -> mt_finalize_items_r(L).
-compile({inline,'tr_decode_repeated_add_elem_data.tags'/3}).
'tr_decode_repeated_add_elem_data.tags'(Elem, L, _) -> mt_add_item_r(Elem, L).
-compile({inline,'tr_encode_data.tags[x]'/2}).
'tr_encode_data.tags[x]'(X, _) -> mt_maptuple_to_pseudomsg_r(X, 'map<string,string>').
-compile({inline,mt_maptuple_to_pseudomsg_r/2}).
mt_maptuple_to_pseudomsg_r({K, V}, RName) -> {RName, K, V}.
-compile({inline,mt_empty_map_r/0}).
mt_empty_map_r() -> [].
-compile({inline,mt_add_item_r/2}).
mt_add_item_r({_RName, K, V}, Acc) -> [{K, V} | Acc].
-compile({inline,mt_finalize_items_r/1}).
mt_finalize_items_r(Acc) -> mt_finalize_items_r_aux(lists:reverse(Acc), dict:new()).
mt_finalize_items_r_aux([{K, V} | Tl], D) -> mt_finalize_items_r_aux(Tl, dict:store(K, V, D));
mt_finalize_items_r_aux([], D) -> dict:to_list(D).
-compile({inline,mt_merge_maptuples_r/2}).
mt_merge_maptuples_r(L1, L2) -> dict:to_list(dict:merge(fun (_Key, _V1, V2) -> V2 end, dict:from_list(L1), dict:from_list(L2))).
get_msg_defs() ->
@ -2726,12 +2542,7 @@ get_msg_defs() ->
{{msg, topic_message}, [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
{{msg, service_params}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
{{msg, service_metrics}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metrics, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
{{msg, data},
[#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
#field{name = device_uuid, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
#field{name = timestamp, fnum = 3, rnum = 4, type = int32, occurrence = optional, opts = []},
#field{name = tags, fnum = 4, rnum = 5, type = {map, string, string}, occurrence = repeated, opts = []},
#field{name = fields, fnum = 5, rnum = 6, type = string, occurrence = optional, opts = []}]},
{{msg, data}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
{{msg, ping},
[#field{name = adcode, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
#field{name = boot_time, fnum = 2, rnum = 3, type = uint32, occurrence = optional, opts = []},
@ -2812,12 +2623,7 @@ find_msg_def(efka_response) ->
find_msg_def(topic_message) -> [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
find_msg_def(service_params) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
find_msg_def(service_metrics) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metrics, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
find_msg_def(data) ->
[#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
#field{name = device_uuid, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
#field{name = timestamp, fnum = 3, rnum = 4, type = int32, occurrence = optional, opts = []},
#field{name = tags, fnum = 4, rnum = 5, type = {map, string, string}, occurrence = repeated, opts = []},
#field{name = fields, fnum = 5, rnum = 6, type = string, occurrence = optional, opts = []}];
find_msg_def(data) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
find_msg_def(ping) ->
[#field{name = adcode, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
#field{name = boot_time, fnum = 2, rnum = 3, type = uint32, occurrence = optional, opts = []},

7
message_pb.proto
View File

@ -56,11 +56,8 @@ message ServiceMetrics {
// 数据传输
message Data {
string service_id = 1;
string device_uuid = 2;
int32 timestamp = 3;
map<string, string> tags = 4;
// 为了数据的灵活性，存储类型无关的二进制流; 兼容更多的业务逻辑情况
string fields = 5;
// measurement[,tag_key=tag_value...] field_key=field_value[,field_key2=field_value2...] [timestamp]
string metric = 2;
}
//#{<<"adcode">> => 0,<<"boot_time">> => 18256077,<<"city">> => <<>>,