oceanbase

1
/**
2
 * Copyright (c) 2021 OceanBase
3
 * OceanBase CE is licensed under Mulan PubL v2.
4
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
5
 * You may obtain a copy of Mulan PubL v2 at:
6
 *          http://license.coscl.org.cn/MulanPubL-2.0
7
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
8
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
9
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10
 * See the Mulan PubL v2 for more details.
11
 */
12

13
#include <cstdio>
14
#include <gtest/gtest.h>
15
#include <signal.h>
16
#define private public
17
#include "lib/allocator/ob_qsync.h"
18
#include "logservice/ob_ls_adapter.h"
19
#include "env/ob_simple_log_cluster_env.h"
20
#undef private
21

22
const std::string TEST_NAME = "apply_func";
23
using namespace oceanbase::common;
24
using namespace oceanbase;
25

26
namespace oceanbase
27
{
28
using namespace logservice;
29
namespace unittest
30
{
31
class MockLSAdapter : public ObLSAdapter
32
{
33
public:
34
  MockLSAdapter() {
35
    ObLSAdapter();
36
    success_count_ = 0;
37
    failure_count_ = 0;
38
  }
39

40
  int wait_append_sync(const share::ObLSID &ls_id)
41
  {
42
    WaitQuiescent(ls_qs_);
43
    CLOG_LOG(INFO, "WaitQuiescent", K(ls_id));
44
    return OB_SUCCESS;
45
  }
46
  void critical_guard()
47
  {
48
    CriticalGuard(ls_qs_);
49
  }
50
  void inc_success_count()
51
  {
52
    ATOMIC_INC(&success_count_);
53
  }
54
  void inc_failure_count()
55
  {
56
    ATOMIC_INC(&failure_count_);
57
  }
58
  int replay(ObLogReplayTask *replay_task){
59
    UNUSED(replay_task);
60
    return OB_SUCCESS;
61
  }
62
  int64_t success_count_;
63
  int64_t failure_count_;
64
  common::ObQSync ls_qs_;
65
};
66

67
class TestObSimpleLogApplyFunc : public ObSimpleLogClusterTestEnv
68
{
69
public:
70
  TestObSimpleLogApplyFunc() :  ObSimpleLogClusterTestEnv()
71
  {}
72
};
73

74
class MockAppendCb : public AppendCb
75
{
76
public:
77
  int on_success()
78
  {
79
    ls_adapter_->inc_success_count();
80
    CLOG_LOG(INFO, "on_success", K(log_id_));
81
    return OB_SUCCESS;
82
  }
83

84
  int on_failure()
85
  {
86
    ls_adapter_->inc_failure_count();
87
    CLOG_LOG(INFO, "on_failure", K(log_id_));
88
    return OB_SUCCESS;
89
  }
90

91
  void init(const int64_t log_id,
92
            MockLSAdapter *ls_adapter)
93
  {
94
    log_id_ = log_id;
95
    ls_adapter_ = ls_adapter;
96
  }
97
  int64_t log_id_;
98
  MockLSAdapter *ls_adapter_;
99
};
100

101
int64_t ObSimpleLogClusterTestBase::member_cnt_ = 3;
102
int64_t ObSimpleLogClusterTestBase::node_cnt_ = 3;
103
std::string ObSimpleLogClusterTestBase::test_name_ = TEST_NAME;
104
bool ObSimpleLogClusterTestBase::need_add_arb_server_  = false;
105

106
TEST_F(TestObSimpleLogApplyFunc, apply)
107
{
108
  const int64_t task_count = 50;
109
  const int64_t truncate_count = 25;
110
  const int64_t id = ATOMIC_AAF(&palf_id_, 1);
111
  ObLSID ls_id(id);
112
  int64_t leader_idx = 0;
113
  LSN basic_lsn(0);
114
  PalfHandleImplGuard leader;
115
  MockAppendCb *cb_array[task_count];
116
  LSN unused_apply_end_lsn;
117
  bool is_apply_done = false;
118
  CLOG_LOG(INFO, "test apply begin", K(id));
119
  EXPECT_EQ(OB_SUCCESS, create_paxos_group(id, leader_idx, leader));
120
  MockLSAdapter ls_adapter;
121
  ls_adapter.init((ObLSService *)(0x1));
122
  ObLogApplyService ap_sv;
123
  for (int i = 0; i < task_count; i++)
124
  {
125
    cb_array[i] = new MockAppendCb();
126
    cb_array[i]->init(i + 1, &ls_adapter);
127
  }
128
  PalfEnv *palf_env;
129
  EXPECT_EQ(OB_SUCCESS, get_palf_env(leader_idx, palf_env));
130
  EXPECT_EQ(OB_SUCCESS, ap_sv.init(palf_env, &ls_adapter));
131
  EXPECT_EQ(OB_SUCCESS, ap_sv.start());
132
  EXPECT_EQ(OB_SUCCESS, ap_sv.add_ls(ls_id));
133
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 1));
134
  const int64_t idx_1 = (leader_idx + 1) % get_node_cnt();
135
  const int64_t idx_2 = (leader_idx + 2) % get_node_cnt();
136

137
  //旧主少数派写日志
138
  block_net(leader_idx, idx_1);
139
  block_net(leader_idx, idx_2);
140
  do {
141
    ObApplyStatus *apply_status = NULL;
142
    ObApplyStatusGuard guard;
143
    ap_sv.get_apply_status(ls_id, guard);
144
    apply_status = guard.get_apply_status();
145
    LSN lsn;
146
    share::SCN scn;
147
    int ret = OB_SUCCESS;
148
    for (int i = 0; i < truncate_count; i++)
149
    {
150
      {
151
        ls_adapter.critical_guard();
152
        CLOG_LOG(INFO, "submit log start", K(i));
153
        share::SCN ref_scn;
154
        ref_scn.convert_for_logservice(i);
155
        ret = submit_log(leader, ref_scn, lsn, scn);
156
        if (OB_SUCC(ret)) {
157
          EXPECT_EQ(true, lsn.is_valid());
158
          cb_array[i]->__set_lsn(lsn);
159
          cb_array[i]->__set_scn(scn);
160
          EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
161
        } else {
162
          break;
163
        }
164
      }
165
      CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
166
    }
167
  } while (0);
168
  share::SCN min_scn;
169
  EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, min_scn));
170
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
171

172
  //切主, truncate旧主日志,预期所有cb都调用on_failure
173
  sleep(15);
174
  while (!is_apply_done)
175
  {
176
    ap_sv.is_apply_done(ls_id, is_apply_done, unused_apply_end_lsn);
177
    usleep(100);
178
  }
179

180
  //切回旧主写日志,预期所有cb都调用on_success
181
  unblock_net(leader_idx, idx_1);
182
  unblock_net(leader_idx, idx_2);
183
  int64_t new_leader_idx = 0;
184
  PalfHandleImplGuard new_leader;
185
  EXPECT_EQ(OB_SUCCESS, get_leader(id, new_leader, new_leader_idx));
186
  EXPECT_NE(new_leader_idx, leader_idx);
187
  //等待membership同步
188
  sleep(2);
189
  leader.reset();
190
  CLOG_LOG(INFO, "new leader", K(new_leader_idx), K(leader_idx));
191
  EXPECT_EQ(OB_SUCCESS, switch_leader(id, leader_idx, leader));
192
  leader.reset();
193
  sleep(2);
194
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 3));
195
  EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, min_scn));
196
  EXPECT_EQ(OB_SUCCESS, get_leader(id, leader, new_leader_idx));
197
  CLOG_LOG(INFO, "check switch leader", K(new_leader_idx), K(leader_idx));
198
  EXPECT_EQ(new_leader_idx, leader_idx);
199
  share::SCN palf_end_scn;
200
  do {
201
    ObApplyStatus *apply_status = NULL;
202
    ObApplyStatusGuard guard;
203
    ap_sv.get_apply_status(ls_id, guard);
204
    apply_status = guard.get_apply_status();
205
    LSN lsn;
206
    share::SCN scn;
207
    for (int i = truncate_count; i < task_count; i++)
208
    {
209
      EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, scn));
210
      {
211
        int64_t ref_ts = i - truncate_count;
212
        share::SCN ref_scn;
213
        ref_scn.convert_for_logservice(ref_ts);
214
        ls_adapter.critical_guard();
215
        CLOG_LOG(INFO, "submit log start", K(i));
216
        EXPECT_EQ(OB_SUCCESS, submit_log(leader, ref_scn, lsn, scn));
217
        EXPECT_EQ(true, lsn.is_valid());
218
        cb_array[i]->__set_lsn(lsn);
219
        cb_array[i]->__set_scn(scn);
220
        palf_end_scn = scn;
221
        EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
222
      }
223
      CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
224
      EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, scn));
225
    }
226
  } while (0);
227
  CLOG_LOG(INFO, "truncate write finish", K(id));
228

229
  // remove
230
  is_apply_done = false;
231
  while (!is_apply_done)
232
  {
233
    ap_sv.is_apply_done(ls_id, is_apply_done, unused_apply_end_lsn);
234
    usleep(100);
235
  }
236
  share::SCN scn;
237
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
238
  EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, scn));
239
  EXPECT_EQ(scn.get_val_for_logservice(), palf_end_scn.get_val_for_logservice());
240
  EXPECT_EQ(truncate_count, ls_adapter.failure_count_);
241
  EXPECT_EQ(task_count- truncate_count, ls_adapter.success_count_);
242
  for (int i = 0; i < task_count; i++)
243
  {
244
    delete cb_array[i];
245
  }
246
  EXPECT_EQ(OB_SUCCESS, ap_sv.remove_ls(ls_id));
247
  ap_sv.stop();
248
  ap_sv.wait();
249
  ap_sv.destroy();
250
  CLOG_LOG(INFO, "test apply finish", K(id));
251
}
252

253
TEST_F(TestObSimpleLogApplyFunc, get_max_decided_scn)
254
{
255
  const int64_t task_count = 10;
256
  const int64_t id = ATOMIC_AAF(&palf_id_, 1);
257
  ObLSID ls_id(id);
258
  int64_t leader_idx = 0;
259
  LSN basic_lsn(0);
260
  PalfHandleImplGuard leader;
261
  MockAppendCb *cb_array[task_count];
262
  CLOG_LOG(INFO, "test get_max_decided_scn begin", K(id));
263
  EXPECT_EQ(OB_SUCCESS, create_paxos_group(id, leader_idx, leader));
264
  MockLSAdapter ls_adapter;
265
  ls_adapter.init((ObLSService *)(0x1));
266
  ObLogApplyService ap_sv;
267
  //submit log first
268
  for (int i = 0; i < task_count; i++)
269
  {
270
    cb_array[i] = new MockAppendCb();
271
    cb_array[i]->init(i + 1, &ls_adapter);
272
  }
273
  PalfEnv *palf_env;
274
  ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
275
  EXPECT_EQ(OB_SUCCESS, get_palf_env(leader_idx, palf_env));
276
  EXPECT_EQ(OB_SUCCESS, ap_sv.init(palf_env, &ls_adapter));
277
  EXPECT_EQ(OB_SUCCESS, ap_sv.start());
278
  EXPECT_EQ(OB_SUCCESS, ap_sv.add_ls(ls_id));
279
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 1));
280
  const int64_t idx_1 = (leader_idx + 1) % get_node_cnt();
281
  const int64_t idx_2 = (leader_idx + 2) % get_node_cnt();
282
  share::SCN ref_scn;
283
  int i = 0;
284
  LSN lsn;
285
  share::SCN scn;
286

287
  do {
288
    ObApplyStatus *apply_status = NULL;
289
    ObApplyStatusGuard guard;
290
    ap_sv.get_apply_status(ls_id, guard);
291
    apply_status = guard.get_apply_status();
292
    ASSERT_TRUE(NULL != apply_status);
293
    int ret = OB_SUCCESS;
294
    for (i = 0; i < task_count; i++)
295
    {
296
      {
297
        ls_adapter.critical_guard();
298
        CLOG_LOG(INFO, "submit log start", K(i));
299
        ref_scn.convert_for_logservice(i);
300
        ret = submit_log(leader, ref_scn, lsn, scn);
301
        if (OB_SUCC(ret)) {
302
          EXPECT_EQ(true, lsn.is_valid());
303
          cb_array[i]->__set_lsn(lsn);
304
          cb_array[i]->__set_scn(scn);
305
          EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
306
        } else {
307
          break;
308
        }
309
      }
310
      CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
311
    }
312
  } while (0);
313

314
  LSN max_lsn = leader.palf_handle_impl_->sw_.get_max_lsn();
315
  SCN max_scn = leader.palf_handle_impl_->sw_.get_max_scn();
316

317
  //wait apply done
318
  wait_lsn_until_slide(max_lsn, leader);
319
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
320
  LSN apply_end_lsn;
321
  bool is_apply_done = false;
322
  while (!is_apply_done)
323
  {
324
    ap_sv.is_apply_done(ls_id, is_apply_done, apply_end_lsn);
325
    usleep(100);
326
  }
327
  EXPECT_EQ(max_lsn, apply_end_lsn);
328

329
  //switch leader and submit  log
330
  PalfHandleImplGuard new_leader;
331
  EXPECT_EQ(OB_SUCCESS, switch_leader(id, idx_1, new_leader));
332

333
  PalfHandleImplGuard new_get_leader;
334
  int64_t new_leader_idx = 0;
335
  EXPECT_EQ(OB_SUCCESS, get_leader(id, new_get_leader, new_leader_idx));
336
  EXPECT_NE(new_leader_idx, leader_idx);
337

338
  int ret = OB_SUCCESS;
339
  ++i;
340
  ref_scn.convert_for_logservice(i);
341
  ret = submit_log(new_leader, ref_scn, lsn, scn);
342
  EXPECT_EQ(OB_SUCCESS, ret);
343
  ++i;
344
  ref_scn.convert_for_logservice(i);
345
  ret = submit_log(new_leader, ref_scn, lsn, scn);
346
  EXPECT_EQ(OB_SUCCESS, ret);
347

348
  LSN end_lsn;
349
  do {
350
    end_lsn = leader.palf_handle_impl_->get_end_lsn();
351
    if (end_lsn <= max_lsn) {
352
      CLOG_LOG(INFO, "wait end_lsn", K(end_lsn), K(max_lsn));
353
      sleep (1);
354
    }
355
  } while (end_lsn <= max_lsn);
356
  share::SCN max_decided_scn;
357
  EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, max_decided_scn));
358
  CLOG_LOG(INFO, "check here", K(max_scn), K(max_decided_scn), K(end_lsn), K(max_lsn), K(apply_end_lsn));
359
  EXPECT_EQ(max_decided_scn, max_scn);
360

361
  for (int i = 0; i < task_count; i++)
362
  {
363
    delete cb_array[i];
364
  }
365

366
  ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
367
  EXPECT_EQ(OB_SUCCESS, ap_sv.remove_ls(ls_id));
368
  ap_sv.stop();
369
  ap_sv.wait();
370
  ap_sv.destroy();
371
  CLOG_LOG(INFO, "test get_max_decided_scn finish", K(id));
372
}
373

374
TEST_F(TestObSimpleLogApplyFunc, get_max_decided_scn_no_more_log_after_switch_to_follower)
375
{
376
  const int64_t task_count = 10;
377
  const int64_t id = ATOMIC_AAF(&palf_id_, 1);
378
  ObLSID ls_id(id);
379
  int64_t leader_idx = 0;
380
  LSN basic_lsn(0);
381
  PalfHandleImplGuard leader;
382
  MockAppendCb *cb_array[task_count];
383
  CLOG_LOG(INFO, "test get_max_decided_scn_no_more_log_after_switch_to_follower begin", K(id));
384
  EXPECT_EQ(OB_SUCCESS, create_paxos_group(id, leader_idx, leader));
385
  MockLSAdapter ls_adapter;
386
  ls_adapter.init((ObLSService *)(0x1));
387
  ObLogApplyService ap_sv;
388
  //submit log first
389
  for (int i = 0; i < task_count; i++)
390
  {
391
    cb_array[i] = new MockAppendCb();
392
    cb_array[i]->init(i + 1, &ls_adapter);
393
  }
394
  PalfEnv *palf_env;
395
  ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
396
  EXPECT_EQ(OB_SUCCESS, get_palf_env(leader_idx, palf_env));
397
  EXPECT_EQ(OB_SUCCESS, ap_sv.init(palf_env, &ls_adapter));
398
  EXPECT_EQ(OB_SUCCESS, ap_sv.start());
399
  EXPECT_EQ(OB_SUCCESS, ap_sv.add_ls(ls_id));
400
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_leader(ls_id, 1));
401
  const int64_t idx_1 = (leader_idx + 1) % get_node_cnt();
402
  const int64_t idx_2 = (leader_idx + 2) % get_node_cnt();
403
  share::SCN ref_scn;
404
  int i = 0;
405
  LSN lsn;
406
  share::SCN scn;
407

408
  do {
409
    ObApplyStatus *apply_status = NULL;
410
    ObApplyStatusGuard guard;
411
    ap_sv.get_apply_status(ls_id, guard);
412
    apply_status = guard.get_apply_status();
413
    ASSERT_TRUE(NULL != apply_status);
414
    int ret = OB_SUCCESS;
415
    for (i = 0; i < task_count; i++)
416
    {
417
      {
418
        ls_adapter.critical_guard();
419
        CLOG_LOG(INFO, "submit log start", K(i));
420
        ref_scn.convert_for_logservice(i);
421
        ret = submit_log(leader, ref_scn, lsn, scn);
422
        if (OB_SUCC(ret)) {
423
          EXPECT_EQ(true, lsn.is_valid());
424
          cb_array[i]->__set_lsn(lsn);
425
          cb_array[i]->__set_scn(scn);
426
          EXPECT_EQ(OB_SUCCESS, apply_status->push_append_cb(cb_array[i]));
427
        } else {
428
          break;
429
        }
430
      }
431
      CLOG_LOG(INFO, "submit log finish", K(i), K(lsn), K(scn));
432
    }
433
  } while (0);
434

435
  LSN max_lsn = leader.palf_handle_impl_->sw_.get_max_lsn();
436
  SCN max_scn = leader.palf_handle_impl_->sw_.get_max_scn();
437

438
  //wait apply done
439
  wait_lsn_until_slide(max_lsn, leader);
440
  EXPECT_EQ(OB_SUCCESS, ap_sv.switch_to_follower(ls_id));
441
  LSN apply_end_lsn;
442
  bool is_apply_done = false;
443
  while (!is_apply_done)
444
  {
445
    ap_sv.is_apply_done(ls_id, is_apply_done, apply_end_lsn);
446
    usleep(100);
447
  }
448
  EXPECT_EQ(max_lsn, apply_end_lsn);
449

450
  share::SCN max_decided_scn;
451
  EXPECT_EQ(OB_SUCCESS, ap_sv.get_max_applied_scn(ls_id, max_decided_scn));
452
  CLOG_LOG(INFO, "check here", K(max_scn), K(max_decided_scn), K(max_lsn), K(apply_end_lsn));
453
  EXPECT_EQ(max_decided_scn, max_scn);
454

455
  ObTenantEnv::set_tenant(get_cluster()[leader_idx]->get_tenant_base());
456
  EXPECT_EQ(OB_SUCCESS, ap_sv.remove_ls(ls_id));
457
  ap_sv.stop();
458
  ap_sv.wait();
459
  ap_sv.destroy();
460
  CLOG_LOG(INFO, "test get_max_decided_scn_with_no_more_log finish", K(id));
461
}
462

463
} // unitest
464
} // oceanbase
465

466
int main(int argc, char **argv)
467
{
468
  RUN_SIMPLE_LOG_CLUSTER_TEST(TEST_NAME);
469
}
470