/Users/deen/code/yugabyte-db/src/yb/master/yql_peers_vtable.cc
Line | Count | Source (jump to first uncovered line) |
1 | | // Copyright (c) YugaByte, Inc. |
2 | | // |
3 | | // Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except |
4 | | // in compliance with the License. You may obtain a copy of the License at |
5 | | // |
6 | | // http://www.apache.org/licenses/LICENSE-2.0 |
7 | | // |
8 | | // Unless required by applicable law or agreed to in writing, software distributed under the License |
9 | | // is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
10 | | // or implied. See the License for the specific language governing permissions and limitations |
11 | | // under the License. |
12 | | // |
13 | | |
14 | | #include "yb/master/yql_peers_vtable.h" |
15 | | |
16 | | #include "yb/common/ql_protocol.pb.h" |
17 | | #include "yb/common/ql_type.h" |
18 | | #include "yb/common/schema.h" |
19 | | |
20 | | #include "yb/master/master.h" |
21 | | #include "yb/master/master_heartbeat.pb.h" |
22 | | #include "yb/master/ts_descriptor.h" |
23 | | |
24 | | #include "yb/rpc/messenger.h" |
25 | | |
26 | | #include "yb/util/net/dns_resolver.h" |
27 | | #include "yb/util/net/inetaddress.h" |
28 | | #include "yb/util/status_log.h" |
29 | | |
30 | | namespace yb { |
31 | | namespace master { |
32 | | |
33 | | using std::string; |
34 | | using std::shared_ptr; |
35 | | using std::unique_ptr; |
36 | | using std::map; |
37 | | |
38 | | namespace { |
39 | | |
40 | | const std::string kPeer = "peer"; |
41 | | const std::string kDataCenter = "data_center"; |
42 | | const std::string kHostId = "host_id"; |
43 | | const std::string kPreferredIp = "preferred_ip"; |
44 | | const std::string kRack = "rack"; |
45 | | const std::string kReleaseVersion = "release_version"; |
46 | | const std::string kRPCAddress = "rpc_address"; |
47 | | const std::string kSchemaVersion = "schema_version"; |
48 | | const std::string kTokens = "tokens"; |
49 | | |
50 | | } // namespace |
51 | | |
52 | | PeersVTable::PeersVTable(const TableName& table_name, |
53 | | const NamespaceName& namespace_name, |
54 | | Master* const master) |
55 | 2.00k | : YQLVirtualTable(table_name, namespace_name, master, CreateSchema()) { |
56 | 2.00k | } |
57 | | |
58 | | Result<std::shared_ptr<QLRowBlock>> PeersVTable::RetrieveData( |
59 | 46.6k | const QLReadRequestPB& request) const { |
60 | | // Retrieve all lives nodes known by the master. |
61 | | // TODO: Ideally we would like to populate this table with all valid nodes of the cluster, but |
62 | | // currently the master just has a list of all nodes it has heard from and which one of those |
63 | | // are dead. As a result, the master can't distinguish between nodes that are part of the |
64 | | // cluster and are dead vs nodes that have been removed from the cluster. Since, we might |
65 | | // change the cluster topology often, for now its safe to just have the live nodes here. |
66 | 46.6k | vector<shared_ptr<TSDescriptor> > descs; |
67 | 46.6k | GetSortedLiveDescriptors(&descs); |
68 | | |
69 | 46.6k | auto& resolver = master_->messenger()->resolver(); |
70 | | |
71 | | // Collect all unique ip addresses. |
72 | 46.6k | InetAddress remote_endpoint(VERIFY_RESULT(resolver.Resolve(request.remote_endpoint().host()))); |
73 | | |
74 | 46.6k | const auto& proxy_uuid = request.proxy_uuid(); |
75 | | |
76 | | // Populate the YQL rows. |
77 | 46.6k | auto vtable = std::make_shared<QLRowBlock>(schema()); |
78 | | |
79 | 46.6k | struct Entry { |
80 | 46.6k | size_t index; |
81 | 46.6k | TSInformationPB ts_info; |
82 | 46.6k | util::PublicPrivateIPFutures ts_ips; |
83 | 46.6k | }; |
84 | | |
85 | 46.6k | std::vector<Entry> entries; |
86 | 46.6k | entries.reserve(descs.size()); |
87 | | |
88 | 46.6k | size_t index = 0; |
89 | 141k | for (const auto& desc : descs) { |
90 | 141k | size_t current_index = index++; |
91 | | |
92 | | // This is thread safe since all operations are reads. |
93 | 141k | TSInformationPB ts_info = *desc->GetTSInformationPB(); |
94 | | |
95 | 141k | if (!proxy_uuid.empty()) { |
96 | 141k | if (desc->permanent_uuid() == proxy_uuid) { |
97 | 46.5k | continue; |
98 | 46.5k | } |
99 | 18.4E | } else { |
100 | | // In case of old proxy, fallback to old endpoint based mechanism. |
101 | 18.4E | if (util::RemoteEndpointMatchesTServer(ts_info, remote_endpoint)) { |
102 | 1 | continue; |
103 | 1 | } |
104 | 95.1k | } |
105 | | |
106 | 95.1k | entries.push_back({current_index, std::move(ts_info)}); |
107 | 95.1k | auto& entry = entries.back(); |
108 | 95.1k | entry.ts_ips = util::GetPublicPrivateIPFutures(entry.ts_info, &resolver); |
109 | 95.1k | } |
110 | | |
111 | 95.1k | for (const auto& entry : entries) { |
112 | | // The system.peers table has one entry for each of its peers, whereas there is no entry for |
113 | | // the node that the CQL client connects to. In this case, this node is the 'remote_endpoint' |
114 | | // in QLReadRequestPB since that is address of the CQL proxy which sent this request. As a |
115 | | // result, skip 'remote_endpoint' in the results. |
116 | 95.1k | auto private_ip = entry.ts_ips.private_ip_future.get(); |
117 | 95.1k | if (!private_ip.ok()) { |
118 | 0 | LOG(ERROR) << "Failed to get private ip from " << entry.ts_info.ShortDebugString() |
119 | 0 | << ": " << private_ip.status(); |
120 | 0 | continue; |
121 | 0 | } |
122 | | |
123 | 95.1k | auto public_ip = entry.ts_ips.public_ip_future.get(); |
124 | 95.1k | if (!public_ip.ok()) { |
125 | 0 | LOG(ERROR) << "Failed to get public ip from " << entry.ts_info.ShortDebugString() |
126 | 0 | << ": " << public_ip.status(); |
127 | 0 | continue; |
128 | 0 | } |
129 | | |
130 | | // Need to use only 1 rpc address per node since system.peers has only 1 entry for each host, |
131 | | // so pick the first one. |
132 | 95.1k | QLRow &row = vtable->Extend(); |
133 | 95.1k | RETURN_NOT_OK(SetColumnValue(kPeer, InetAddress(*public_ip), &row)); |
134 | 95.1k | RETURN_NOT_OK(SetColumnValue(kRPCAddress, InetAddress(*public_ip), &row)); |
135 | 95.1k | RETURN_NOT_OK(SetColumnValue(kPreferredIp, InetAddress(*private_ip), &row)); |
136 | | |
137 | | // Datacenter and rack. |
138 | 95.1k | CloudInfoPB cloud_info = entry.ts_info.registration().common().cloud_info(); |
139 | 95.1k | RETURN_NOT_OK(SetColumnValue(kDataCenter, cloud_info.placement_region(), &row)); |
140 | 95.1k | RETURN_NOT_OK(SetColumnValue(kRack, cloud_info.placement_zone(), &row)); |
141 | | |
142 | | // HostId. |
143 | 95.1k | Uuid host_id; |
144 | 95.1k | RETURN_NOT_OK(host_id.FromHexString(entry.ts_info.tserver_instance().permanent_uuid())); |
145 | 95.1k | RETURN_NOT_OK(SetColumnValue(kHostId, host_id, &row)); |
146 | 95.1k | RETURN_NOT_OK(SetColumnValue(yb::master::kSystemTablesReleaseVersionColumn, |
147 | 95.1k | yb::master::kSystemTablesReleaseVersion, &row)); |
148 | | |
149 | | // schema_version. |
150 | 95.1k | Uuid schema_version = VERIFY_RESULT(Uuid::FromString(master::kDefaultSchemaVersion)); |
151 | 95.1k | RETURN_NOT_OK(SetColumnValue(kSchemaVersion, schema_version, &row)); |
152 | | |
153 | | // Tokens. |
154 | 95.1k | RETURN_NOT_OK(SetColumnValue( |
155 | 95.1k | kTokens, util::GetTokensValue(entry.index, descs.size()), &row)); |
156 | 95.1k | } |
157 | | |
158 | 46.6k | return vtable; |
159 | 46.6k | } |
160 | | |
161 | 2.00k | Schema PeersVTable::CreateSchema() const { |
162 | 2.00k | SchemaBuilder builder; |
163 | 2.00k | CHECK_OK(builder.AddHashKeyColumn(kPeer, QLType::Create(DataType::INET))); |
164 | 2.00k | CHECK_OK(builder.AddColumn(kDataCenter, QLType::Create(DataType::STRING))); |
165 | 2.00k | CHECK_OK(builder.AddColumn(kHostId, QLType::Create(DataType::UUID))); |
166 | 2.00k | CHECK_OK(builder.AddColumn(kPreferredIp, QLType::Create(DataType::INET))); |
167 | 2.00k | CHECK_OK(builder.AddColumn(kRack, QLType::Create(DataType::STRING))); |
168 | 2.00k | CHECK_OK(builder.AddColumn(kReleaseVersion, QLType::Create(DataType::STRING))); |
169 | 2.00k | CHECK_OK(builder.AddColumn(kRPCAddress, QLType::Create(DataType::INET))); |
170 | 2.00k | CHECK_OK(builder.AddColumn(kSchemaVersion, QLType::Create(DataType::UUID))); |
171 | 2.00k | CHECK_OK(builder.AddColumn(kTokens, QLType::CreateTypeSet(DataType::STRING))); |
172 | 2.00k | return builder.Build(); |
173 | 2.00k | } |
174 | | |
175 | | } // namespace master |
176 | | } // namespace yb |