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
#include "ob_balance_info.h"
15

16
#include "lib/container/ob_array.h"
17
#include "lib/container/ob_array_iterator.h"
18
#include "lib/container/ob_se_array.h"
19
#include "lib/container/ob_se_array_iterator.h"
20
#include "lib/hash/ob_hashset.h"
21
#include "lib/profile/ob_trace_id.h"
22
#include "share/ob_global_stat_proxy.h"
23
#include "share/schema/ob_table_schema.h"
24
#include "share/schema/ob_schema_getter_guard.h"
25
#include "share/schema/ob_part_mgr_util.h"
26
#include "ob_unit_manager.h"
27
#include "ob_zone_manager.h"
28
#include "ob_root_utils.h"
29
#include "ob_root_service.h"
30
#include "ob_resource_weight_parser.h"
31
#include "observer/ob_server_struct.h"
32
#include "observer/omt/ob_tenant_config_mgr.h"
33
#include "lib/mysqlclient/ob_mysql_proxy.h"
34
#include "storage/ob_file_system_router.h"
35
#include "storage/tx/ob_i_ts_source.h"
36

37
using namespace oceanbase::common;
38
using namespace oceanbase::common::hash;
39
using namespace oceanbase::rootserver;
40
using namespace oceanbase::share;
41
using namespace oceanbase::share::schema;
42

43
////////////////
44
ObStatisticsCalculator::ObStatisticsCalculator()
45
    :sum_(0)
46
{}
47

48
void ObStatisticsCalculator::reset()
49
{
50
  values_.reset();
51
  sum_ = 0;
52
}
53

54
int ObStatisticsCalculator::add_value(double v)
55
{
56
  sum_ += v;
57
  return values_.push_back(v);
58
}
59

60
double ObStatisticsCalculator::get_avg()
61
{
62
  double avg = 0;
63
  if (values_.count() > 0) {
64
    avg = sum_ / static_cast<double>(values_.count());
65
  }
66
  return avg;
67
}
68

69
double ObStatisticsCalculator::get_standard_deviation()
70
{
71
  double sd = 0;
72
  int64_t n = values_.count();
73
  if (n > 0) {
74
    double avg = get_avg();
75
    FOREACH(it, values_) {
76
      double d = (*it) - avg;
77
      sd += d * d;
78
    }
79
    sd = sqrt(sd/static_cast<double>(n));
80
  }
81
  return sd;
82
}
83

84
int ZoneUnit::assign(const ZoneUnit &other)
85
{
86
  int ret = OB_SUCCESS;
87
  zone_ = other.zone_;
88
  active_unit_cnt_ = other.active_unit_cnt_;
89

90
  load_imbalance_ = other.load_imbalance_;
91
  cpu_imbalance_ = other.cpu_imbalance_;
92
  disk_imbalance_ = other.disk_imbalance_;
93
  iops_imbalance_ = other.iops_imbalance_;
94
  memory_imbalance_ = other.memory_imbalance_;
95
  load_avg_ = other.load_avg_;
96
  cpu_avg_ = other.cpu_avg_;
97
  disk_avg_ = other.disk_avg_;
98
  iops_avg_ = other.iops_avg_;
99
  memory_avg_ = other.memory_avg_;
100

101
  tg_pg_cnt_ = other.tg_pg_cnt_;
102
  if (OB_FAIL(copy_assign(all_unit_, other.all_unit_))) {
103
    LOG_WARN("failed to assign all_unit_", K(ret));
104
  }
105
  return ret;
106
}
107

108
bool ServerStat::can_migrate_in() const
109
{
110
  return !blocked_ && active_ && online_;;
111
}
112

113

114
int UnitStat::assign(const UnitStat &other)
115
{
116
  int ret = OB_SUCCESS;
117
  server_ = other.server_;
118
  in_pool_ = other.in_pool_;
119
  load_factor_ = other.load_factor_;
120
  capacity_ = other.capacity_;
121
  load_ = other.load_;
122
  tg_pg_cnt_ = other.tg_pg_cnt_;
123
  outside_replica_cnt_ = other.outside_replica_cnt_;
124
  inside_replica_cnt_ = other.inside_replica_cnt_;
125
  if (OB_FAIL(copy_assign(info_, other.info_))) {
126
    LOG_WARN("failed to assign info_", K(ret));
127
  }
128
  return ret;
129
}
130

131
UnitStat &UnitStat::operator=(const UnitStat &other)
132
{
133
  int ret = OB_SUCCESS;
134
  if (this != &other) {
135
    if (OB_FAIL(assign(other))) {
136
      LOG_WARN("fail to assign", K(ret));
137
    }
138
  }
139
  return *this;
140
}
141

142
double UnitStat::get_load_if(ObResourceWeight &weights,
143
                                      const LoadFactor &load_factor, const bool plus) const
144
{
145
  LoadFactor new_factor = load_factor_;
146
  if (plus) {
147
    new_factor += load_factor;
148
  } else {
149
    new_factor -= load_factor;
150
  }
151
  return weights.cpu_weight_ * (new_factor.get_cpu_usage()/get_cpu_limit())
152
         + weights.memory_weight_ * (new_factor.get_memory_usage()/get_memory_limit())
153
         + weights.disk_weight_ * (new_factor.get_disk_usage()/get_disk_limit())
154
         + weights.iops_weight_ * (new_factor.get_iops_usage()/get_iops_limit());
155
}
156

157
double UnitStat::calc_load(ObResourceWeight &weights,
158
                           const LoadFactor &load_factor) const
159
{
160
  return weights.cpu_weight_ * (load_factor.get_cpu_usage()/get_cpu_limit())
161
         + weights.memory_weight_ * (load_factor.get_memory_usage()/get_memory_limit())
162
         + weights.disk_weight_ * (load_factor.get_disk_usage()/get_disk_limit())
163
         + weights.iops_weight_ * (load_factor.get_iops_usage()/get_iops_limit());
164
}
165

166
int ServerReplicaCountMgr::init(const ObIArray<ObAddr> &servers)
167
{
168
  int ret = OB_SUCCESS;
169
  // allow init twice
170
  inited_ = false;
171
  server_replica_counts_.reuse();
172
  ServerReplicaCount server_replica_count;
173
  FOREACH_CNT_X(server, servers, OB_SUCCESS == ret) {
174
    if (!server->is_valid()) {
175
      ret = OB_INVALID_ARGUMENT;
176
      LOG_WARN("invalid server", "server", *server, K(ret));
177
    } else {
178
      server_replica_count.reset();
179
      server_replica_count.server_ = *server;
180
      server_replica_count.replica_count_ = 0;
181
      if (OB_FAIL(server_replica_counts_.push_back(server_replica_count))) {
182
        LOG_WARN("push_back failed", K(ret));
183
      }
184
    }
185
  }
186
  if (OB_SUCC(ret)) {
187
    inited_ = true;
188
  }
189
  return ret;
190
}
191

192
int ServerReplicaCountMgr::accumulate(
193
    const ObAddr &server, const int64_t cnt)
194
{
195
  int ret = OB_SUCCESS;
196
  if (!inited_) {
197
    ret = OB_NOT_INIT;
198
    LOG_WARN("not init", K(ret));
199
  } else if (!server.is_valid() || cnt < 0) {
200
    ret = OB_INVALID_ARGUMENT;
201
    LOG_WARN("invalid server", K(server), K(cnt), K(ret));
202
  } else {
203
    FOREACH_CNT(server_replica_count, server_replica_counts_) {
204
      if (server_replica_count->server_ == server) {
205
        server_replica_count->replica_count_ += cnt;
206
        break;
207
      }
208
    }
209
  }
210
  return ret;
211
}
212

213
int ServerReplicaCountMgr::get_replica_count(
214
    const ObAddr &server, int64_t &replica_count)
215
{
216
  int ret = OB_SUCCESS;
217
  if (!inited_) {
218
    ret = OB_NOT_INIT;
219
    LOG_WARN("not init", K(ret));
220
  } else if (!server.is_valid()) {
221
    ret = OB_INVALID_ARGUMENT;
222
    LOG_WARN("invalid server", K(server), K(ret));
223
  } else {
224
    bool found = false;
225
    FOREACH_CNT_X(server_replica_count, server_replica_counts_, !found) {
226
      if (server_replica_count->server_ == server) {
227
        replica_count = server_replica_count->replica_count_;
228
        found = true;
229
      }
230
    }
231
    if (!found) {
232
      ret = OB_ENTRY_NOT_EXIST;
233
    }
234
  }
235
  return ret;
236
}
237