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
#define USING_LOG_PREFIX RS_LB
14

15
#include "ob_root_balancer.h"
16

17
#include "lib/ob_define.h"
18
#include "lib/mysqlclient/ob_mysql_proxy.h"
19
#include "lib/stat/ob_diagnose_info.h"
20
#include "share/config/ob_server_config.h"
21
#include "share/schema/ob_multi_version_schema_service.h"
22
#include "share/ob_unit_replica_counter.h"
23
#include "rootserver/ob_rs_event_history_table_operator.h"
24
#include "rootserver/ob_root_service.h"
25
#include "observer/ob_server_struct.h"
26
#include "ob_unit_manager.h"
27
#include "ob_balance_info.h"
28

29
namespace oceanbase
30
{
31

32
using namespace common;
33
using namespace share;
34
using namespace share::schema;
35
using namespace obrpc;
36
namespace rootserver
37
{
38
using namespace balancer;
39

40
int64_t ObRootBalanceIdling::get_idle_interval_us()
41
{
42
  const int64_t min_idle_time = 10 * 1000000;
43
  return min_idle_time;
44
}
45

46
ObRootBalancer::ObRootBalancer()
47
  : ObRsReentrantThread(true), inited_(false), active_(0), idling_(stop_, *this),
48
    server_balancer_(),
49
    disaster_recovery_worker_(stop_),
50
    rootservice_util_checker_(stop_)
51
{
52
}
53

54
ObRootBalancer::~ObRootBalancer()
55
{
56
}
57

58
int ObRootBalancer::init(common::ObServerConfig &cfg,
59
    share::schema::ObMultiVersionSchemaService &schema_service,
60
    ObUnitManager &unit_mgr,
61
    ObServerManager &server_mgr,
62
    ObZoneManager &zone_mgr,
63
    ObSrvRpcProxy &rpc_proxy,
64
    ObAddr &self_addr,
65
    ObMySQLProxy &sql_proxy,
66
    ObDRTaskMgr &dr_task_mgr)
67
{
68
  int ret = OB_SUCCESS;
69
  static const int64_t root_balancer_thread_cnt = 1;
70
  if (inited_) {
71
    ret = OB_INIT_TWICE;
72
    LOG_WARN("root balancer already inited", K(ret));
73
  } else if (OB_UNLIKELY(nullptr == GCTX.lst_operator_)) {
74
    ret = OB_ERR_UNEXPECTED;
75
    LOG_WARN("lst_operator_ ptr is null", KR(ret));
76
  } else if (OB_FAIL(server_balancer_.init(schema_service, unit_mgr,
77
                                           zone_mgr, server_mgr, sql_proxy))) {
78
    LOG_WARN("init failed", K(ret));
79
  } else if (OB_FAIL(create(root_balancer_thread_cnt, "RootBalance"))) {
80
    LOG_WARN("create root balancer thread failed", K(ret), K(root_balancer_thread_cnt));
81
  } else if (OB_FAIL(disaster_recovery_worker_.init(
82
          self_addr, cfg, zone_mgr,
83
          dr_task_mgr, *GCTX.lst_operator_, schema_service, rpc_proxy, sql_proxy))) {
84
    LOG_WARN("fail to init disaster recovery worker", KR(ret));
85
  } else if (OB_FAIL(rootservice_util_checker_.init(
86
          unit_mgr, zone_mgr, *GCTX.rs_rpc_proxy_, self_addr, schema_service, sql_proxy, *GCTX.lst_operator_))) {
87
    LOG_WARN("fail to init rootservice util checker", KR(ret));
88
  } else {
89
    inited_ = true;
90
  }
91
  return ret;
92
}
93

94
int ObRootBalancer::idle() const
95
{
96
  int ret = OB_SUCCESS;
97
  if (!inited_) {
98
    ret = OB_NOT_INIT;
99
    LOG_WARN("not init", K(ret));
100
  } else if (OB_FAIL(idling_.idle())) {
101
    LOG_WARN("idle failed", K(ret));
102
  } else {
103
    LOG_INFO("root balance idle", "idle_time", idling_.get_idle_interval_us());
104
  }
105
  return ret;
106
}
107

108
void ObRootBalancer::wakeup()
109
{
110
  if (!inited_) {
111
    LOG_WARN_RET(OB_NOT_INIT, "not init");
112
  } else {
113
    idling_.wakeup();
114
  }
115
}
116

117
void ObRootBalancer::stop()
118
{
119
  int ret = OB_SUCCESS;
120
  if (!inited_) {
121
    ret = OB_NOT_INIT;
122
    LOG_WARN("not init", K(ret));
123
  } else {
124
    ObRsReentrantThread::stop();
125
    idling_.wakeup();
126
  }
127
}
128

129
int ObRootBalancer::check_stop() const
130
{
131
  int ret = OB_SUCCESS;
132
  if (!inited_) {
133
    ret = OB_NOT_INIT;
134
    LOG_WARN("not init", K(ret));
135
  } else {
136
    ret = stop_ ? OB_CANCELED : OB_SUCCESS;
137
  }
138
  return ret;
139
}
140

141
void ObRootBalancer::set_active()
142
{
143
  const int64_t old_val = ATOMIC_SET(&active_, 1);
144
  UNUSED(old_val);
145
  wakeup();
146
}
147

148
void ObRootBalancer::run3()
149
{
150
  LOG_INFO("root balance start");
151
  int ret = OB_SUCCESS;
152
  if (!inited_) {
153
    ret = OB_NOT_INIT;
154
    LOG_WARN("not init", K(ret));
155
  } else if (OB_FAIL(do_balance())) {
156
    LOG_WARN("all balance failed", K(ret));
157
  }
158
  LOG_INFO("root balance stop", K(ret), K(stop_));
159
}
160

161
int ObRootBalancer::do_balance()
162
{
163
  int ret = OB_SUCCESS;
164
  int64_t failed_times = 0;
165
  if (!inited_) {
166
    ret = OB_NOT_INIT;
167
    LOG_WARN("not inited", K(ret));
168
  }
169

170
  ObArray<uint64_t> all_tenant;
171
  while (OB_SUCC(ret) && !stop_) {
172
    update_last_run_timestamp();
173

174
    if (!ATOMIC_LOAD(&active_)) {
175
      if (OB_FAIL(idle())) {
176
        LOG_WARN("idle failed", K(ret));
177
        break;
178
      } else {
179
        continue;
180
      }
181
    }
182
    if (OB_FAIL(all_balance())) {
183
      failed_times++;
184
      LOG_WARN("do balance round failed", K(ret), K(failed_times));
185
      ret = OB_SUCCESS;
186
      EVENT_ADD(RS_BALANCER_FAIL_COUNT, 1);
187
    } else {
188
      failed_times = 0;
189
      EVENT_ADD(RS_BALANCER_SUCC_COUNT, 1);
190
    }
191

192
    // idle after success or failed 2 times
193
    if (0 == failed_times || failed_times >= 2) {
194
      if (OB_FAIL(idle())) {
195
        LOG_WARN("idle failed", K(ret));
196
      }
197
    }
198
  }
199
  return ret;
200
}
201

202
int ObRootBalancer::all_balance()
203
{
204
  int ret = OB_SUCCESS;
205
  int tmp_ret = OB_SUCCESS;
206
  if (!inited_) {
207
    ret = OB_NOT_INIT;
208
    LOG_WARN("not init", K(ret));
209
  } else {
210
    if (OB_SUCC(ret)) {
211
      if (OB_FAIL(disaster_recovery_worker_.try_disaster_recovery())) {
212
        LOG_WARN("fail to try disaster recovery", KR(ret));
213
      }
214
    }
215

216
    if (OB_SUCC(ret)) {
217
      if (OB_FAIL(rootservice_util_checker_.rootservice_util_check())) {
218
        LOG_WARN("fail to do rootservice util check", KR(ret));
219
      }
220
    }
221

222
    if (OB_SUCC(ret)) {
223
      if (OB_SUCCESS != (tmp_ret = server_balancer_.balance_servers())) {
224
        LOG_WARN("do unit balance failed", K(tmp_ret));
225
      }
226
    }
227
    if (OB_SUCC(ret)) {
228
      if (OB_FAIL(server_balancer_.tenant_group_balance())) {
229
        LOG_WARN("fail to do tenant group balance", KR(ret));
230
      }
231
    }
232
  }
233
  return ret;
234
}
235

236
int64_t ObRootBalancer::get_schedule_interval() const
237
{
238
  return idling_.get_idle_interval_us();
239
}
240

241

242
} // end namespace rootserver
243
} // end namespace oceanbase
244