1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
#ifndef CERT_TRANS_LOG_CLUSTER_STATE_CONTROLLER_H_
#define CERT_TRANS_LOG_CLUSTER_STATE_CONTROLLER_H_
#include <condition_variable>
#include <functional>
#include <map>
#include <memory>
#include <queue>
#include <string>
#include "fetcher/continuous_fetcher.h"
#include "log/etcd_consistent_store.h"
#include "log/logged_entry.h"
#include "proto/ct.pb.h"
#include "util/libevent_wrapper.h"
#include "util/masterelection.h"
#include "util/statusor.h"
namespace cert_trans {
class Database;
// A class which updates & maintains the states of the individual cluster
// member nodes, and uses this information to determine the overall serving
// state of the cluster.
//
// In particular, this class:
// - calculates the optimal STH for the cluster to serve at any given time.
// - determines whether this node is eligible to participare in the election,
// and leaves/joins the election as appropriate.
class ClusterStateController {
public:
ClusterStateController(util::Executor* executor,
const std::shared_ptr<libevent::Base>& base,
UrlFetcher* url_fetcher, Database* database,
ConsistentStore* store, MasterElection* election,
ContinuousFetcher* fetcher);
~ClusterStateController();
// Updates *this* node's ClusterNodeState to reflect the new STH available.
void NewTreeHead(const ct::SignedTreeHead& sth);
// Gets the current (if any) calculated serving STH for the cluster.
// If there is such an STH then return true and |sth| is populated, returns
// false otherwise.
//
// Note that this simply returns this node's interpretation of the optimum
// serving STH, the current master/contents of the servingSTH file may
// differ.
//
// Really only intended for testing.
util::StatusOr<ct::SignedTreeHead> GetCalculatedServingSTH() const;
void GetLocalNodeState(ct::ClusterNodeState* state) const;
// Publishes this node's listening address in its ClusterNodeState, so that
// other nodes can request entries from its database.
void SetNodeHostPort(const std::string& host, const uint16_t port);
void RefreshNodeState();
bool NodeIsStale() const;
// Returns a vector of the other nodes in the cluster which are able to serve
// the cluster's current ServingSTH. Does not include this node in the
// returned list regardless of its freshness.
std::vector<ct::ClusterNodeState> GetFreshNodes() const;
private:
class ClusterPeer : public Peer {
public:
ClusterPeer(const std::shared_ptr<libevent::Base>& base,
UrlFetcher* fetcher, const ct::ClusterNodeState& state);
int64_t TreeSize() const override;
void UpdateClusterNodeState(const ct::ClusterNodeState& new_state);
ct::ClusterNodeState state() const;
std::pair<std::string, int> GetHostPort() const;
private:
mutable std::mutex lock_;
ct::ClusterNodeState state_;
DISALLOW_COPY_AND_ASSIGN(ClusterPeer);
};
// Updates the representation of *this* node's state in the consistent store.
void PushLocalNodeState(const std::unique_lock<std::mutex>& lock);
// Entry point for the watcher callback.
// Called whenever a node changes its node state.
void OnClusterStateUpdated(
const std::vector<Update<ct::ClusterNodeState>>& updates);
// Entry point for the config watcher callback.
// Called whenever the ClusterConfig is changed.
void OnClusterConfigUpdated(const Update<ct::ClusterConfig>& update);
// Entry point for the config watcher callback.
// Called whenever the ClusterConfig is changed.
void OnServingSthUpdated(const Update<ct::SignedTreeHead>& update);
// Calculates the STH which should be served by the cluster, given the
// current state of the nodes.
// If this node is the cluster master then the calculated serving STH is
// pushed out to the consistent store.
void CalculateServingSTH(const std::unique_lock<std::mutex>& lock);
// Determines whether this node should be participating in the election based
// on the current node's state.
void DetermineElectionParticipation(
const std::unique_lock<std::mutex>& lock);
// Thread entry point for ServingSTH updater thread.
void ClusterServingSTHUpdater();
const std::shared_ptr<libevent::Base> base_;
UrlFetcher* const url_fetcher_; // Not owned by us
Database* const database_; // Not owned by us
ConsistentStore* const store_; // Not owned by us
MasterElection* const election_; // Not owned by us
ContinuousFetcher* const fetcher_; // Not owned by us
util::SyncTask watch_config_task_;
util::SyncTask watch_node_states_task_;
util::SyncTask watch_serving_sth_task_;
ct::ClusterConfig cluster_config_;
mutable std::mutex mutex_; // covers the members below:
ct::ClusterNodeState local_node_state_;
std::map<std::string, const std::shared_ptr<ClusterPeer>> all_peers_;
std::unique_ptr<ct::SignedTreeHead> calculated_serving_sth_;
std::unique_ptr<ct::SignedTreeHead> actual_serving_sth_;
bool exiting_;
bool update_required_;
std::condition_variable update_required_cv_;
std::thread cluster_serving_sth_update_thread_;
friend class ClusterStateControllerTest;
DISALLOW_COPY_AND_ASSIGN(ClusterStateController);
};
} // namespace cert_trans
#endif // CERT_TRANS_LOG_CLUSTER_STATE_CONTROLLER_H__
|
