<commit_before>%% @author Couchbase <<EMAIL>>
%% @copyright 2012 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.

%% This service provides a way to get an information about remote
%% clusters. The main consumer for it is xdc_rep_manager.
%%
%% There're several important functions exposed:
%%
%%  - fetch_remote_cluster/{1,2}
%%
%%    Basically returns a list of nodes that remote cluster have. Used by
%%    menelaus_web_remote_clusters. This function always goes to remote
%%    cluster no matter if there's a cached information or not.
%%
%%  - get_remote_bucket/{3,4}
%%
%%    Returns remote bucket information by remote cluster name and bucket
%%    name. The most important piece here is a vbucket map for that
%%    bucket. Vbucket map is returned as an erlang dict mapping vbuckets to
%%    replication chains. The chain contains ready URLs to remote vbucket
%%    databases so that xdc_rep_manager could just easily use it without any
%%    additional preparation. There's one exception though. Replication chains
%%    can contain 'undefined' atoms where there's corresponding master or
%%    replica for the vbucket.
%%
%%    `Through` parameter controls if client can accept stale
%%    information. When set to 'true' the most up to date information will be
%%    queried from the remote cluster. When set to 'false' cached information
%%    (if it's existent) will be returned.
%%
%%  - get_remote_bucket_by_ref/{2,3}
%%
%%    Same as previous functions. But remote cluster and bucket is identified
%%    by the 'target' reference stored in replication document.
%%
%%  - remote_bucket_reference/2
%%
%%    Construct remote bucket reference that can be used by
%%    get_remote_bucket_by_ref functions.
%%
%%  - get_memcached_vbucket_info_by_ref/{3, 4}
%%
%%    -> {ok, {Host :: binary(), MemcachedPort :: integer()}, #remote_bucket{}}
%%       | {error, _} | {error, _, _} % see get_remote_bucket_by_ref for errors
%%
%%    returns information about accessing given vbucket of given
%%    bucket via memcached protocol


%% The service maintains protected ets table that is used to cache obtained
%% information. It's structure is as follows.
%%
%%   clusters: an ordered list of UUIDs of cached remote clusters
%%
%%   <UUID>: a cached information for the remote cluster denoted by <UUID>
%%
%%   {buckets, <UUID>}: an ordered list of buckets cached for a cluster
%%                      denoted by <UUID>
%%
%%   {bucket, <UUID>, <RemoteBucketName>}: a cached information about remote
%%                                         bucket <RemoteBucketName> on the
%%                                         cluster denoted by <UUID>

-module(remote_clusters_info).

-behaviour(gen_server).

-type cache_through_mode() :: boolean() | true_using_new_connection.

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

-export([fetch_remote_cluster/1, fetch_remote_cluster/2,
         get_remote_bucket/3, get_remote_bucket/4,
         get_remote_bucket_by_ref/2, get_remote_bucket_by_ref/3,
         remote_bucket_reference/2, parse_remote_bucket_reference/1,
         invalidate_remote_bucket/2, invalidate_remote_bucket_by_ref/1,
         find_cluster_by_uuid/1,
         get_memcached_vbucket_info_by_ref/3,
         get_memcached_vbucket_info_by_ref/4]).

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

-import(menelaus_web_remote_clusters,
        [get_remote_clusters/0, cas_remote_clusters/2]).

-include("ns_common.hrl").
-include("couch_db.hrl").
-include("remote_clusters_info.hrl").
-include_lib("eunit/include/eunit.hrl").


-define(CACHE, ?MODULE).

-define(GC_INTERVAL,
        ns_config:read_key_fast(
          {node, node(), remote_clusters_info_gc_interval}, 60000)).
-define(FETCH_CLUSTER_TIMEOUT,
        ns_config:get_timeout_fast(remote_info_fetch_cluster, 15000)).
-define(GET_BUCKET_TIMEOUT,
        ns_config:get_timeout_fast(remote_info_get_bucket, 30000)).

-define(CAS_TRIES,
        ns_config:read_key_fast(
          {node, node(), remote_clusters_info_cas_tries}, 10)).
-define(CONFIG_UPDATE_INTERVAL,
        ns_config:read_key_fast(
          {node, node(), remote_clusters_info_config_update_interval}, 10000)).

-record(state, {cache_path :: string(),
                scheduled_config_updates :: set(),
                remote_bucket_requests :: dict(),
                remote_bucket_waiters :: dict(),
                remote_bucket_waiters_trefs :: dict()}).

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

-spec fetch_remote_cluster(list()) -> {ok, #remote_cluster{}} |
                                      {error, timeout} |
                                      {error, rest_error, Msg, Details} |
                                      {error, client_error, Msg} |
                                      {error, bad_value, Msg} |
                                      {error, {bad_value, Field}, Msg} |
                                      {error, not_capable, Msg} |
                                      {error, {missing_field, Field}, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
fetch_remote_cluster(Cluster) ->
    fetch_remote_cluster(Cluster, ?FETCH_CLUSTER_TIMEOUT).

-spec fetch_remote_cluster(list(), integer()) ->
                                  {ok, #remote_cluster{}} |
                                  {error, timeout} |
                                  {error, rest_error, Msg, Details} |
                                  {error, client_error, Msg} |
                                  {error, bad_value, Msg} |
                                  {error, {bad_value, Field}, Msg} |
                                  {error, not_capable, Msg} |
                                  {error, {missing_field, Field}, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
fetch_remote_cluster(Cluster, Timeout) ->
    gen_server:call(?MODULE, {fetch_remote_cluster, Cluster, Timeout}, infinity).

-spec get_remote_bucket_by_ref(binary(), boolean()) ->
                                      {ok, #remote_bucket{}} |
                                      {error, cluster_not_found, Msg} |
                                      {error, timeout} |
                                      {error, rest_error, Msg, Details} |
                                      {error, client_error, Msg} |
                                      {error, bad_value, Msg} |
                                      {error, {bad_value, Field}, Msg} |
                                      {error, {missing_field, Field}, Msg} |
                                      {error, all_nodes_failed, Msg} |
                                      {error, other_cluster, Msg} |
                                      {error, not_capable, Msg} |
                                      {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket_by_ref(Reference, Through) ->
    get_remote_bucket_by_ref(Reference, Through, ?GET_BUCKET_TIMEOUT).

-spec get_remote_bucket_by_ref(binary(), boolean(), integer()) ->
                                      {ok, #remote_bucket{}} |
                                      {error, cluster_not_found, Msg} |
                                      {error, timeout} |
                                      {error, rest_error, Msg, Details} |
                                      {error, client_error, Msg} |
                                      {error, bad_value, Msg} |
                                      {error, {bad_value, Field}, Msg} |
                                      {error, {missing_field, Field}, Msg} |
                                      {error, all_nodes_failed, Msg} |
                                      {error, other_cluster, Msg} |
                                      {error, not_capable, Msg} |
                                      {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket_by_ref(Reference, Through, Timeout) ->
    {ok, {ClusterUUID, BucketName}} = parse_remote_bucket_reference(Reference),
    Cluster = find_cluster_by_uuid(ClusterUUID),
    gen_server:call(?MODULE,
                    {get_remote_bucket, Cluster,
                                        BucketName, Through, Timeout}, infinity).

invalidate_remote_bucket_by_ref(Reference) ->
    {ok, {ClusterUUID, BucketName}} = parse_remote_bucket_reference(Reference),
    invalidate_remote_bucket(ClusterUUID, BucketName).

-spec get_remote_bucket(string(), bucket_name(), cache_through_mode()) ->
                               {ok, #remote_bucket{}} |
                               {error, cluster_not_found, Msg} |
                               {error, timeout} |
                               {error, rest_error, Msg, Details} |
                               {error, client_error, Msg} |
                               {error, bad_value, Msg} |
                               {error, {bad_value, Field}, Msg} |
                               {error, {missing_field, Field}, Msg} |
                               {error, all_nodes_failed, Msg} |
                               {error, other_cluster, Msg} |
                               {error, not_capable, Msg} |
                               {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket(ClusterName, Bucket, Through) ->
    get_remote_bucket(ClusterName, Bucket, Through, ?GET_BUCKET_TIMEOUT).

-spec get_remote_bucket(string(), bucket_name(), cache_through_mode(), integer()) ->
                               {ok, #remote_bucket{}} |
                               {error, cluster_not_found, Msg} |
                               {error, timeout} |
                               {error, rest_error, Msg, Details} |
                               {error, client_error, Msg} |
                               {error, bad_value, Msg} |
                               {error, {bad_value, Field}, Msg} |
                               {error, {missing_field, Field}, Msg} |
                               {error, all_nodes_failed, Msg} |
                               {error, other_cluster, Msg} |
                               {error, not_capable, Msg} |
                               {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket(ClusterName, Bucket, Through, Timeout) ->
    Cluster = find_cluster_by_name(ClusterName),
    gen_server:call(?MODULE,
                    {get_remote_bucket, Cluster, Bucket, Through, Timeout}, infinity).


invalidate_remote_bucket(ClusterName, Bucket) ->
    Cluster = find_cluster_by_name(ClusterName),
    case Cluster of
        {error, _, _} ->
            Cluster;
        _ ->
            gen_server:call(?MODULE,
                            {invalidate_remote_bucket, Cluster, Bucket}, infinity)
    end.<commit_msg>Propagate error from find_cluster_by_name correctly.<commit_after>%% @author Couchbase <<EMAIL>>
%% @copyright 2012 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.

%% This service provides a way to get an information about remote
%% clusters. The main consumer for it is xdc_rep_manager.
%%
%% There're several important functions exposed:
%%
%%  - fetch_remote_cluster/{1,2}
%%
%%    Basically returns a list of nodes that remote cluster have. Used by
%%    menelaus_web_remote_clusters. This function always goes to remote
%%    cluster no matter if there's a cached information or not.
%%
%%  - get_remote_bucket/{3,4}
%%
%%    Returns remote bucket information by remote cluster name and bucket
%%    name. The most important piece here is a vbucket map for that
%%    bucket. Vbucket map is returned as an erlang dict mapping vbuckets to
%%    replication chains. The chain contains ready URLs to remote vbucket
%%    databases so that xdc_rep_manager could just easily use it without any
%%    additional preparation. There's one exception though. Replication chains
%%    can contain 'undefined' atoms where there's corresponding master or
%%    replica for the vbucket.
%%
%%    `Through` parameter controls if client can accept stale
%%    information. When set to 'true' the most up to date information will be
%%    queried from the remote cluster. When set to 'false' cached information
%%    (if it's existent) will be returned.
%%
%%  - get_remote_bucket_by_ref/{2,3}
%%
%%    Same as previous functions. But remote cluster and bucket is identified
%%    by the 'target' reference stored in replication document.
%%
%%  - remote_bucket_reference/2
%%
%%    Construct remote bucket reference that can be used by
%%    get_remote_bucket_by_ref functions.
%%
%%  - get_memcached_vbucket_info_by_ref/{3, 4}
%%
%%    -> {ok, {Host :: binary(), MemcachedPort :: integer()}, #remote_bucket{}}
%%       | {error, _} | {error, _, _} % see get_remote_bucket_by_ref for errors
%%
%%    returns information about accessing given vbucket of given
%%    bucket via memcached protocol


%% The service maintains protected ets table that is used to cache obtained
%% information. It's structure is as follows.
%%
%%   clusters: an ordered list of UUIDs of cached remote clusters
%%
%%   <UUID>: a cached information for the remote cluster denoted by <UUID>
%%
%%   {buckets, <UUID>}: an ordered list of buckets cached for a cluster
%%                      denoted by <UUID>
%%
%%   {bucket, <UUID>, <RemoteBucketName>}: a cached information about remote
%%                                         bucket <RemoteBucketName> on the
%%                                         cluster denoted by <UUID>

-module(remote_clusters_info).

-behaviour(gen_server).

-type cache_through_mode() :: boolean() | true_using_new_connection.

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

-export([fetch_remote_cluster/1, fetch_remote_cluster/2,
         get_remote_bucket/3, get_remote_bucket/4,
         get_remote_bucket_by_ref/2, get_remote_bucket_by_ref/3,
         remote_bucket_reference/2, parse_remote_bucket_reference/1,
         invalidate_remote_bucket/2, invalidate_remote_bucket_by_ref/1,
         find_cluster_by_uuid/1,
         get_memcached_vbucket_info_by_ref/3,
         get_memcached_vbucket_info_by_ref/4]).

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

-import(menelaus_web_remote_clusters,
        [get_remote_clusters/0, cas_remote_clusters/2]).

-include("ns_common.hrl").
-include("couch_db.hrl").
-include("remote_clusters_info.hrl").
-include_lib("eunit/include/eunit.hrl").


-define(CACHE, ?MODULE).

-define(GC_INTERVAL,
        ns_config:read_key_fast(
          {node, node(), remote_clusters_info_gc_interval}, 60000)).
-define(FETCH_CLUSTER_TIMEOUT,
        ns_config:get_timeout_fast(remote_info_fetch_cluster, 15000)).
-define(GET_BUCKET_TIMEOUT,
        ns_config:get_timeout_fast(remote_info_get_bucket, 30000)).

-define(CAS_TRIES,
        ns_config:read_key_fast(
          {node, node(), remote_clusters_info_cas_tries}, 10)).
-define(CONFIG_UPDATE_INTERVAL,
        ns_config:read_key_fast(
          {node, node(), remote_clusters_info_config_update_interval}, 10000)).

-record(state, {cache_path :: string(),
                scheduled_config_updates :: set(),
                remote_bucket_requests :: dict(),
                remote_bucket_waiters :: dict(),
                remote_bucket_waiters_trefs :: dict()}).

start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

-spec fetch_remote_cluster(list()) -> {ok, #remote_cluster{}} |
                                      {error, timeout} |
                                      {error, rest_error, Msg, Details} |
                                      {error, client_error, Msg} |
                                      {error, bad_value, Msg} |
                                      {error, {bad_value, Field}, Msg} |
                                      {error, not_capable, Msg} |
                                      {error, {missing_field, Field}, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
fetch_remote_cluster(Cluster) ->
    fetch_remote_cluster(Cluster, ?FETCH_CLUSTER_TIMEOUT).

-spec fetch_remote_cluster(list(), integer()) ->
                                  {ok, #remote_cluster{}} |
                                  {error, timeout} |
                                  {error, rest_error, Msg, Details} |
                                  {error, client_error, Msg} |
                                  {error, bad_value, Msg} |
                                  {error, {bad_value, Field}, Msg} |
                                  {error, not_capable, Msg} |
                                  {error, {missing_field, Field}, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
fetch_remote_cluster(Cluster, Timeout) ->
    gen_server:call(?MODULE, {fetch_remote_cluster, Cluster, Timeout}, infinity).

-spec get_remote_bucket_by_ref(binary(), boolean()) ->
                                      {ok, #remote_bucket{}} |
                                      {error, cluster_not_found, Msg} |
                                      {error, timeout} |
                                      {error, rest_error, Msg, Details} |
                                      {error, client_error, Msg} |
                                      {error, bad_value, Msg} |
                                      {error, {bad_value, Field}, Msg} |
                                      {error, {missing_field, Field}, Msg} |
                                      {error, all_nodes_failed, Msg} |
                                      {error, other_cluster, Msg} |
                                      {error, not_capable, Msg} |
                                      {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket_by_ref(Reference, Through) ->
    get_remote_bucket_by_ref(Reference, Through, ?GET_BUCKET_TIMEOUT).

-spec get_remote_bucket_by_ref(binary(), boolean(), integer()) ->
                                      {ok, #remote_bucket{}} |
                                      {error, cluster_not_found, Msg} |
                                      {error, timeout} |
                                      {error, rest_error, Msg, Details} |
                                      {error, client_error, Msg} |
                                      {error, bad_value, Msg} |
                                      {error, {bad_value, Field}, Msg} |
                                      {error, {missing_field, Field}, Msg} |
                                      {error, all_nodes_failed, Msg} |
                                      {error, other_cluster, Msg} |
                                      {error, not_capable, Msg} |
                                      {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket_by_ref(Reference, Through, Timeout) ->
    {ok, {ClusterUUID, BucketName}} = parse_remote_bucket_reference(Reference),
    Cluster = find_cluster_by_uuid(ClusterUUID),
    gen_server:call(?MODULE,
                    {get_remote_bucket, Cluster,
                                        BucketName, Through, Timeout}, infinity).

invalidate_remote_bucket_by_ref(Reference) ->
    {ok, {ClusterUUID, BucketName}} = parse_remote_bucket_reference(Reference),
    invalidate_remote_bucket(ClusterUUID, BucketName).

-spec get_remote_bucket(string(), bucket_name(), cache_through_mode()) ->
                               {ok, #remote_bucket{}} |
                               {error, cluster_not_found, Msg} |
                               {error, timeout} |
                               {error, rest_error, Msg, Details} |
                               {error, client_error, Msg} |
                               {error, bad_value, Msg} |
                               {error, {bad_value, Field}, Msg} |
                               {error, {missing_field, Field}, Msg} |
                               {error, all_nodes_failed, Msg} |
                               {error, other_cluster, Msg} |
                               {error, not_capable, Msg} |
                               {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket(ClusterName, Bucket, Through) ->
    get_remote_bucket(ClusterName, Bucket, Through, ?GET_BUCKET_TIMEOUT).

-spec get_remote_bucket(string(), bucket_name(), cache_through_mode(), integer()) ->
                               {ok, #remote_bucket{}} |
                               {error, cluster_not_found, Msg} |
                               {error, timeout} |
                               {error, rest_error, Msg, Details} |
                               {error, client_error, Msg} |
                               {error, bad_value, Msg} |
                               {error, {bad_value, Field}, Msg} |
                               {error, {missing_field, Field}, Msg} |
                               {error, all_nodes_failed, Msg} |
                               {error, other_cluster, Msg} |
                               {error, not_capable, Msg} |
                               {error, not_present, Msg}
  when Details :: {error, term()} | {bad_status, integer(), string()},
       Msg :: binary(),
       Field :: binary().
get_remote_bucket(ClusterName, Bucket, Through, Timeout) ->
    case find_cluster_by_name(ClusterName) of
        {error, _, _} = Error ->
            Error;
        Cluster ->
            gen_server:call(?MODULE,
                            {get_remote_bucket, Cluster,
                             Bucket, Through, Timeout}, infinity)
    end.

invalidate_remote_bucket(ClusterName, Bucket) ->
    Cluster = find_cluster_by_name(ClusterName),
    case Cluster of
        {error, _, _} ->
            Cluster;
        _ ->
            gen_server:call(?MODULE,
                            {invalidate_remote_bucket, Cluster, Bucket}, infinity)
    end.
{'fraction_non_alphanumeric': 0.10773631736149089, 'fraction_numerical': 0.005132304067981995, 'mean_word_length': 1.8717081420633004, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 61, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 92, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '492211', 'n_tokens_mistral': 6559, 'n_tokens_neox': 6377, 'n_words': 1643}