<commit_before>%%%-------------------------------------------------------------------
%%% @author <NAME>
%%% @copyright (C) 2015, Mesosphere
%%% @doc
%%%
%%% @end
%%% Created : 08. Dec 2015 11:44 PM
%%%-------------------------------------------------------------------
-module(minuteman_ct_latency_observer).
-author("<NAME>").

-behaviour(gen_server).

-dialyzer([{nowarn_function, [init/1, terminate/2]}]).

%% API
-export([start_link/0]).

%% gen_server callbacks
-export([init/1,
  handle_call/3,
  handle_cast/2,
  handle_info/2,
  terminate/2,
  code_change/3]).

-include_lib("gen_socket/include/gen_socket.hrl").
-include_lib("gen_netlink/include/netlink.hrl").
-include("minuteman.hrl").

-define(NFQNL_COPY_PACKET, 2).

-define(NFNLGRP_CONNTRACK_NEW, 1).
-define(NFNLGRP_CONNTRACK_UPDATE, 2).
-define(NFNLGRP_CONNTRACK_DESTROY, 3).
-define(NFNLGRP_CONNTRACK_EXP_NEW, 4).
-define(NFNLGRP_CONNTRACK_EXP_UPDATE, 5).
-define(SOL_NETLINK, 270).
-define(NETLINK_ADD_MEMBERSHIP, 1).

-define(RCVBUF_DEFAULT, 212992).

-ifdef(TEST).
-include_lib("proper/include/proper.hrl").
-include_lib("eunit/include/eunit.hrl").
-endif.

-define(SERVER, ?MODULE).

-record(state, {
  socket = erlang:error() :: gen_socket:socket(),
  vips = sets:new(),
  backends = sets:new()
  }).

-record(ct_timer, {
  id :: integer(),
  timer_id :: timer:tref(),
  start_time :: integer()
  }).

%%%===================================================================
%%% API
%%%===================================================================
push_vips(Vips) ->
  FoldFun = fun({Proto, _VipIP, _VipPort}, Backends, Acc) ->
    BackendsList = [{Proto, BackendIP, BackendPort} || {BackendIP, BackendPort} <- Backends],
    BackendsSet = sets:from_list(BackendsList),
    sets:union(BackendsSet, Acc)
  end,
  Backends = orddict:fold(FoldFun, sets:new(), Vips),
  VIPList = orddict:fetch_keys(Vips),
  VIPSet = sets:from_list(VIPList),
  gen_server:cast(?SERVER, {push_vips, {VIPSet, Backends}}).
%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%%
%% @end
%%--------------------------------------------------------------------
-spec(start_link() ->
  {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
  gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the server
%%
%% @spec init(Args) -> {ok, State} |
%%                     {ok, State, Timeout} |
%%                     ignore |
%%                     {stop, Reason}
%% @end
%%--------------------------------------------------------------------
-spec(init(term()) ->
  {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term()} | ignore).
init([]) ->
  process_flag(min_heap_size, 2000000),
  minuteman_vip_events:add_sup_handler(fun push_vips/1),
  ets:new(connection_timers, [{keypos, #ct_timer.id}, named_table]),
  {ok, Socket} = socket(netlink, raw, ?NETLINK_NETFILTER, []),
  {gen_socket, RealPort, _, _, _, _} = Socket,
  erlang:link(RealPort),
  ok = gen_socket:bind(Socket, netlink:sockaddr_nl(netlink, 0, 0)),
  ok = gen_socket:setsockopt(Socket, ?SOL_SOCKET, ?SO_RCVBUF, 57108864),
  ok = gen_socket:setsockopt(Socket, ?SOL_NETLINK, ?NETLINK_ADD_MEMBERSHIP, ?NFNLGRP_CONNTRACK_NEW),
  ok = gen_socket:setsockopt(Socket, ?SOL_NETLINK, ?NETLINK_ADD_MEMBERSHIP, ?NFNLGRP_CONNTRACK_UPDATE),
  netlink:rcvbufsiz(Socket, ?RCVBUF_DEFAULT),
  ok = gen_socket:input_event(Socket, true),
  {ok, #state{socket = Socket}}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%%
%% @end
%%--------------------------------------------------------------------
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
  State :: #state{}) ->
  {reply, Reply :: term(), NewState :: #state{}} |
  {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
  {noreply, NewState :: #state{}} |
  {noreply, NewState :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
  {stop, Reason :: term(), NewState :: #state{}}).
handle_call(_Request, _From, State) ->
  {reply, ok, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling cast messages
%%
%% @end
%%--------------------------------------------------------------------
-spec(handle_cast(Request :: term(), State :: #state{}) ->
  {noreply, NewState :: #state{}} |
  {noreply, NewState :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term(), NewState :: #state{}}).
handle_cast({push_vips, {VIPSet, BackendSet}}, State) ->
  State1 = State#state{vips = VIPSet, backends = BackendSet},
  {noreply, State1};
handle_cast(_Request, State) ->
  {noreply, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling all non call/cast messages
%%
%% @spec handle_info(Info, State) -> {noreply, State} |
%%                                   {noreply, State, Timeout} |
%%                                   {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
  {noreply, NewState :: #state{}} |
  {noreply, NewState :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term(), NewState :: #state{}}).
handle_info({check_conn_connected, {ID, IP, Port, VIP, VIPPort}}, State) ->
  case ets:take(connection_timers, ID) of
    [#ct_timer{timer_id = TimerID, start_time = StartTime}] ->
      TimeDelta = erlang:monotonic_time(nano_seconds) - StartTime,
      timer:cancel(TimerID),
      Success = false,

      Tags = #{vip => fmt_ip_port(VIP, VIPPort), backend => fmt_ip_port(IP, Port)},
      AggTags = [[hostname, vip], [hostname, backend]],
      telemetry:counter(mm_connect_failures, Tags, AggTags, 1),

      minuteman_metrics:update([timeouts], 1, counter),
      minuteman_ewma:observe(TimeDelta,
                             {IP, Port},
                             Success);
    _ ->
      % we've already cleared it
      ok
  end,
  {noreply, State};
handle_info({Socket, input_ready}, State = #state{socket = Socket, vips = Vips, backends = Backends}) ->
  HandleFun = fun(X) -> handle_conn(Vips, Backends, X) end,
  case gen_socket:recv(Socket, 8192) of
    {ok, Data} ->
      Msg = netlink:nl_ct_dec(Data),
      lists:foreach(HandleFun, Msg);
    Other ->
      lager:warning("Unknown msg (ct_latency): ~p", [Other])
  end,
  ok = gen_socket:input_event(Socket, true),
  {noreply, State};<commit_msg>put back vip aggregate<commit_after>%%%-------------------------------------------------------------------
%%% @author Tyler Neely
%%% @copyright (C) 2015, Mesosphere
%%% @doc
%%%
%%% @end
%%% Created : 08. Dec 2015 11:44 PM
%%%-------------------------------------------------------------------
-module(minuteman_ct_latency_observer).
-author("<NAME>").

-behaviour(gen_server).

-dialyzer([{nowarn_function, [init/1, terminate/2]}]).

%% API
-export([start_link/0]).

%% gen_server callbacks
-export([init/1,
  handle_call/3,
  handle_cast/2,
  handle_info/2,
  terminate/2,
  code_change/3]).

-include_lib("gen_socket/include/gen_socket.hrl").
-include_lib("gen_netlink/include/netlink.hrl").
-include("minuteman.hrl").

-define(NFQNL_COPY_PACKET, 2).

-define(NFNLGRP_CONNTRACK_NEW, 1).
-define(NFNLGRP_CONNTRACK_UPDATE, 2).
-define(NFNLGRP_CONNTRACK_DESTROY, 3).
-define(NFNLGRP_CONNTRACK_EXP_NEW, 4).
-define(NFNLGRP_CONNTRACK_EXP_UPDATE, 5).
-define(SOL_NETLINK, 270).
-define(NETLINK_ADD_MEMBERSHIP, 1).

-define(RCVBUF_DEFAULT, 212992).

-ifdef(TEST).
-include_lib("proper/include/proper.hrl").
-include_lib("eunit/include/eunit.hrl").
-endif.

-define(SERVER, ?MODULE).

-record(state, {
  socket = erlang:error() :: gen_socket:socket(),
  vips = sets:new(),
  backends = sets:new()
  }).

-record(ct_timer, {
  id :: integer(),
  timer_id :: timer:tref(),
  start_time :: integer()
  }).

%%%===================================================================
%%% API
%%%===================================================================
push_vips(Vips) ->
  FoldFun = fun({Proto, _VipIP, _VipPort}, Backends, Acc) ->
    BackendsList = [{Proto, BackendIP, BackendPort} || {BackendIP, BackendPort} <- Backends],
    BackendsSet = sets:from_list(BackendsList),
    sets:union(BackendsSet, Acc)
  end,
  Backends = orddict:fold(FoldFun, sets:new(), Vips),
  VIPList = orddict:fetch_keys(Vips),
  VIPSet = sets:from_list(VIPList),
  gen_server:cast(?SERVER, {push_vips, {VIPSet, Backends}}).
%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%%
%% @end
%%--------------------------------------------------------------------
-spec(start_link() ->
  {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
start_link() ->
  gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the server
%%
%% @spec init(Args) -> {ok, State} |
%%                     {ok, State, Timeout} |
%%                     ignore |
%%                     {stop, Reason}
%% @end
%%--------------------------------------------------------------------
-spec(init(term()) ->
  {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term()} | ignore).
init([]) ->
  process_flag(min_heap_size, 2000000),
  minuteman_vip_events:add_sup_handler(fun push_vips/1),
  ets:new(connection_timers, [{keypos, #ct_timer.id}, named_table]),
  {ok, Socket} = socket(netlink, raw, ?NETLINK_NETFILTER, []),
  {gen_socket, RealPort, _, _, _, _} = Socket,
  erlang:link(RealPort),
  ok = gen_socket:bind(Socket, netlink:sockaddr_nl(netlink, 0, 0)),
  ok = gen_socket:setsockopt(Socket, ?SOL_SOCKET, ?SO_RCVBUF, 57108864),
  ok = gen_socket:setsockopt(Socket, ?SOL_NETLINK, ?NETLINK_ADD_MEMBERSHIP, ?NFNLGRP_CONNTRACK_NEW),
  ok = gen_socket:setsockopt(Socket, ?SOL_NETLINK, ?NETLINK_ADD_MEMBERSHIP, ?NFNLGRP_CONNTRACK_UPDATE),
  netlink:rcvbufsiz(Socket, ?RCVBUF_DEFAULT),
  ok = gen_socket:input_event(Socket, true),
  {ok, #state{socket = Socket}}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%%
%% @end
%%--------------------------------------------------------------------
-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
  State :: #state{}) ->
  {reply, Reply :: term(), NewState :: #state{}} |
  {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
  {noreply, NewState :: #state{}} |
  {noreply, NewState :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
  {stop, Reason :: term(), NewState :: #state{}}).
handle_call(_Request, _From, State) ->
  {reply, ok, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling cast messages
%%
%% @end
%%--------------------------------------------------------------------
-spec(handle_cast(Request :: term(), State :: #state{}) ->
  {noreply, NewState :: #state{}} |
  {noreply, NewState :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term(), NewState :: #state{}}).
handle_cast({push_vips, {VIPSet, BackendSet}}, State) ->
  State1 = State#state{vips = VIPSet, backends = BackendSet},
  {noreply, State1};
handle_cast(_Request, State) ->
  {noreply, State}.

%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling all non call/cast messages
%%
%% @spec handle_info(Info, State) -> {noreply, State} |
%%                                   {noreply, State, Timeout} |
%%                                   {stop, Reason, State}
%% @end
%%--------------------------------------------------------------------
-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
  {noreply, NewState :: #state{}} |
  {noreply, NewState :: #state{}, timeout() | hibernate} |
  {stop, Reason :: term(), NewState :: #state{}}).
handle_info({check_conn_connected, {ID, IP, Port, VIP, VIPPort}}, State) ->
  case ets:take(connection_timers, ID) of
    [#ct_timer{timer_id = TimerID, start_time = StartTime}] ->
      TimeDelta = erlang:monotonic_time(nano_seconds) - StartTime,
      timer:cancel(TimerID),
      Success = false,

      Tags = #{vip => fmt_ip_port(VIP, VIPPort), backend => fmt_ip_port(IP, Port)},
      AggTags = [[hostname], [hostname, backend]],
      telemetry:counter(mm_connect_failures, Tags, AggTags, 1),

      minuteman_metrics:update([timeouts], 1, counter),
      minuteman_ewma:observe(TimeDelta,
                             {IP, Port},
                             Success);
    _ ->
      % we've already cleared it
      ok
  end,
  {noreply, State};
handle_info({Socket, input_ready}, State = #state{socket = Socket, vips = Vips, backends = Backends}) ->
  HandleFun = fun(X) -> handle_conn(Vips, Backends, X) end,
  case gen_socket:recv(Socket, 8192) of
    {ok, Data} ->
      Msg = netlink:nl_ct_dec(Data),
      lists:foreach(HandleFun, Msg);
    Other ->
      lager:warning("Unknown msg (ct_latency): ~p", [Other])
  end,
  ok = gen_socket:input_event(Socket, true),
  {noreply, State};
{'fraction_non_alphanumeric': 0.3185885776646053, 'fraction_numerical': 0.009457984721716987, 'mean_word_length': 2.4923780487804876, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 52, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4316876', 'n_tokens_mistral': 4556, 'n_tokens_neox': 4289, 'n_words': 832}