YugabyteDB (2.13.1.0-b60, 21121d69985fbf76aa6958d8f04a9bfa936293b5)

Coverage Report

Created: 2022-03-22 16:43

/Users/deen/code/yugabyte-db/src/yb/integration-tests/ts_itest-base.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/integration-tests/ts_itest-base.h"
15
16
#include "yb/client/client.h"
17
#include "yb/client/schema.h"
18
#include "yb/client/table.h"
19
20
#include "yb/gutil/strings/split.h"
21
22
#include "yb/integration-tests/cluster_verifier.h"
23
#include "yb/integration-tests/external_mini_cluster.h"
24
#include "yb/integration-tests/external_mini_cluster_fs_inspector.h"
25
26
#include "yb/master/master_client.proxy.h"
27
#include "yb/master/master_cluster.proxy.h"
28
29
#include "yb/rpc/rpc_controller.h"
30
31
#include "yb/server/server_base.proxy.h"
32
33
#include "yb/util/opid.h"
34
#include "yb/util/status_log.h"
35
36
DEFINE_string(ts_flags, "", "Flags to pass through to tablet servers");
37
DEFINE_string(master_flags, "", "Flags to pass through to masters");
38
39
DEFINE_int32(num_tablet_servers, 3, "Number of tablet servers to start");
40
DEFINE_int32(num_replicas, 3, "Number of replicas per tablet server");
41
42
namespace yb {
43
namespace tserver {
44
45
TabletServerIntegrationTestBase::TabletServerIntegrationTestBase()
46
119
    : random_(SeedRandom()) {}
47
48
85
TabletServerIntegrationTestBase::~TabletServerIntegrationTestBase() = default;
49
50
void TabletServerIntegrationTestBase::AddExtraFlags(
51
230
    const std::string& flags_str, std::vector<std::string>* flags) {
52
230
  if (flags_str.empty()) {
53
229
    return;
54
229
  }
55
1
  std::vector<std::string> split_flags = strings::Split(flags_str, " ");
56
2
  for (const std::string& flag : split_flags) {
57
2
    flags->push_back(flag);
58
2
  }
59
1
}
60
61
void TabletServerIntegrationTestBase::CreateCluster(
62
    const std::string& data_root_path,
63
    const std::vector<std::string>& non_default_ts_flags,
64
115
    const std::vector<std::string>& non_default_master_flags) {
65
115
  LOG(INFO) << "Starting cluster with:";
66
115
  LOG(INFO) << "--------------";
67
115
  LOG(INFO) << FLAGS_num_tablet_servers << " tablet servers";
68
115
  LOG(INFO) << FLAGS_num_replicas << " replicas per TS";
69
115
  LOG(INFO) << "--------------";
70
71
115
  ExternalMiniClusterOptions opts;
72
115
  opts.num_tablet_servers = FLAGS_num_tablet_servers;
73
115
  opts.data_root = GetTestPath(data_root_path);
74
75
  // If the caller passed no flags use the default ones, where we stress consensus by setting
76
  // low timeouts and frequent cache misses.
77
115
  if (non_default_ts_flags.empty()) {
78
37
    opts.extra_tserver_flags.push_back("--log_cache_size_limit_mb=10");
79
37
    opts.extra_tserver_flags.push_back(strings::Substitute("--consensus_rpc_timeout_ms=$0",
80
37
                                                           FLAGS_consensus_rpc_timeout_ms));
81
78
  } else {
82
256
    for (const std::string& flag : non_default_ts_flags) {
83
256
      opts.extra_tserver_flags.push_back(flag);
84
256
    }
85
78
  }
86
  // Disable load balancer for master by default for these tests. You can override this through
87
  // setting flags in the passed in non_default_master_flags argument.
88
115
  opts.extra_master_flags.push_back("--enable_load_balancing=false");
89
115
  opts.extra_master_flags.push_back(yb::Format("--replication_factor=$0", FLAGS_num_replicas));
90
266
  for (const std::string& flag : non_default_master_flags) {
91
266
    opts.extra_master_flags.push_back(flag);
92
266
  }
93
94
115
  AddExtraFlags(FLAGS_ts_flags, &opts.extra_tserver_flags);
95
115
  AddExtraFlags(FLAGS_master_flags, &opts.extra_master_flags);
96
97
115
  UpdateMiniClusterOptions(&opts);
98
99
115
  cluster_.reset(new ExternalMiniCluster(opts));
100
115
  ASSERT_OK(cluster_->Start());
101
115
  inspect_.reset(new itest::ExternalMiniClusterFsInspector(cluster_.get()));
102
115
  CreateTSProxies();
103
115
}
104
105
// Creates TSServerDetails instance for each TabletServer and stores them
106
// in 'tablet_servers_'.
107
115
void TabletServerIntegrationTestBase::CreateTSProxies() {
108
115
  CHECK(tablet_servers_.empty());
109
115
  tablet_servers_ = CHECK_RESULT(itest::CreateTabletServerMap(cluster_.get()));
110
115
}
111
112
// Waits that all replicas for a all tablets of 'kTableName' table are online
113
// and creates the tablet_replicas_ map.
114
69
void TabletServerIntegrationTestBase::WaitForReplicasAndUpdateLocations() {
115
69
  int num_retries = 0;
116
117
69
  bool replicas_missing = true;
118
69
  do {
119
69
    std::unordered_multimap<std::string, itest::TServerDetails*> tablet_replicas;
120
69
    master::GetTableLocationsRequestPB req;
121
69
    master::GetTableLocationsResponsePB resp;
122
69
    rpc::RpcController controller;
123
69
    kTableName.SetIntoTableIdentifierPB(req.mutable_table());
124
69
    controller.set_timeout(MonoDelta::FromSeconds(1));
125
69
    CHECK_OK(cluster_->GetLeaderMasterProxy<master::MasterClientProxy>().GetTableLocations(
126
69
        req, &resp, &controller));
127
69
    CHECK_OK(controller.status());
128
69
    CHECK
(!resp.has_error()) << "Response had an error: " << resp.error().ShortDebugString()0
;
129
130
73
    for (const master::TabletLocationsPB& location : resp.tablet_locations()) {
131
210
      for (const master::TabletLocationsPB_ReplicaPB& replica : location.replicas()) {
132
210
        auto server = FindOrDie(tablet_servers_,
133
210
                                replica.ts_info().permanent_uuid()).get();
134
210
        tablet_replicas.emplace(location.tablet_id(), server);
135
210
      }
136
137
73
      if (tablet_replicas.count(location.tablet_id()) < implicit_cast<size_t>(FLAGS_num_replicas)) {
138
0
        LOG(WARNING)<< "Couldn't find the leader and/or replicas. Location: "
139
0
            << location.ShortDebugString();
140
0
        replicas_missing = true;
141
0
        SleepFor(MonoDelta::FromSeconds(1));
142
0
        num_retries++;
143
0
        break;
144
0
      }
145
146
73
      replicas_missing = false;
147
73
    }
148
69
    if (!replicas_missing) {
149
69
      tablet_replicas_ = tablet_replicas;
150
69
    }
151
69
  } while (replicas_missing && 
num_retries < kMaxRetries0
);
152
69
}
153
154
// Returns the last committed leader of the consensus configuration. Tries to get it from master
155
// but then actually tries to the get the committed consensus configuration to make sure.
156
itest::TServerDetails* TabletServerIntegrationTestBase::GetLeaderReplicaOrNull(
157
85
    const std::string& tablet_id) {
158
85
  std::string leader_uuid;
159
85
  Status master_found_leader_result = GetTabletLeaderUUIDFromMaster(tablet_id, &leader_uuid);
160
161
  // See if the master is up to date. I.e. if it does report a leader and if the
162
  // replica it reports as leader is still alive and (at least thinks) its still
163
  // the leader.
164
85
  itest::TServerDetails* leader;
165
85
  if (master_found_leader_result.ok()) {
166
85
    leader = GetReplicaWithUuidOrNull(tablet_id, leader_uuid);
167
85
    if (leader && GetReplicaStatusAndCheckIfLeader(leader, tablet_id,
168
85
                                                   MonoDelta::FromMilliseconds(100)).ok()) {
169
33
      return leader;
170
33
    }
171
85
  }
172
173
  // The replica we got from the master (if any) is either dead or not the leader.
174
  // Find the actual leader.
175
52
  pair<itest::TabletReplicaMap::iterator, itest::TabletReplicaMap::iterator> range =
176
52
      tablet_replicas_.equal_range(tablet_id);
177
52
  std::vector<itest::TServerDetails*> replicas_copy;
178
208
  for (; range.first != range.second; 
++range.first156
) {
179
156
    replicas_copy.push_back((*range.first).second);
180
156
  }
181
182
52
  std::random_shuffle(replicas_copy.begin(), replicas_copy.end());
183
156
  for (itest::TServerDetails* replica : replicas_copy) {
184
156
    if (GetReplicaStatusAndCheckIfLeader(replica, tablet_id,
185
156
                                         MonoDelta::FromMilliseconds(100)).ok()) {
186
0
      return replica;
187
0
    }
188
156
  }
189
52
  return NULL;
190
52
}
191
192
Status TabletServerIntegrationTestBase::GetLeaderReplicaWithRetries(
193
    const std::string& tablet_id,
194
    itest::TServerDetails** leader,
195
33
    int max_attempts) {
196
33
  int attempts = 0;
197
85
  while (attempts < max_attempts) {
198
85
    *leader = GetLeaderReplicaOrNull(tablet_id);
199
85
    if (*leader) {
200
33
      return Status::OK();
201
33
    }
202
52
    attempts++;
203
52
    SleepFor(MonoDelta::FromMilliseconds(100 * attempts));
204
52
  }
205
0
  return STATUS(NotFound, "Leader replica not found");
206
33
}
207
208
Status TabletServerIntegrationTestBase::GetTabletLeaderUUIDFromMaster(const std::string& tablet_id,
209
85
                                                                      std::string* leader_uuid) {
210
85
  master::GetTableLocationsRequestPB req;
211
85
  master::GetTableLocationsResponsePB resp;
212
85
  rpc::RpcController controller;
213
85
  controller.set_timeout(MonoDelta::FromMilliseconds(100));
214
85
  kTableName.SetIntoTableIdentifierPB(req.mutable_table());
215
216
85
  RETURN_NOT_OK(cluster_->GetMasterProxy<master::MasterClientProxy>().GetTableLocations(
217
85
      req, &resp, &controller));
218
85
  for (const master::TabletLocationsPB& loc : resp.tablet_locations()) {
219
85
    if (loc.tablet_id() == tablet_id) {
220
163
      for (const master::TabletLocationsPB::ReplicaPB& replica : loc.replicas()) {
221
163
        if (replica.role() == PeerRole::LEADER) {
222
85
          *leader_uuid = replica.ts_info().permanent_uuid();
223
85
          return Status::OK();
224
85
        }
225
163
      }
226
85
    }
227
85
  }
228
0
  return STATUS(NotFound, "Unable to find leader for tablet", tablet_id);
229
85
}
230
231
itest::TServerDetails* TabletServerIntegrationTestBase::GetReplicaWithUuidOrNull(
232
    const std::string& tablet_id,
233
89
    const std::string& uuid) {
234
89
  pair<itest::TabletReplicaMap::iterator, itest::TabletReplicaMap::iterator> range =
235
89
      tablet_replicas_.equal_range(tablet_id);
236
167
  for (; range.first != range.second; 
++range.first78
) {
237
166
    if ((*range.first).second->instance_id.permanent_uuid() == uuid) {
238
88
      return (*range.first).second;
239
88
    }
240
166
  }
241
1
  return NULL;
242
89
}
243
244
// Gets the locations of the consensus configuration and waits until all replicas
245
// are available for all tablets.
246
69
void TabletServerIntegrationTestBase::WaitForTSAndReplicas() {
247
69
  int num_retries = 0;
248
  // make sure the replicas are up and find the leader
249
69
  while (true) {
250
69
    if (num_retries >= kMaxRetries) {
251
0
      FAIL() << " Reached max. retries while looking up the config.";
252
0
    }
253
254
69
    Status status = cluster_->WaitForTabletServerCount(FLAGS_num_tablet_servers,
255
69
                                                       MonoDelta::FromSeconds(5));
256
69
    if (status.IsTimedOut()) {
257
0
      LOG(WARNING)<< "Timeout waiting for all replicas to be online, retrying...";
258
0
      num_retries++;
259
0
      continue;
260
0
    }
261
69
    break;
262
69
  }
263
69
  WaitForReplicasAndUpdateLocations();
264
69
}
265
266
// Removes a set of servers from the replicas_ list.
267
// Handy for controlling who to validate against after killing servers.
268
void TabletServerIntegrationTestBase::PruneFromReplicas(
269
0
    const std::unordered_set<std::string>& uuids) {
270
0
  auto iter = tablet_replicas_.begin();
271
0
  while (iter != tablet_replicas_.end()) {
272
0
    if (uuids.count((*iter).second->instance_id.permanent_uuid()) != 0) {
273
0
      iter = tablet_replicas_.erase(iter);
274
0
      continue;
275
0
    }
276
0
    ++iter;
277
0
  }
278
279
0
  for (const std::string& uuid : uuids) {
280
0
    tablet_servers_.erase(uuid);
281
0
  }
282
0
}
283
284
void TabletServerIntegrationTestBase::GetOnlyLiveFollowerReplicas(
285
    const std::string& tablet_id,
286
7
    std::vector<itest::TServerDetails*>* followers) {
287
7
  followers->clear();
288
7
  itest::TServerDetails* leader;
289
7
  CHECK_OK(GetLeaderReplicaWithRetries(tablet_id, &leader));
290
291
7
  std::vector<itest::TServerDetails*> replicas;
292
7
  pair<itest::TabletReplicaMap::iterator, itest::TabletReplicaMap::iterator> range =
293
7
      tablet_replicas_.equal_range(tablet_id);
294
28
  for (; range.first != range.second; 
++range.first21
) {
295
21
    replicas.push_back((*range.first).second);
296
21
  }
297
298
21
  for (itest::TServerDetails* replica : replicas) {
299
21
    if (leader != NULL &&
300
21
        replica->instance_id.permanent_uuid() == leader->instance_id.permanent_uuid()) {
301
7
      continue;
302
7
    }
303
14
    Status s = GetReplicaStatusAndCheckIfLeader(replica, tablet_id,
304
14
                                                MonoDelta::FromMilliseconds(100));
305
14
    if (s.IsIllegalState()) {
306
14
      followers->push_back(replica);
307
14
    }
308
14
  }
309
7
}
310
311
// Return the index within 'replicas' for the replica which is farthest ahead.
312
int64_t TabletServerIntegrationTestBase::GetFurthestAheadReplicaIdx(
313
    const std::string& tablet_id,
314
0
    const std::vector<itest::TServerDetails*>& replicas) {
315
0
  auto op_ids = CHECK_RESULT(GetLastOpIdForEachReplica(
316
0
      tablet_id, replicas, consensus::RECEIVED_OPID, MonoDelta::FromSeconds(10)));
317
318
0
  int64 max_index = 0;
319
0
  ssize_t max_replica_index = -1;
320
0
  for (size_t i = 0; i < op_ids.size(); i++) {
321
0
    if (op_ids[i].index > max_index) {
322
0
      max_index = op_ids[i].index;
323
0
      max_replica_index = i;
324
0
    }
325
0
  }
326
327
0
  CHECK_NE(max_replica_index, -1);
328
329
0
  return max_replica_index;
330
0
}
331
332
0
Status TabletServerIntegrationTestBase::ShutdownServerWithUUID(const std::string& uuid) {
333
0
  for (size_t i = 0; i < cluster_->num_tablet_servers(); i++) {
334
0
    ExternalTabletServer* ts = cluster_->tablet_server(i);
335
0
    if (ts->instance_id().permanent_uuid() == uuid) {
336
0
      ts->Shutdown();
337
0
      return Status::OK();
338
0
    }
339
0
  }
340
0
  return STATUS(NotFound, "Unable to find server with UUID", uuid);
341
0
}
342
343
0
Status TabletServerIntegrationTestBase::RestartServerWithUUID(const std::string& uuid) {
344
0
  for (size_t i = 0; i < cluster_->num_tablet_servers(); i++) {
345
0
    ExternalTabletServer* ts = cluster_->tablet_server(i);
346
0
    if (ts->instance_id().permanent_uuid() == uuid) {
347
0
      ts->Shutdown();
348
0
      RETURN_NOT_OK(CheckTabletServersAreAlive(tablet_servers_.size()-1));
349
0
      RETURN_NOT_OK(ts->Restart());
350
0
      RETURN_NOT_OK(CheckTabletServersAreAlive(tablet_servers_.size()));
351
0
      return Status::OK();
352
0
    }
353
0
  }
354
0
  return STATUS(NotFound, "Unable to find server with UUID", uuid);
355
0
}
356
357
// Since we're fault-tolerant we might mask when a tablet server is
358
// dead. This returns Status::IllegalState() if fewer than 'num_tablet_servers'
359
// are alive.
360
2
Status TabletServerIntegrationTestBase::CheckTabletServersAreAlive(size_t num_tablet_servers) {
361
2
  size_t live_count = 0;
362
2
  std::string error = strings::Substitute("Fewer than $0 TabletServers were alive. Dead TSs: ",
363
2
                                          num_tablet_servers);
364
2
  rpc::RpcController controller;
365
6
  for (const itest::TabletServerMap::value_type& entry : tablet_servers_) {
366
6
    controller.Reset();
367
6
    controller.set_timeout(MonoDelta::FromSeconds(10));
368
6
    server::PingRequestPB req;
369
6
    server::PingResponsePB resp;
370
6
    Status s = entry.second->generic_proxy->Ping(req, &resp, &controller);
371
6
    if (!s.ok()) {
372
0
      error += "\n" + entry.second->ToString() +  " (" + s.ToString() + ")";
373
0
      continue;
374
0
    }
375
6
    live_count++;
376
6
  }
377
2
  if (live_count < num_tablet_servers) {
378
0
    return STATUS(IllegalState, error);
379
0
  }
380
2
  return Status::OK();
381
2
}
382
383
86
void TabletServerIntegrationTestBase::TearDown() {
384
86
  client_.reset();
385
86
  if (cluster_) {
386
158
    for (const auto* daemon : cluster_->master_daemons()) {
387
316
      EXPECT_TRUE(daemon->IsShutdown() || daemon->IsProcessAlive()) << "Daemon: " << daemon->id();
388
158
    }
389
258
    
for (const auto* daemon : cluster_->tserver_daemons())82
{
390
516
      EXPECT_TRUE(daemon->IsShutdown() || daemon->IsProcessAlive()) << "Daemon: " << daemon->id();
391
258
    }
392
82
    cluster_->Shutdown();
393
82
  }
394
86
  tablet_servers_.clear();
395
86
  TabletServerTestBase::TearDown();
396
86
}
397
398
109
Result<std::unique_ptr<client::YBClient>> TabletServerIntegrationTestBase::CreateClient() {
399
  // Connect to the cluster.
400
109
  client::YBClientBuilder builder;
401
193
  for (const auto* master : cluster_->master_daemons()) {
402
193
    builder.add_master_server_addr(AsString(master->bound_rpc_addr()));
403
193
  }
404
109
  return builder.Build();
405
109
}
406
407
// Create a table with a single tablet.
408
67
void TabletServerIntegrationTestBase::CreateTable() {
409
67
  ASSERT_OK(client_->CreateNamespaceIfNotExists(kTableName.namespace_name(),
410
67
                                                kTableName.namespace_type()));
411
412
67
  ASSERT_OK(table_.Create(kTableName, 1, client::YBSchema(schema_), client_.get()));
413
67
}
414
415
// Starts an external cluster with a single tablet and a number of replicas equal
416
// to 'FLAGS_num_replicas'. The caller can pass 'ts_flags' to specify non-default
417
// flags to pass to the tablet servers.
418
void TabletServerIntegrationTestBase::BuildAndStart(
419
    const std::vector<std::string>& ts_flags,
420
69
    const std::vector<std::string>& master_flags) {
421
69
  CreateCluster("raft_consensus-itest-cluster", ts_flags, master_flags);
422
69
  client_ = 
ASSERT_RESULT67
(67
CreateClient());
423
67
  ASSERT_NO_FATALS(CreateTable());
424
67
  WaitForTSAndReplicas();
425
67
  CHECK_GT(tablet_replicas_.size(), 0);
426
67
  tablet_id_ = (*tablet_replicas_.begin()).first;
427
67
}
428
429
10
void TabletServerIntegrationTestBase::AssertAllReplicasAgree(size_t expected_result_count) {
430
10
  ClusterVerifier cluster_verifier(cluster_.get());
431
10
  ASSERT_NO_FATALS(cluster_verifier.CheckCluster());
432
6
  ASSERT_NO_FATALS(cluster_verifier.CheckRowCount(
433
6
      kTableName, ClusterVerifier::EXACTLY, expected_result_count));
434
6
}
435
436
0
client::YBTableType TabletServerIntegrationTestBase::table_type() {
437
0
  return client::YBTableType::YQL_TABLE_TYPE;
438
0
}
439
440
}  // namespace tserver
441
}  // namespace yb