<commit_before>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2016 <NAME>.  All Rights Reserved.
%%
%% This file is provided to you 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.
%%
%% -------------------------------------------------------------------

%% @doc Read configuration information from Marathon and auto-cluster
%%      Erlang nodes based on this.

-module(sprinter).
-author("<NAME> <<EMAIL>>").

-behaviour(gen_server).

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

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

-include("lasp.hrl").

-define(REFRESH_INTERVAL, 2000).
-define(REFRESH_MESSAGE,  refresh).

-define(NODES_INTERVAL, 20000).
-define(NODES_MESSAGE,  nodes).

-define(BUILD_GRAPH_INTERVAL, 120000).
-define(BUILD_GRAPH_MESSAGE,  build_graph).

-define(ARTIFACT_INTERVAL, 1000).
-define(ARTIFACT_MESSAGE,  artifact).

%% State record.
-record(state, {attempted_nodes, graph}).

%%%===================================================================
%%% API
%%%===================================================================

%% @doc Same as start_link([]).
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
    start_link([]).

%% @doc Start and link to calling process.
-spec start_link(list())-> {ok, pid()} | ignore | {error, term()}.
start_link(Opts) ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, Opts, []).

graph() ->
    gen_server:call(?MODULE, graph, infinity).

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

%% @private
-spec init([]) -> {ok, #state{}}.
init([]) ->
    %% Don't start the timer if we're not running in Mesos.
    case os:getenv("MESOS_TASK_ID") of
        false ->
            lager:info("Not running in Mesos; disabling Marathon service."),
            ok;
        _ ->
            %% Configure erlcloud.
            S3Host = "s3.amazonaws.com",
            AccessKeyId = os:getenv("AWS_ACCESS_KEY_ID"),
            SecretAccessKey = os:getenv("AWS_SECRET_ACCESS_KEY"),
            erlcloud_s3:configure(AccessKeyId, SecretAccessKey, S3Host),

            %% Create S3 bucket.
            try
                BucketName = bucket_name(),
                lager:info("Creating bucket: ~p", [BucketName]),
                ok = erlcloud_s3:create_bucket(BucketName)
            catch
                _:{aws_error, Error} ->
                    lager:info("Bucket creation failed: ~p", [Error]),
                    ok
            end,

            %% Stall messages; Plumtree has a race on startup, again.
            timer:send_after(?NODES_INTERVAL, ?NODES_MESSAGE),

            %% Only construct the graph and attempt to repair the graph
            %% from the designated server node.
            case partisan_config:get(tag, client) of
                server ->
                    timer:send_after(?BUILD_GRAPH_INTERVAL, ?BUILD_GRAPH_MESSAGE);
                client ->
                    ok
            end,

            %% All nodes should upload artifacts.
            timer:send_after(?ARTIFACT_INTERVAL, ?ARTIFACT_MESSAGE),

            %% All nodes should attempt to refresh the membership.
            timer:send_after(?REFRESH_INTERVAL, ?REFRESH_MESSAGE)
    end,
    {ok, #state{attempted_nodes=sets:new(), graph=digraph:new()}}.

%% @private
-spec handle_call(term(), {pid(), term()}, #state{}) ->
    {reply, term(), #state{}}.

%% @private
handle_call(graph, _From, #state{graph=Graph}=State) ->
    Vertices = digraph:vertices(Graph),
    Edges = lists:map(fun(Edge) ->
                      {_E, V1, V2, _Label} = digraph:edge(Graph, Edge),
                      {V1, V2}
              end, digraph:edges(Graph)),
    {reply, {ok, {Vertices, Edges}}, State};
handle_call(Msg, _From, State) ->
    _ = lager:warning("Unhandled messages: ~p", [Msg]),
    {reply, ok, State}.

%% @private
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast(Msg, State) ->
    _ = lager:warning("Unhandled messages: ~p", [Msg]),
    {noreply, State}.

%% @private
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
handle_info(?REFRESH_MESSAGE, #state{attempted_nodes=SeenNodes}=State) ->
    Tag = partisan_config:get(tag, client),
    PeerServiceManager = lasp_config:get(peer_service_manager,
                                         partisan_peer_service),

    %% Get list of nodes to connect to: this specialized logic isn't
    %% required when the node count is small, but is required with a
    %% larger node count to ensure the network stabilizes correctly
    %% because HyParView doesn't guarantee graph connectivity: it is
    %% only probabilistic.
    %%
    ToConnectNodes = case {Tag, PeerServiceManager} of
        {client, partisan_client_server_peer_service_manager} ->
            %% If we're a client, and we're in client/server mode, then
            %% always connect with the server.
            servers_from_marathon();
        {server, partisan_client_server_peer_service_manager} ->
            %% If we're a server, and we're in client/server mode, then
            %% always initiate connections with clients.
            clients_from_marathon();
        {client, partisan_hyparview_peer_service_manager} ->
            %% If we're the server, and we're in HyParView, clients will
            %% ask the server to join the overlay and force outbound
            %% conenctions to the clients.
            servers_from_marathon();
        {server, partisan_hyparview_peer_service_manager} ->
            %% If we're in HyParView, and we're a client, only ever
            %% do nothing -- force all connection to go through the
            %% server.
            sets:new()
    end,

    %% Attempt to connect nodes that are not connected.
    AttemptedNodes = maybe_connect(ToConnectNodes, SeenNodes),

    %% Add random jitter.
    Jitter = rand_compat:uniform(?REFRESH_INTERVAL),
    timer:send_after(?REFRESH_INTERVAL + Jitter, ?REFRESH_MESSAGE),

    {noreply, State#state{attempted_nodes=AttemptedNodes}};
handle_info(?NODES_MESSAGE, State) ->
    {ok, Nodes} = lasp_peer_service:members(),
    _ = lager:info("Currently connected nodes via peer service: ~p", [Nodes]),
    timer:send_after(?NODES_INTERVAL, ?NODES_MESSAGE),
    {noreply, State};
handle_info(?ARTIFACT_MESSAGE, State) ->
    % _ = lager:info("Uploading artifact!"),

    %% Get bucket name.
    BucketName = bucket_name(),

    %% Get current membership.
    {ok, Nodes} = lasp_peer_service:members(),

    %% Store membership.
    Node = prefix(atom_to_list(node())),
    Membership = term_to_binary(Nodes),
    try
        erlcloud_s3:put_object(BucketName, Node, Membership)
    catch
        _:{aws_error, Error} ->
            lager:info("Could not upload artifact: ~p", [Error])
    end,

    % _ = lager:info("Uploading complete!"),

    timer:send_after(?ARTIFACT_INTERVAL, ?ARTIFACT_MESSAGE),
    {noreply, State};
handle_info(?BUILD_GRAPH_MESSAGE, State) -><commit_msg>Remove useless code.<commit_after>%% -------------------------------------------------------------------
%%
%% Copyright (c) 2016 <NAME>.  All Rights Reserved.
%%
%% This file is provided to you 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.
%%
%% -------------------------------------------------------------------

%% @doc Read configuration information from Marathon and auto-cluster
%%      Erlang nodes based on this.

-module(sprinter).
-author("<NAME> <<EMAIL>>").

-behaviour(gen_server).

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

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

-include("lasp.hrl").

-define(REFRESH_INTERVAL, 2000).
-define(REFRESH_MESSAGE,  refresh).

-define(NODES_INTERVAL, 20000).
-define(NODES_MESSAGE,  nodes).

-define(BUILD_GRAPH_INTERVAL, 120000).
-define(BUILD_GRAPH_MESSAGE,  build_graph).

-define(ARTIFACT_INTERVAL, 1000).
-define(ARTIFACT_MESSAGE,  artifact).

%% State record.
-record(state, {attempted_nodes, graph}).

%%%===================================================================
%%% API
%%%===================================================================

%% @doc Same as start_link([]).
-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
    start_link([]).

%% @doc Start and link to calling process.
-spec start_link(list())-> {ok, pid()} | ignore | {error, term()}.
start_link(Opts) ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, Opts, []).

graph() ->
    gen_server:call(?MODULE, graph, infinity).

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

%% @private
-spec init([]) -> {ok, #state{}}.
init([]) ->
    %% Don't start the timer if we're not running in Mesos.
    case os:getenv("MESOS_TASK_ID") of
        false ->
            lager:info("Not running in Mesos; disabling Marathon service."),
            ok;
        _ ->
            %% Configure erlcloud.
            S3Host = "s3.amazonaws.com",
            AccessKeyId = os:getenv("AWS_ACCESS_KEY_ID"),
            SecretAccessKey = os:getenv("AWS_SECRET_ACCESS_KEY"),
            erlcloud_s3:configure(AccessKeyId, SecretAccessKey, S3Host),

            %% Create S3 bucket.
            try
                BucketName = bucket_name(),
                lager:info("Creating bucket: ~p", [BucketName]),
                ok = erlcloud_s3:create_bucket(BucketName)
            catch
                _:{aws_error, Error} ->
                    lager:info("Bucket creation failed: ~p", [Error]),
                    ok
            end,

            %% Stall messages; Plumtree has a race on startup, again.
            timer:send_after(?NODES_INTERVAL, ?NODES_MESSAGE),

            %% Only construct the graph and attempt to repair the graph
            %% from the designated server node.
            case partisan_config:get(tag, client) of
                server ->
                    timer:send_after(?BUILD_GRAPH_INTERVAL, ?BUILD_GRAPH_MESSAGE);
                client ->
                    ok
            end,

            %% All nodes should upload artifacts.
            timer:send_after(?ARTIFACT_INTERVAL, ?ARTIFACT_MESSAGE),

            %% All nodes should attempt to refresh the membership.
            timer:send_after(?REFRESH_INTERVAL, ?REFRESH_MESSAGE)
    end,
    {ok, #state{attempted_nodes=sets:new(), graph=digraph:new()}}.

%% @private
-spec handle_call(term(), {pid(), term()}, #state{}) ->
    {reply, term(), #state{}}.

%% @private
handle_call(graph, _From, #state{graph=Graph}=State) ->
    Vertices = digraph:vertices(Graph),
    Edges = lists:map(fun(Edge) ->
                      {_E, V1, V2, _Label} = digraph:edge(Graph, Edge),
                      {V1, V2}
              end, digraph:edges(Graph)),
    {reply, {ok, {Vertices, Edges}}, State};
handle_call(Msg, _From, State) ->
    _ = lager:warning("Unhandled messages: ~p", [Msg]),
    {reply, ok, State}.

%% @private
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast(Msg, State) ->
    _ = lager:warning("Unhandled messages: ~p", [Msg]),
    {noreply, State}.

%% @private
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
handle_info(?REFRESH_MESSAGE, #state{attempted_nodes=SeenNodes}=State) ->
    Tag = partisan_config:get(tag, client),
    PeerServiceManager = lasp_config:get(peer_service_manager,
                                         partisan_peer_service),

    %% Get list of nodes to connect to: this specialized logic isn't
    %% required when the node count is small, but is required with a
    %% larger node count to ensure the network stabilizes correctly
    %% because HyParView doesn't guarantee graph connectivity: it is
    %% only probabilistic.
    %%
    ToConnectNodes = case {Tag, PeerServiceManager} of
        {client, partisan_client_server_peer_service_manager} ->
            %% If we're a client, and we're in client/server mode, then
            %% always connect with the server.
            servers_from_marathon();
        {server, partisan_client_server_peer_service_manager} ->
            %% If we're a server, and we're in client/server mode, then
            %% always initiate connections with clients.
            clients_from_marathon();
        {client, partisan_hyparview_peer_service_manager} ->
            %% If we're the server, and we're in HyParView, clients will
            %% ask the server to join the overlay and force outbound
            %% conenctions to the clients.
            servers_from_marathon();
        {server, partisan_hyparview_peer_service_manager} ->
            %% If we're in HyParView, and we're a client, only ever
            %% do nothing -- force all connection to go through the
            %% server.
            sets:new()
    end,

    %% Attempt to connect nodes that are not connected.
    AttemptedNodes = maybe_connect(ToConnectNodes, SeenNodes),

    %% Add random jitter.
    Jitter = rand_compat:uniform(?REFRESH_INTERVAL),
    timer:send_after(?REFRESH_INTERVAL + Jitter, ?REFRESH_MESSAGE),

    {noreply, State#state{attempted_nodes=AttemptedNodes}};
handle_info(?NODES_MESSAGE, State) ->
    %% Do nothing right now.
    timer:send_after(?NODES_INTERVAL, ?NODES_MESSAGE),
    {noreply, State};
handle_info(?ARTIFACT_MESSAGE, State) ->
    % _ = lager:info("Uploading artifact!"),

    %% Get bucket name.
    BucketName = bucket_name(),

    %% Get current membership.
    {ok, Nodes} = lasp_peer_service:members(),

    %% Store membership.
    Node = prefix(atom_to_list(node())),
    Membership = term_to_binary(Nodes),
    try
        erlcloud_s3:put_object(BucketName, Node, Membership)
    catch
        _:{aws_error, Error} ->
            lager:info("Could not upload artifact: ~p", [Error])
    end,

    % _ = lager:info("Uploading complete!"),

    timer:send_after(?ARTIFACT_INTERVAL, ?ARTIFACT_MESSAGE),
    {noreply, State};
handle_info(?BUILD_GRAPH_MESSAGE, State) ->
{'fraction_non_alphanumeric': 0.17405475880052151, 'fraction_numerical': 0.006127770534550195, 'mean_word_length': 2.7362396492937164, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5767110', 'n_tokens_mistral': 4671, 'n_tokens_neox': 4342, 'n_words': 1163}