<commit_before>%% @author Couchbase, Inc <<EMAIL>>
%% @copyright 2015 Couchbase, Inc.
%%
%% 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.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
-module(index_stats_collector).

-include_lib("eunit/include/eunit.hrl").

-include("ns_common.hrl").

-include("ns_stats.hrl").

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

%% callbacks
-export([init/1, handle_info/2, grab_stats/1, process_stats/5]).

-record(state, {indexer :: atom(),
                default_stats,
                buckets}).

server_name(Indexer) ->
    list_to_atom(?MODULE_STRING "-" ++ atom_to_list(Indexer:get_type())).

ets_name(Indexer) ->
    list_to_atom(?MODULE_STRING "_names-" ++ atom_to_list(Indexer:get_type())).

start_link(Indexer) ->
    base_stats_collector:start_link({local, server_name(Indexer)}, ?MODULE, Indexer).

init(Indexer) ->
    ets:new(ets_name(Indexer), [protected, named_table]),

    Self = self(),
    ns_pubsub:subscribe_link(
      ns_config_events,
      fun ({buckets, Buckets}) ->
              BucketConfigs = proplists:get_value(configs, Buckets, []),
              BucketsList = ns_bucket:get_bucket_names_of_type(membase, couchstore, BucketConfigs) ++
                  ns_bucket:get_bucket_names_of_type(membase, ephemeral, BucketConfigs),
              Self ! {buckets, BucketsList};
          (_) ->
              ok
      end),

    Buckets = lists:map(fun list_to_binary/1,
                        ns_bucket:get_bucket_names_of_type(membase, couchstore) ++
                            ns_bucket:get_bucket_names_of_type(membase, ephemeral)),
    Defaults = [{Indexer:global_index_stat(atom_to_binary(Stat, latin1)), 0}
                || Stat <- Indexer:get_gauges() ++ Indexer:get_counters() ++ Indexer:get_computed()],


    {ok, #state{indexer = Indexer,
                buckets = Buckets,
                default_stats = finalize_index_stats(Defaults)}}.

find_type(_, []) ->
    not_found;
find_type(Name, [{Type, Metrics} | Rest]) ->
    MaybeMetric = [Name || M <- Metrics,
                           atom_to_binary(M, latin1) =:= Name],

    case MaybeMetric of
        [_] ->
            Type;
        _ ->
            find_type(Name, Rest)
    end.

do_recognize_name(_Indexer, <<"needs_restart">>) ->
    {status, index_needs_restart};
do_recognize_name(_Indexer, <<"num_connections">>) ->
    {status, index_num_connections};
do_recognize_name(Indexer, K) ->
    Type = find_type(K, [{service_gauge, Indexer:get_service_gauges()},
                         {service_counter, Indexer:get_service_counters()}]),
    case Type of
        not_found ->
            do_recognize_bucket_metric(Indexer, K);
        _ ->
            NewKey = list_to_binary(Indexer:service_stat_prefix() ++
                                        binary_to_list(K)),
            {Type, NewKey}
    end.

do_recognize_bucket_metric(Indexer, K) ->
    case binary:split(K, <<":">>, [global]) of
        [Bucket, Index, Metric] ->
            Type = find_type(Metric, [{gauge, Indexer:get_gauges()},
                                      {counter, Indexer:get_counters()}]),

            case Type of
                not_found ->
                    undefined;
                _ ->
                    {Type, {Bucket, Index, Metric}}
            end;
        _ ->
            undefined
    end.

recognize_name(Indexer, Ets, K) ->
    case ets:lookup(Ets, K) of
        [{K, Type, NewK}] ->
            {Type, NewK};
        [{K, undefined}] ->
            undefined;
        [] ->
            case do_recognize_name(Indexer, K) of
                undefined ->
                    ets:insert(Ets, {K, undefined}),
                    undefined;
                {Type, NewK} ->
                    ets:insert(Ets, {K, Type, NewK}),
                    {Type, NewK}
            end
    end.

massage_stats(Indexer, Ets, GrabbedStats) ->
    massage_stats(Indexer, Ets, GrabbedStats, [], [], [], [], []).

massage_stats(_Indexer, _Ets, [], AccGauges, AccCounters, AccStatus, AccSysGauges, AccSysCounters) ->
    {AccGauges, AccCounters, AccStatus, AccSysGauges, AccSysCounters};
massage_stats(Indexer, Ets, [{K, V} | Rest], AccGauges, AccCounters,
              AccStatus, AccSysGauges, AccSysCounters) ->
    case recognize_name(Indexer, Ets, K) of
        undefined ->
            massage_stats(Indexer, Ets, Rest, AccGauges, AccCounters, AccStatus,
                          AccSysGauges, AccSysCounters);
        {counter, NewK} ->
            massage_stats(Indexer, Ets, Rest, AccGauges,
                          [{NewK, V} | AccCounters], AccStatus, AccSysGauges, AccSysCounters);
        {gauge, NewK} ->
            massage_stats(Indexer, Ets, Rest, [{NewK, V} | AccGauges],
                          AccCounters, AccStatus, AccSysGauges, AccSysCounters);
        {service_gauge, NewK} ->
            massage_stats(Indexer, Ets, Rest, AccGauges, AccCounters,
                          AccStatus, [{NewK, V} | AccSysGauges], AccSysCounters);
        {service_counter, NewK} ->
            massage_stats(Indexer, Ets, Rest, AccGauges, AccCounters,
                          AccStatus, AccSysGauges, [{NewK, V} | AccSysCounters]);
        {status, NewK} ->
            massage_stats(Indexer, Ets, Rest, AccGauges, AccCounters,
                          [{NewK, V} | AccStatus], AccSysGauges, AccSysCounters)
    end.

grab_stats(#state{indexer = Indexer}) ->
    case ns_cluster_membership:should_run_service(ns_config:latest(), Indexer:get_type(), node()) of
        true ->
            do_grab_stats(Indexer);
        false ->
            []
    end.

do_grab_stats(Indexer) ->
    case Indexer:grab_stats() of
        {ok, {[_|_] = Stats}} ->
            Stats;
        {ok, Other} ->
            ?log_error("Got invalid stats response for ~p:~n~p", [Indexer, Other]),
            [];
        {error, _} ->
            []
    end.

process_stats(TS, GrabbedStats, PrevCounters, PrevTS, #state{indexer = Indexer,
                                                             buckets = KnownBuckets,
                                                             default_stats = Defaults} = State) ->
    {Gauges0, Counters, Status, ServiceGauges, ServiceCounters} =
        massage_stats(Indexer, ets_name(Indexer), GrabbedStats),
    Gauges = Indexer:compute_gauges(Gauges0) ++ Gauges0,

    {Stats, SortedBucketCounters} =
        base_stats_collector:calculate_counters(TS, Gauges, Counters, PrevCounters, PrevTS),

    index_status_keeper:update(Indexer, Status),
    AllServiceGauges = Indexer:compute_service_gauges(ServiceGauges) ++ ServiceGauges,
    {ServiceStats1, SortedServiceCounters} =
        base_stats_collector:calculate_counters(TS, AllServiceGauges, ServiceCounters, PrevCounters, PrevTS),
    ServiceStats = [{Indexer:service_event_name(),
                     finalize_index_stats(ServiceStats1)}],
    Prefix = Indexer:prefix(),
    AggregatedStats =
        [{Prefix ++ binary_to_list(Bucket), Values} ||
            {Bucket, Values} <- aggregate_index_stats(Indexer, Stats, KnownBuckets, Defaults)] ++ ServiceStats,<commit_msg>re-factoring of index_stats_collector<commit_after>%% @author Couchbase, Inc <<EMAIL>>
%% @copyright 2015 Couchbase, Inc.
%%
%% 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.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
-module(index_stats_collector).

-include_lib("eunit/include/eunit.hrl").

-include("ns_common.hrl").

-include("ns_stats.hrl").

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

%% callbacks
-export([init/1, handle_info/2, grab_stats/1, process_stats/5]).

-record(state, {indexer :: atom(),
                default_stats,
                buckets}).

-record(stats_accumulators, {
          gauges = [],
          counters = [],
          sys_gauges = [],
          sys_counters = [],
          status = []
         }).

server_name(Indexer) ->
    list_to_atom(?MODULE_STRING "-" ++ atom_to_list(Indexer:get_type())).

ets_name(Indexer) ->
    list_to_atom(?MODULE_STRING "_names-" ++ atom_to_list(Indexer:get_type())).

start_link(Indexer) ->
    base_stats_collector:start_link({local, server_name(Indexer)}, ?MODULE, Indexer).

init(Indexer) ->
    ets:new(ets_name(Indexer), [protected, named_table]),

    Self = self(),
    ns_pubsub:subscribe_link(
      ns_config_events,
      fun ({buckets, Buckets}) ->
              BucketConfigs = proplists:get_value(configs, Buckets, []),
              BucketsList = ns_bucket:get_bucket_names_of_type(membase, couchstore, BucketConfigs) ++
                  ns_bucket:get_bucket_names_of_type(membase, ephemeral, BucketConfigs),
              Self ! {buckets, BucketsList};
          (_) ->
              ok
      end),

    Buckets = lists:map(fun list_to_binary/1,
                        ns_bucket:get_bucket_names_of_type(membase, couchstore) ++
                            ns_bucket:get_bucket_names_of_type(membase, ephemeral)),
    Defaults = [{Indexer:global_index_stat(atom_to_binary(Stat, latin1)), 0}
                || Stat <- Indexer:get_gauges() ++ Indexer:get_counters() ++ Indexer:get_computed()],


    {ok, #state{indexer = Indexer,
                buckets = Buckets,
                default_stats = finalize_index_stats(Defaults)}}.

find_type(_, []) ->
    not_found;
find_type(Name, [{Type, Metrics} | Rest]) ->
    MaybeMetric = [Name || M <- Metrics,
                           atom_to_binary(M, latin1) =:= Name],

    case MaybeMetric of
        [_] ->
            Type;
        _ ->
            find_type(Name, Rest)
    end.

do_recognize_name(_Indexer, <<"needs_restart">>) ->
    {#stats_accumulators.status, index_needs_restart};
do_recognize_name(_Indexer, <<"num_connections">>) ->
    {#stats_accumulators.status, index_num_connections};
do_recognize_name(Indexer, K) ->
    Type = find_type(K, [{#stats_accumulators.sys_gauges, Indexer:get_service_gauges()},
                         {#stats_accumulators.sys_counters, Indexer:get_service_counters()}]),
    case Type of
        not_found ->
            do_recognize_bucket_metric(Indexer, K);
        _ ->
            NewKey = list_to_binary(Indexer:service_stat_prefix() ++
                                        binary_to_list(K)),
            {Type, NewKey}
    end.

do_recognize_bucket_metric(Indexer, K) ->
    case binary:split(K, <<":">>, [global]) of
        [Bucket, Index, Metric] ->
            Type = find_type(Metric, [{#stats_accumulators.gauges, Indexer:get_gauges()},
                                      {#stats_accumulators.counters, Indexer:get_counters()}]),

            case Type of
                not_found ->
                    undefined;
                _ ->
                    {Type, {Bucket, Index, Metric}}
            end;
        _ ->
            undefined
    end.

recognize_name(Indexer, Ets, K) ->
    case ets:lookup(Ets, K) of
        [{K, Type, NewK}] ->
            {Type, NewK};
        [{K, undefined}] ->
            undefined;
        [] ->
            case do_recognize_name(Indexer, K) of
                undefined ->
                    ets:insert(Ets, {K, undefined}),
                    undefined;
                {Type, NewK} ->
                    ets:insert(Ets, {K, Type, NewK}),
                    {Type, NewK}
            end
    end.

massage_stats(Indexer, Ets, GrabbedStats) ->
    massage_stats(Indexer, Ets, GrabbedStats, #stats_accumulators{}).

massage_stats(_Indexer, _Ets, [], Acc) ->
    Acc;
massage_stats(Indexer, Ets, [{K, V} | Rest], Acc) ->
    case recognize_name(Indexer, Ets, K) of
        undefined ->
            massage_stats(Indexer, Ets, Rest, Acc);
        {Pos, NewK} ->
            massage_stats(
              Indexer, Ets, Rest, setelement(Pos, Acc, [{NewK, V} | element(Pos, Acc)]))
    end.

grab_stats(#state{indexer = Indexer}) ->
    case ns_cluster_membership:should_run_service(ns_config:latest(), Indexer:get_type(), node()) of
        true ->
            do_grab_stats(Indexer);
        false ->
            []
    end.

do_grab_stats(Indexer) ->
    case Indexer:grab_stats() of
        {ok, {[_|_] = Stats}} ->
            Stats;
        {ok, Other} ->
            ?log_error("Got invalid stats response for ~p:~n~p", [Indexer, Other]),
            [];
        {error, _} ->
            []
    end.

process_stats(TS, GrabbedStats, PrevCounters, PrevTS, #state{indexer = Indexer,
                                                             buckets = KnownBuckets,
                                                             default_stats = Defaults} = State) ->
    MassagedStats =
        massage_stats(Indexer, ets_name(Indexer), GrabbedStats),

    CalculateStats =
        fun (GaugesPos, CountersPos, ComputeGauges) ->
                Gauges0 = element(GaugesPos, MassagedStats),
                Gauges = Indexer:ComputeGauges(Gauges0) ++ Gauges0,
                Counters = element(CountersPos, MassagedStats),
                base_stats_collector:calculate_counters(TS, Gauges, Counters, PrevCounters, PrevTS)
        end,

    index_status_keeper:update(Indexer, MassagedStats#stats_accumulators.status),

    {Stats, SortedBucketCounters} =
        CalculateStats(#stats_accumulators.gauges, #stats_accumulators.counters, compute_gauges),
    {ServiceStats1, SortedServiceCounters} =
        CalculateStats(#stats_accumulators.sys_gauges, #stats_accumulators.sys_counters,
                       compute_service_gauges),

    ServiceStats = [{Indexer:service_event_name(),
                     finalize_index_stats(ServiceStats1)}],
    Prefix = Indexer:prefix(),
    AggregatedStats =
        [{Prefix ++ binary_to_list(Bucket), Values} ||
            {Bucket, Values} <- aggregate_index_stats(Indexer, Stats, KnownBuckets, Defaults)] ++
        ServiceStats,
{'fraction_non_alphanumeric': 0.12416199206457791, 'fraction_numerical': 0.0030099876864140103, 'mean_word_length': 2.330827067669173, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 102, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 13, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3925502', 'n_tokens_mistral': 4655, 'n_tokens_neox': 4486, 'n_words': 874}